@Test public void query() throws InterruptedException { List<Object> messages = new ArrayList<>(); CountDownLatch latch = new CountDownLatch(1); SseEmitter emitter = new SseEmitter(5000L) { @Override public void send(Object object) throws IOException { messages.add(messages); } @Override public void send(SseEventBuilder builder) throws IOException { messages.add(builder.build()); } }; emitter.onError(t -> { t.printStackTrace(); latch.countDown(); }); emitter.onCompletion(latch::countDown); emitter.onTimeout(latch::countDown); testSubject.query(Topology.DEFAULT_CONTEXT, "aggregateIdentifier contains \"demo\" | limit( 10)", QueryEventsRequestStreamObserver.TIME_WINDOW_CUSTOM, true, false, "token", emitter); latch.await(1, TimeUnit.SECONDS); assertEquals(13, messages.size()); }
@Async public void query(String context, String queryString, String timeWindow, boolean liveUpdates, boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter) { Sender oldSender = senderPerClient.remove(clientToken); if (oldSender != null) { logger.debug("Stopping sender for {}", clientToken); oldSender.stop(); } try { EventStore eventStore = eventStoreManager.getEventStore(context, forceReadFromLeader); if (eventStore == null) { sseEmitter.send(SseEmitter.event().name("error").data("No leader for context: " + context)); sseEmitter.complete(); return; } Sender sender = new Sender(sseEmitter, eventStore, context, queryString, timeWindow, liveUpdates); senderPerClient.put(clientToken, sender); sseEmitter.onTimeout(sender::stop); } catch (Exception e) { try { logger.warn("Error while processing query {} - {}", queryString, e.getMessage(), e); sseEmitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (IOException ignore) { } sseEmitter.complete(); } }
HttpStreamingQuery { @Async public void query(String context, String queryString, String timeWindow, boolean liveUpdates, boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter) { Sender oldSender = senderPerClient.remove(clientToken); if (oldSender != null) { logger.debug("Stopping sender for {}", clientToken); oldSender.stop(); } try { EventStore eventStore = eventStoreManager.getEventStore(context, forceReadFromLeader); if (eventStore == null) { sseEmitter.send(SseEmitter.event().name("error").data("No leader for context: " + context)); sseEmitter.complete(); return; } Sender sender = new Sender(sseEmitter, eventStore, context, queryString, timeWindow, liveUpdates); senderPerClient.put(clientToken, sender); sseEmitter.onTimeout(sender::stop); } catch (Exception e) { try { logger.warn("Error while processing query {} - {}", queryString, e.getMessage(), e); sseEmitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (IOException ignore) { } sseEmitter.complete(); } } }
HttpStreamingQuery { @Async public void query(String context, String queryString, String timeWindow, boolean liveUpdates, boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter) { Sender oldSender = senderPerClient.remove(clientToken); if (oldSender != null) { logger.debug("Stopping sender for {}", clientToken); oldSender.stop(); } try { EventStore eventStore = eventStoreManager.getEventStore(context, forceReadFromLeader); if (eventStore == null) { sseEmitter.send(SseEmitter.event().name("error").data("No leader for context: " + context)); sseEmitter.complete(); return; } Sender sender = new Sender(sseEmitter, eventStore, context, queryString, timeWindow, liveUpdates); senderPerClient.put(clientToken, sender); sseEmitter.onTimeout(sender::stop); } catch (Exception e) { try { logger.warn("Error while processing query {} - {}", queryString, e.getMessage(), e); sseEmitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (IOException ignore) { } sseEmitter.complete(); } } HttpStreamingQuery(EventStoreLocator eventStoreManager); }
HttpStreamingQuery { @Async public void query(String context, String queryString, String timeWindow, boolean liveUpdates, boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter) { Sender oldSender = senderPerClient.remove(clientToken); if (oldSender != null) { logger.debug("Stopping sender for {}", clientToken); oldSender.stop(); } try { EventStore eventStore = eventStoreManager.getEventStore(context, forceReadFromLeader); if (eventStore == null) { sseEmitter.send(SseEmitter.event().name("error").data("No leader for context: " + context)); sseEmitter.complete(); return; } Sender sender = new Sender(sseEmitter, eventStore, context, queryString, timeWindow, liveUpdates); senderPerClient.put(clientToken, sender); sseEmitter.onTimeout(sender::stop); } catch (Exception e) { try { logger.warn("Error while processing query {} - {}", queryString, e.getMessage(), e); sseEmitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (IOException ignore) { } sseEmitter.complete(); } } HttpStreamingQuery(EventStoreLocator eventStoreManager); @Async void query(String context, String queryString, String timeWindow, boolean liveUpdates,
                      boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter); }
HttpStreamingQuery { @Async public void query(String context, String queryString, String timeWindow, boolean liveUpdates, boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter) { Sender oldSender = senderPerClient.remove(clientToken); if (oldSender != null) { logger.debug("Stopping sender for {}", clientToken); oldSender.stop(); } try { EventStore eventStore = eventStoreManager.getEventStore(context, forceReadFromLeader); if (eventStore == null) { sseEmitter.send(SseEmitter.event().name("error").data("No leader for context: " + context)); sseEmitter.complete(); return; } Sender sender = new Sender(sseEmitter, eventStore, context, queryString, timeWindow, liveUpdates); senderPerClient.put(clientToken, sender); sseEmitter.onTimeout(sender::stop); } catch (Exception e) { try { logger.warn("Error while processing query {} - {}", queryString, e.getMessage(), e); sseEmitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (IOException ignore) { } sseEmitter.complete(); } } HttpStreamingQuery(EventStoreLocator eventStoreManager); @Async void query(String context, String queryString, String timeWindow, boolean liveUpdates,
                      boolean forceReadFromLeader, String clientToken, SseEmitter sseEmitter);  }