OperationProcessor extends MonitoredProcess { public Stream<InternalEvent> perform(Stream<InternalEvent> input) { Stream<InternalEvent> output = null; if (this.op instanceof EventOperation) { output = input.map(ievent -> { this.getRuntimeStat().start(); try { InternalEvent i = ((EventOperation) op).perform(ievent); this.getSuccessCountStat().increment(); return i; } catch (OperationException e) { this.getErrorCountStat().increment(); logger.warn(e); return null; } finally { this.getRuntimeStat().stop(); } }); } else if (this.op instanceof MultiplexOperation) { output = input.flatMap(ievent -> { this.getRuntimeStat().start(); try { Stream<InternalEvent> s = ((MultiplexOperation) op).perform(ievent).stream(); this.getSuccessCountStat().increment(); return s; } catch (OperationException e) { this.getErrorCountStat().increment(); logger.warn(e); return Stream.empty(); } finally { this.getRuntimeStat().stop(); } }); } else if (this.op instanceof StreamOperation) { StreamOperation forkOp = (StreamOperation) this.op; output = forkOp.getOutputStream(input); } else if (this.op instanceof FilterOperation) { output = input.filter(ievent -> { try { return ((FilterOperation) this.op).test(ievent); } catch (OperationException e) { this.getErrorCountStat().increment(); logger.warn(e); return false; } finally { this.getRuntimeStat().stop(); } }); } else { throw new OperationException("Invalid type of operation"); } return output.filter(ievent -> { if (ievent == null) { logger.warn(op.getClass().getName() + " produced a null InternalEvent"); return false; } if (ievent.getEventObj() == null) { logger.warn(op.getClass().getName() + " produced a null DeserializedEvent"); return false; } if (ievent.getEventObj().getPayload() == null) { logger.warn(op.getClass().getName() + " produced a null DeserializedEvent payload"); return false; } return true; }); } OperationProcessor(OperationFactory operationFactory); Stream<InternalEvent> perform(Stream<InternalEvent> input); BaseOperation getOperation(); void setOperation(EventOperation operation);  }
@Test public void testNullInternalEventFiltering() throws JsonSyntaxException, UnsupportedEncodingException, IOException, OperationException { DummyOperation op = spy(new DummyOperation()); when(op.perform(any(InternalEvent.class))).thenReturn(null); DummyOperationFactory operationFactory = new DummyOperationFactory(op); OperationProcessor processor = new OperationProcessor(operationFactory); Stream<InternalEvent> stream = processor.perform(Stream.of(new InternalEvent("foo", null, 1))); List<InternalEvent> output = stream.collect(Collectors.toList()); assertEquals(0, output.size()); }
@Test public void testNullDeserializedEventFiltering() throws JsonSyntaxException, UnsupportedEncodingException, IOException, OperationException { DummyOperation op = spy(new DummyOperation()); InternalEvent retEvent = new InternalEvent("foo", null, 1); retEvent.setEventObj(null); when(op.perform(any(InternalEvent.class))).thenReturn(retEvent); DummyOperationFactory operationFactory = new DummyOperationFactory(op); OperationProcessor processor = new OperationProcessor(operationFactory); Stream<InternalEvent> stream = processor.perform(Stream.of(new InternalEvent("foo", null, 1))); List<InternalEvent> output = stream.collect(Collectors.toList()); assertEquals(0, output.size()); }
@Test public void testNullPayloadFiltering() throws JsonSyntaxException, UnsupportedEncodingException, IOException, OperationException { DummyOperation op = spy(new DummyOperation()); InternalEvent retEvent = new InternalEvent("foo", null, 1); retEvent.setEventObj(new DummyDeserializerHelper.DummyStringEvent(null)); when(op.perform(any(InternalEvent.class))).thenReturn(retEvent); DummyOperationFactory operationFactory = new DummyOperationFactory(op); OperationProcessor processor = new OperationProcessor(operationFactory); Stream<InternalEvent> stream = processor.perform(Stream.of(new InternalEvent("foo", null, 1))); List<InternalEvent> output = stream.collect(Collectors.toList()); assertEquals(0, output.size()); }
@Test public void testStatsLogging() throws JsonSyntaxException, UnsupportedEncodingException, IOException, OperationException { DummyOperationFactory mutatorFactory = new DummyOperationFactory(); OperationProcessor processor = new OperationProcessor(mutatorFactory); Stat runtimeStat = mock(Stat.class); Stat successStat = mock(Stat.class); Stat errorStat = mock(Stat.class); processor.setRuntimeStat(runtimeStat); processor.setSuccessCountStat(successStat); processor.setErrorCountStat(errorStat); InternalEvent ievent = new InternalEvent("foo", null, 1); ievent.setEventObj(new DummyDeserializerHelper.DummyStringEvent("test")); Stream<InternalEvent> stream = processor.perform(Stream.of(ievent)); List<InternalEvent> output = stream.collect(Collectors.toList()); verify(runtimeStat, times(1)).start(); verify(runtimeStat, times(1)).stop(); verify(successStat, times(1)).increment(); verify(errorStat, never()).increment(); assertEquals(1, output.size()); }
@Test public void testStatsLoggingOnError() { DummyOperation operation = mock(DummyOperation.class); DummyOperationFactory mutatorFactory = new DummyOperationFactory(operation); OperationProcessor processor = new OperationProcessor(mutatorFactory); InternalEvent ievent = new InternalEvent("a", null, 1); doThrow(new OperationException("Expceted")).when(operation).perform(ievent); Stat runtimeStat = mock(Stat.class); Stat successStat = mock(Stat.class); Stat errorStat = mock(Stat.class); processor.setRuntimeStat(runtimeStat); processor.setSuccessCountStat(successStat); processor.setErrorCountStat(errorStat); Stream<InternalEvent> stream = processor.perform(Stream.of(ievent)); List<InternalEvent> output = stream.collect(Collectors.toList()); verify(runtimeStat, times(1)).start(); verify(runtimeStat, times(1)).stop(); verify(successStat, never()).increment(); verify(errorStat, times(1)).increment(); assertEquals(0, output.size()); }