@Test void shouldDrainAndNoopForNonexistentSubscribers() { when(streamManager.hasSubscriptions(anyString())).thenReturn(false); drainer.run(); verify(transmitter, times(0)) .send(any(Event.class), any()); }
@Override public void run() { try { MDC.put(LOGGING_CONTEXT_KEY, LOGGING_CONTEXT_VALUE); final long startTime = clock.millis(); Set<String> streams = streamManager.getRegisteredStreams(); for (String stream : streams) { final List<Event> streamEventList = new ArrayList<>(); try { final Optional<BlockingQueue<Event>> streamQueueO = streamManager.getQueueForStream(stream); if (streamQueueO.isPresent()) { BlockingQueue<Event> streamEventQ = streamQueueO.get(); streamEventQ.drainTo(streamEventList); int queueDepth = streamEventList.size(); if (LOG.isTraceEnabled()) { LOG.trace("Queue drained size: {} for stream {}", queueDepth, stream); } streamManager.getStreamMetrics(stream) .ifPresent(m -> { m.getMantisEventsQueuedGauge().set((double) queueDepth); if (queueDepth > 0) { m.updateLastEventOnStreamTimestamp(); } }); streamEventList.stream() .map(e -> process(stream, e)) .filter(Objects::nonNull) .forEach(e -> eventTransmitter.send(e, stream)); streamEventList.clear(); } } catch (Exception e) { LOG.warn("Exception processing events for stream {}", stream, e); } } final long processingTime = clock.millis() - startTime; mantisEventDrainTimer.record(processingTime, TimeUnit.MILLISECONDS); } finally { MDC.remove(LOGGING_CONTEXT_KEY); } }
EventDrainer implements Runnable { @Override public void run() { try { MDC.put(LOGGING_CONTEXT_KEY, LOGGING_CONTEXT_VALUE); final long startTime = clock.millis(); Set<String> streams = streamManager.getRegisteredStreams(); for (String stream : streams) { final List<Event> streamEventList = new ArrayList<>(); try { final Optional<BlockingQueue<Event>> streamQueueO = streamManager.getQueueForStream(stream); if (streamQueueO.isPresent()) { BlockingQueue<Event> streamEventQ = streamQueueO.get(); streamEventQ.drainTo(streamEventList); int queueDepth = streamEventList.size(); if (LOG.isTraceEnabled()) { LOG.trace("Queue drained size: {} for stream {}", queueDepth, stream); } streamManager.getStreamMetrics(stream) .ifPresent(m -> { m.getMantisEventsQueuedGauge().set((double) queueDepth); if (queueDepth > 0) { m.updateLastEventOnStreamTimestamp(); } }); streamEventList.stream() .map(e -> process(stream, e)) .filter(Objects::nonNull) .forEach(e -> eventTransmitter.send(e, stream)); streamEventList.clear(); } } catch (Exception e) { LOG.warn("Exception processing events for stream {}", stream, e); } } final long processingTime = clock.millis() - startTime; mantisEventDrainTimer.record(processingTime, TimeUnit.MILLISECONDS); } finally { MDC.remove(LOGGING_CONTEXT_KEY); } } }
EventDrainer implements Runnable { @Override public void run() { try { MDC.put(LOGGING_CONTEXT_KEY, LOGGING_CONTEXT_VALUE); final long startTime = clock.millis(); Set<String> streams = streamManager.getRegisteredStreams(); for (String stream : streams) { final List<Event> streamEventList = new ArrayList<>(); try { final Optional<BlockingQueue<Event>> streamQueueO = streamManager.getQueueForStream(stream); if (streamQueueO.isPresent()) { BlockingQueue<Event> streamEventQ = streamQueueO.get(); streamEventQ.drainTo(streamEventList); int queueDepth = streamEventList.size(); if (LOG.isTraceEnabled()) { LOG.trace("Queue drained size: {} for stream {}", queueDepth, stream); } streamManager.getStreamMetrics(stream) .ifPresent(m -> { m.getMantisEventsQueuedGauge().set((double) queueDepth); if (queueDepth > 0) { m.updateLastEventOnStreamTimestamp(); } }); streamEventList.stream() .map(e -> process(stream, e)) .filter(Objects::nonNull) .forEach(e -> eventTransmitter.send(e, stream)); streamEventList.clear(); } } catch (Exception e) { LOG.warn("Exception processing events for stream {}", stream, e); } } final long processingTime = clock.millis() - startTime; mantisEventDrainTimer.record(processingTime, TimeUnit.MILLISECONDS); } finally { MDC.remove(LOGGING_CONTEXT_KEY); } } EventDrainer(MrePublishConfiguration config,
                 StreamManager streamManager,
                 Registry registry,
                 EventProcessor eventProcessor,
                 EventTransmitter eventTransmitter,
                 Clock clock); }
EventDrainer implements Runnable { @Override public void run() { try { MDC.put(LOGGING_CONTEXT_KEY, LOGGING_CONTEXT_VALUE); final long startTime = clock.millis(); Set<String> streams = streamManager.getRegisteredStreams(); for (String stream : streams) { final List<Event> streamEventList = new ArrayList<>(); try { final Optional<BlockingQueue<Event>> streamQueueO = streamManager.getQueueForStream(stream); if (streamQueueO.isPresent()) { BlockingQueue<Event> streamEventQ = streamQueueO.get(); streamEventQ.drainTo(streamEventList); int queueDepth = streamEventList.size(); if (LOG.isTraceEnabled()) { LOG.trace("Queue drained size: {} for stream {}", queueDepth, stream); } streamManager.getStreamMetrics(stream) .ifPresent(m -> { m.getMantisEventsQueuedGauge().set((double) queueDepth); if (queueDepth > 0) { m.updateLastEventOnStreamTimestamp(); } }); streamEventList.stream() .map(e -> process(stream, e)) .filter(Objects::nonNull) .forEach(e -> eventTransmitter.send(e, stream)); streamEventList.clear(); } } catch (Exception e) { LOG.warn("Exception processing events for stream {}", stream, e); } } final long processingTime = clock.millis() - startTime; mantisEventDrainTimer.record(processingTime, TimeUnit.MILLISECONDS); } finally { MDC.remove(LOGGING_CONTEXT_KEY); } } EventDrainer(MrePublishConfiguration config,
                 StreamManager streamManager,
                 Registry registry,
                 EventProcessor eventProcessor,
                 EventTransmitter eventTransmitter,
                 Clock clock); @Override void run(); }
EventDrainer implements Runnable { @Override public void run() { try { MDC.put(LOGGING_CONTEXT_KEY, LOGGING_CONTEXT_VALUE); final long startTime = clock.millis(); Set<String> streams = streamManager.getRegisteredStreams(); for (String stream : streams) { final List<Event> streamEventList = new ArrayList<>(); try { final Optional<BlockingQueue<Event>> streamQueueO = streamManager.getQueueForStream(stream); if (streamQueueO.isPresent()) { BlockingQueue<Event> streamEventQ = streamQueueO.get(); streamEventQ.drainTo(streamEventList); int queueDepth = streamEventList.size(); if (LOG.isTraceEnabled()) { LOG.trace("Queue drained size: {} for stream {}", queueDepth, stream); } streamManager.getStreamMetrics(stream) .ifPresent(m -> { m.getMantisEventsQueuedGauge().set((double) queueDepth); if (queueDepth > 0) { m.updateLastEventOnStreamTimestamp(); } }); streamEventList.stream() .map(e -> process(stream, e)) .filter(Objects::nonNull) .forEach(e -> eventTransmitter.send(e, stream)); streamEventList.clear(); } } catch (Exception e) { LOG.warn("Exception processing events for stream {}", stream, e); } } final long processingTime = clock.millis() - startTime; mantisEventDrainTimer.record(processingTime, TimeUnit.MILLISECONDS); } finally { MDC.remove(LOGGING_CONTEXT_KEY); } } EventDrainer(MrePublishConfiguration config,
                 StreamManager streamManager,
                 Registry registry,
                 EventProcessor eventProcessor,
                 EventTransmitter eventTransmitter,
                 Clock clock); @Override void run(); static final String LOGGING_CONTEXT_KEY; static final String LOGGING_CONTEXT_VALUE; }