ForkOperation implements StreamOperation { public Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input) { List<Stream<InternalEvent>> forkOutputStreams = new ArrayList<Stream<InternalEvent>>(opProcsInForks.size()); this.queues = new ArrayList<Queue<InternalEvent>>(opProcsInForks.size()); for (List<OperationProcessor> opProcsInFork : opProcsInForks) { Queue<InternalEvent> queue = new Queue<InternalEvent>(new LinkedBlockingQueue<InternalEvent>(opProcsInFork.size())); this.queues.add(queue); Stream<InternalEvent> forkInput = queue.jdkStream(); for (OperationProcessor opProcInFork : opProcsInFork) { forkInput = opProcInFork.perform(forkInput); } forkOutputStreams.add(forkInput); } Queue<InternalEvent> outputQueue = new Queue<InternalEvent>(new LinkedBlockingQueue<InternalEvent>(this.queues.size())); AtomicInteger lock = new AtomicInteger(forkOutputStreams.size()); forkOutputStreams.forEach(stream -> { this.es.execute(new StreamToQueue(stream, outputQueue, lock)); }); new Thread(new Runnable() { @Override public void run() { input.forEach(ievent -> { queues.forEach(queue -> { queue.offer(ievent.copy()); }); }); for (Queue<InternalEvent> queue : queues) { queue.close(); } } }).start(); return outputQueue.jdkStream(); } ForkOperation(List<List<OperationProcessor>> opProcsInForks); Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input);  }
@Test public void testSingleFork() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork1); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(2, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); List<String> expected = Arrays.asList("0", "1"); assertEquals(2, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testTwoForks() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); fork2.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork2); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(2, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); List<String> expected = Arrays.asList("0", "1", "0", "1"); assertEquals(4, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testTwoForksSlow() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyThrottleOperationFactory())); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); fork2.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork2); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(1000, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); assertEquals(2000, actual.size()); }
@Test public void testThreeForks() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyThrottleOperationFactory())); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); fork2.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork2); List<OperationProcessor> fork3 = new ArrayList<OperationProcessor>(); fork3.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork3); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(10, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); assertEquals(30, actual.size()); }
@Test public void testNullOpFork1() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); DummyNullOperation nullOp = new DummyNullOperation(); fork2.add(new OperationProcessor(new DummyOperationFactory(nullOp))); forks.add(fork2); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(10, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); assertEquals(10, actual.size()); }
@Test public void testNullOpFork() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); DummyNullOperation nullOp = new DummyNullOperation(); fork1.add(new OperationProcessor(new DummyOperationFactory(nullOp))); forks.add(fork1); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(10, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(InternalEvent::getEventString).collect(Collectors.toList()); assertEquals(0, actual.size()); }
@Test public void testForkMonitoringSeries() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyThrottleOperationFactory())); fork1.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork1); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(10, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> outputEvents = output.map(InternalEvent::getEventString).collect(Collectors.toList()); Monitor m = Monitor.getInstance(); List<Stat> stats = m.getStats(); Map<String, String> actual = new HashMap<String, String>(); for (Stat stat : stats) { String key = String.format("%s.%s", stat.getTags().iterator().next().getValue(), stat.getName()); actual.put(key, "" + stat.getValue()); } assertEquals("0", actual .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.error.count")); assertEquals("10", actual .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.success.count")); assertEquals("10", actual.get( "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.success.count")); assertEquals("0", actual.get( "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.error.count")); }
@Test public void testForkMonitoringParallel() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); fork1.add(new OperationProcessor(new DummyOperationFactory())); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); fork2.add(new OperationProcessor(new DummyThrottleOperationFactory())); forks.add(fork2); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(10, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> outputEvents = output.map(InternalEvent::getEventString).collect(Collectors.toList()); Monitor m = Monitor.getInstance(); List<Stat> stats = m.getStats(); Map<String, String> actual = new HashMap<String, String>(); for (Stat stat : stats) { String key = String.format("%s.%s", stat.getTags().iterator().next().getValue(), stat.getName()); actual.put(key, "" + stat.getValue()); } assertEquals("0", actual .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.error.count")); assertEquals("10", actual .get("com.nextdoor.bender.testutils.DummyOperationHelper.DummyOperation.success.count")); assertEquals("10", actual.get( "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.success.count")); assertEquals("0", actual.get( "com.nextdoor.bender.testutils.DummyThrottleOperationHelper.DummyThrottleOperation.error.count")); }
@Test public void testEventCloning() { List<List<OperationProcessor>> forks = new ArrayList<List<OperationProcessor>>(); List<OperationProcessor> fork1 = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory pos = new DummyAppendOperationFactory(); DummyAppendOperationConfig posConf = new DummyAppendOperationConfig(); posConf.setAppendStr("+"); pos.setConf(posConf); fork1.add(new OperationProcessor(pos)); forks.add(fork1); List<OperationProcessor> fork2 = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory neg = new DummyAppendOperationFactory(); DummyAppendOperationConfig negConf = new DummyAppendOperationConfig(); negConf.setAppendStr("-"); neg.setConf(negConf); fork2.add(new OperationProcessor(neg)); forks.add(fork2); ForkOperation op = new ForkOperation(forks); Queue<InternalEvent> inputQueue = new Queue<InternalEvent>(); supply(2, inputQueue); Stream<InternalEvent> input = inputQueue.stream(); Stream<InternalEvent> output = op.getOutputStream(input); List<String> actual = output.map(m -> { return m.getEventObj().getPayload().toString(); }).collect(Collectors.toList()); List<String> expected = Arrays.asList("0+", "1+", "0-", "1-"); assertEquals(4, actual.size()); assertTrue(expected.containsAll(actual)); }