ConditionalOperation implements StreamOperation { public Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input) { List<Stream<InternalEvent>> outputStreams = new ArrayList<Stream<InternalEvent>>(this.conditionsAndProcs.size()); this.filtersAndQueues = new ArrayList<Pair<FilterOperation, Queue<InternalEvent>>>(this.conditionsAndProcs.size()); for (Pair<FilterOperation, List<OperationProcessor>> filterAndProcs : this.conditionsAndProcs) { FilterOperation filter = filterAndProcs.getLeft(); List<OperationProcessor> procs = filterAndProcs.getRight(); Queue<InternalEvent> queue = new Queue<InternalEvent>(new LinkedBlockingQueue<InternalEvent>(procs.size())); this.filtersAndQueues .add(new ImmutablePair<FilterOperation, Queue<InternalEvent>>(filter, queue)); Stream<InternalEvent> conditionInput = queue.jdkStream(); for (OperationProcessor proc : procs) { conditionInput = proc.perform(conditionInput); } outputStreams.add(conditionInput); } Queue<InternalEvent> outputQueue = new Queue<InternalEvent>( new LinkedBlockingQueue<InternalEvent>(this.conditionsAndProcs.size())); AtomicInteger lock = new AtomicInteger(outputStreams.size()); outputStreams.forEach(stream -> { this.es.execute(new StreamToQueue(stream, outputQueue, lock)); }); new Thread(new Runnable() { @Override public void run() { input.forEach(ievent -> { boolean matches = false; for (Pair<FilterOperation, Queue<InternalEvent>> filterAndQueue : filtersAndQueues) { FilterOperation filter = filterAndQueue.getLeft(); if (filter.test(ievent)) { filterAndQueue.getRight().offer(ievent); matches = true; break; } } if (!matches && !filterNonMatch) { outputQueue.offer(ievent); } }); for (Pair<FilterOperation, Queue<InternalEvent>> filterAndQueue : filtersAndQueues) { filterAndQueue.getRight().close(); } } }).start(); return outputQueue.jdkStream(); } ConditionalOperation(
      List<Pair<FilterOperation, List<OperationProcessor>>> conditionsAndProcs,
      boolean filterNonMatch); Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input);  }
@Test public void testSingleConditionMatch() { List<Pair<FilterOperation, List<OperationProcessor>>> conditions = new ArrayList<Pair<FilterOperation, List<OperationProcessor>>>(); List<OperationProcessor> case1Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory pos = new DummyAppendOperationFactory(); DummyAppendOperationConfig posConf = new DummyAppendOperationConfig(); posConf.setAppendStr("+"); pos.setConf(posConf); case1Ops.add(new OperationProcessor(pos)); FilterOperation filter = new BasicFilterOperation(true); conditions.add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(filter, case1Ops)); ConditionalOperation op = new ConditionalOperation(conditions, false); 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+"); assertEquals(2, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testSingleConditionNoMatch() { List<Pair<FilterOperation, List<OperationProcessor>>> conditions = new ArrayList<Pair<FilterOperation, List<OperationProcessor>>>(); List<OperationProcessor> case1Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory pos = new DummyAppendOperationFactory(); DummyAppendOperationConfig posConf = new DummyAppendOperationConfig(); posConf.setAppendStr("+"); pos.setConf(posConf); case1Ops.add(new OperationProcessor(pos)); FilterOperation filter = new BasicFilterOperation(false); conditions.add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(filter, case1Ops)); ConditionalOperation op = new ConditionalOperation(conditions, false); 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"); assertEquals(2, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testTwoConditions() { List<Pair<FilterOperation, List<OperationProcessor>>> conditions = new ArrayList<Pair<FilterOperation, List<OperationProcessor>>>(); List<OperationProcessor> case1Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory pos = new DummyAppendOperationFactory(); DummyAppendOperationConfig posConf = new DummyAppendOperationConfig(); posConf.setAppendStr("+"); pos.setConf(posConf); case1Ops.add(new OperationProcessor(pos)); FilterOperation case1Filter = new BasicFilterOperation(false); conditions .add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(case1Filter, case1Ops)); List<OperationProcessor> case2Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory neg = new DummyAppendOperationFactory(); DummyAppendOperationConfig negConf = new DummyAppendOperationConfig(); negConf.setAppendStr("-"); neg.setConf(negConf); case2Ops.add(new OperationProcessor(neg)); FilterOperation case2Filter = new BasicFilterOperation(true); conditions .add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(case2Filter, case2Ops)); ConditionalOperation op = new ConditionalOperation(conditions, false); 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-"); assertEquals(2, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testTwoConditionsNoMatch() { List<Pair<FilterOperation, List<OperationProcessor>>> conditions = new ArrayList<Pair<FilterOperation, List<OperationProcessor>>>(); List<OperationProcessor> case1Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory pos = new DummyAppendOperationFactory(); DummyAppendOperationConfig posConf = new DummyAppendOperationConfig(); posConf.setAppendStr("+"); pos.setConf(posConf); case1Ops.add(new OperationProcessor(pos)); FilterOperation case1Filter = new BasicFilterOperation(false); conditions .add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(case1Filter, case1Ops)); List<OperationProcessor> case2Ops = new ArrayList<OperationProcessor>(); DummyAppendOperationFactory neg = new DummyAppendOperationFactory(); DummyAppendOperationConfig negConf = new DummyAppendOperationConfig(); negConf.setAppendStr("-"); neg.setConf(negConf); case2Ops.add(new OperationProcessor(neg)); FilterOperation case2Filter = new BasicFilterOperation(false); conditions .add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(case2Filter, case2Ops)); ConditionalOperation op = new ConditionalOperation(conditions, false); 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"); assertEquals(2, actual.size()); assertTrue(expected.containsAll(actual)); }
@Test public void testFilterCondition() { List<Pair<FilterOperation, List<OperationProcessor>>> conditions = new ArrayList<Pair<FilterOperation, List<OperationProcessor>>>(); List<OperationProcessor> case1Ops = new ArrayList<OperationProcessor>(); BasicFilterOperationFactory fOp = new BasicFilterOperationFactory(); BasicFilterOperationConfig fOpConf = new BasicFilterOperationConfig(); fOpConf.setPass(false); fOp.setConf(fOpConf); case1Ops.add(new OperationProcessor(fOp)); FilterOperation case1Filter = new BasicFilterOperation(true); conditions .add(new ImmutablePair<FilterOperation, List<OperationProcessor>>(case1Filter, case1Ops)); ConditionalOperation op = new ConditionalOperation(conditions, false); 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()); assertEquals(0, actual.size()); }