@Test void shouldReturnEmptyEventForStream() throws Exception { when(streamManager.hasSubscriptions(anyString())).thenReturn(false); Event event = new Event(); event.set("k1", "v1"); Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event); assertNull(actual); Subscription subscription = mock(MQLSubscription.class); when(subscription.matches(any(Event.class))).thenReturn(false); Set<Subscription> subscriptions = new ConcurrentSkipListSet<>(); subscriptions.add(subscription); when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions); actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event); assertNull(actual); }
public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; }
EventProcessor { public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; } }
EventProcessor { public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; } EventProcessor(MrePublishConfiguration config, StreamManager streamManager, Tee tee); }
EventProcessor { public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; } EventProcessor(MrePublishConfiguration config, StreamManager streamManager, Tee tee); Event process(String stream, Event event); }
EventProcessor { public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; } EventProcessor(MrePublishConfiguration config, StreamManager streamManager, Tee tee); Event process(String stream, Event event);  }