@Test public void test() throws Exception { Forker forker = new Forker(); MyFlowable<StreamEntity<byte[]>> flowable = new MyFlowable<>(); RecordStreamWithMetadata<byte[], String> stream = new RecordStreamWithMetadata<>(flowable, GlobalMetadata.<String>builder().schema("schema").build()); WorkUnitState workUnitState = new WorkUnitState(); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, "3"); Forker.ForkedStream<byte[], String> forkedStream = forker.forkStream(stream, new MyForkOperator(), workUnitState); Assert.assertEquals(forkedStream.getForkedStreams().size(), 3); Queue<StreamEntity<byte[]>> output0 = new LinkedList<>(); forkedStream.getForkedStreams().get(0).getRecordStream().subscribe(output0::add); Queue<StreamEntity<byte[]>> output1 = new LinkedList<>(); forkedStream.getForkedStreams().get(1).getRecordStream().subscribe(output1::add); Queue<StreamEntity<byte[]>> output2 = new LinkedList<>(); forkedStream.getForkedStreams().get(2).getRecordStream().subscribe(output2::add); flowable._subscriber.onNext(new RecordEnvelope<>(new byte[]{1, 1, 1})); Assert.assertTrue(output0.poll() instanceof RecordEnvelope); Assert.assertTrue(output1.poll() instanceof RecordEnvelope); Assert.assertTrue(output2.poll() instanceof RecordEnvelope); flowable._subscriber.onNext(new RecordEnvelope<>(new byte[]{1, 0, 0})); Assert.assertTrue(output0.poll() instanceof RecordEnvelope); Assert.assertNull(output1.poll()); Assert.assertNull(output2.poll()); flowable._subscriber.onNext(new RecordEnvelope<>(new byte[]{0, 1, 1})); Assert.assertNull(output0.poll()); Assert.assertTrue(output1.poll() instanceof RecordEnvelope); Assert.assertTrue(output2.poll() instanceof RecordEnvelope); flowable._subscriber.onNext(new BasicTestControlMessage<byte[]>("control")); Assert.assertTrue(output0.poll() instanceof BasicTestControlMessage); Assert.assertTrue(output1.poll() instanceof BasicTestControlMessage); Assert.assertTrue(output2.poll() instanceof BasicTestControlMessage); flowable._subscriber.onComplete(); }
public <D, S> ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState) throws Exception { int branches = forkOperator.getBranches(workUnitState); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, branches); forkOperator.init(workUnitState); List<Boolean> forkedSchemas = forkOperator.forkSchema(workUnitState, inputStream.getGlobalMetadata().getSchema()); int activeForks = (int) forkedSchemas.stream().filter(b -> b).count(); Preconditions.checkState(forkedSchemas.size() == branches, String .format("Number of forked schemas [%d] is not equal to number of branches [%d]", forkedSchemas.size(), branches)); Flowable<RecordWithForkMap<D>> forkedStream = inputStream.getRecordStream().map(r -> { if (r instanceof RecordEnvelope) { RecordEnvelope<D> recordEnvelope = (RecordEnvelope<D>) r; return new RecordWithForkMap<>(recordEnvelope, forkOperator.forkDataRecord(workUnitState, recordEnvelope.getRecord())); } else if (r instanceof ControlMessage) { return new RecordWithForkMap<D>((ControlMessage<D>) r, branches); } else { throw new IllegalStateException("Expected RecordEnvelope or ControlMessage."); } }); if (activeForks > 1) { forkedStream = forkedStream.share(); } List<RecordStreamWithMetadata<D, S>> forkStreams = Lists.newArrayList(); boolean mustCopy = mustCopy(forkedSchemas); for(int i = 0; i < forkedSchemas.size(); i++) { if (forkedSchemas.get(i)) { final int idx = i; Flowable<StreamEntity<D>> thisStream = forkedStream.filter(new ForkFilter<>(idx)).map(RecordWithForkMap::getRecordCopyIfNecessary); forkStreams.add(inputStream.withRecordStream(thisStream, mustCopy ? (GlobalMetadata<S>) CopyHelper.copy(inputStream.getGlobalMetadata()) : inputStream.getGlobalMetadata())); } else { forkStreams.add(null); } } return new ForkedStream<>(forkStreams); }
Forker { public <D, S> ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState) throws Exception { int branches = forkOperator.getBranches(workUnitState); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, branches); forkOperator.init(workUnitState); List<Boolean> forkedSchemas = forkOperator.forkSchema(workUnitState, inputStream.getGlobalMetadata().getSchema()); int activeForks = (int) forkedSchemas.stream().filter(b -> b).count(); Preconditions.checkState(forkedSchemas.size() == branches, String .format("Number of forked schemas [%d] is not equal to number of branches [%d]", forkedSchemas.size(), branches)); Flowable<RecordWithForkMap<D>> forkedStream = inputStream.getRecordStream().map(r -> { if (r instanceof RecordEnvelope) { RecordEnvelope<D> recordEnvelope = (RecordEnvelope<D>) r; return new RecordWithForkMap<>(recordEnvelope, forkOperator.forkDataRecord(workUnitState, recordEnvelope.getRecord())); } else if (r instanceof ControlMessage) { return new RecordWithForkMap<D>((ControlMessage<D>) r, branches); } else { throw new IllegalStateException("Expected RecordEnvelope or ControlMessage."); } }); if (activeForks > 1) { forkedStream = forkedStream.share(); } List<RecordStreamWithMetadata<D, S>> forkStreams = Lists.newArrayList(); boolean mustCopy = mustCopy(forkedSchemas); for(int i = 0; i < forkedSchemas.size(); i++) { if (forkedSchemas.get(i)) { final int idx = i; Flowable<StreamEntity<D>> thisStream = forkedStream.filter(new ForkFilter<>(idx)).map(RecordWithForkMap::getRecordCopyIfNecessary); forkStreams.add(inputStream.withRecordStream(thisStream, mustCopy ? (GlobalMetadata<S>) CopyHelper.copy(inputStream.getGlobalMetadata()) : inputStream.getGlobalMetadata())); } else { forkStreams.add(null); } } return new ForkedStream<>(forkStreams); } }
Forker { public <D, S> ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState) throws Exception { int branches = forkOperator.getBranches(workUnitState); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, branches); forkOperator.init(workUnitState); List<Boolean> forkedSchemas = forkOperator.forkSchema(workUnitState, inputStream.getGlobalMetadata().getSchema()); int activeForks = (int) forkedSchemas.stream().filter(b -> b).count(); Preconditions.checkState(forkedSchemas.size() == branches, String .format("Number of forked schemas [%d] is not equal to number of branches [%d]", forkedSchemas.size(), branches)); Flowable<RecordWithForkMap<D>> forkedStream = inputStream.getRecordStream().map(r -> { if (r instanceof RecordEnvelope) { RecordEnvelope<D> recordEnvelope = (RecordEnvelope<D>) r; return new RecordWithForkMap<>(recordEnvelope, forkOperator.forkDataRecord(workUnitState, recordEnvelope.getRecord())); } else if (r instanceof ControlMessage) { return new RecordWithForkMap<D>((ControlMessage<D>) r, branches); } else { throw new IllegalStateException("Expected RecordEnvelope or ControlMessage."); } }); if (activeForks > 1) { forkedStream = forkedStream.share(); } List<RecordStreamWithMetadata<D, S>> forkStreams = Lists.newArrayList(); boolean mustCopy = mustCopy(forkedSchemas); for(int i = 0; i < forkedSchemas.size(); i++) { if (forkedSchemas.get(i)) { final int idx = i; Flowable<StreamEntity<D>> thisStream = forkedStream.filter(new ForkFilter<>(idx)).map(RecordWithForkMap::getRecordCopyIfNecessary); forkStreams.add(inputStream.withRecordStream(thisStream, mustCopy ? (GlobalMetadata<S>) CopyHelper.copy(inputStream.getGlobalMetadata()) : inputStream.getGlobalMetadata())); } else { forkStreams.add(null); } } return new ForkedStream<>(forkStreams); }  }
Forker { public <D, S> ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState) throws Exception { int branches = forkOperator.getBranches(workUnitState); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, branches); forkOperator.init(workUnitState); List<Boolean> forkedSchemas = forkOperator.forkSchema(workUnitState, inputStream.getGlobalMetadata().getSchema()); int activeForks = (int) forkedSchemas.stream().filter(b -> b).count(); Preconditions.checkState(forkedSchemas.size() == branches, String .format("Number of forked schemas [%d] is not equal to number of branches [%d]", forkedSchemas.size(), branches)); Flowable<RecordWithForkMap<D>> forkedStream = inputStream.getRecordStream().map(r -> { if (r instanceof RecordEnvelope) { RecordEnvelope<D> recordEnvelope = (RecordEnvelope<D>) r; return new RecordWithForkMap<>(recordEnvelope, forkOperator.forkDataRecord(workUnitState, recordEnvelope.getRecord())); } else if (r instanceof ControlMessage) { return new RecordWithForkMap<D>((ControlMessage<D>) r, branches); } else { throw new IllegalStateException("Expected RecordEnvelope or ControlMessage."); } }); if (activeForks > 1) { forkedStream = forkedStream.share(); } List<RecordStreamWithMetadata<D, S>> forkStreams = Lists.newArrayList(); boolean mustCopy = mustCopy(forkedSchemas); for(int i = 0; i < forkedSchemas.size(); i++) { if (forkedSchemas.get(i)) { final int idx = i; Flowable<StreamEntity<D>> thisStream = forkedStream.filter(new ForkFilter<>(idx)).map(RecordWithForkMap::getRecordCopyIfNecessary); forkStreams.add(inputStream.withRecordStream(thisStream, mustCopy ? (GlobalMetadata<S>) CopyHelper.copy(inputStream.getGlobalMetadata()) : inputStream.getGlobalMetadata())); } else { forkStreams.add(null); } } return new ForkedStream<>(forkStreams); }  ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState); }
Forker { public <D, S> ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState) throws Exception { int branches = forkOperator.getBranches(workUnitState); workUnitState.setProp(ConfigurationKeys.FORK_BRANCHES_KEY, branches); forkOperator.init(workUnitState); List<Boolean> forkedSchemas = forkOperator.forkSchema(workUnitState, inputStream.getGlobalMetadata().getSchema()); int activeForks = (int) forkedSchemas.stream().filter(b -> b).count(); Preconditions.checkState(forkedSchemas.size() == branches, String .format("Number of forked schemas [%d] is not equal to number of branches [%d]", forkedSchemas.size(), branches)); Flowable<RecordWithForkMap<D>> forkedStream = inputStream.getRecordStream().map(r -> { if (r instanceof RecordEnvelope) { RecordEnvelope<D> recordEnvelope = (RecordEnvelope<D>) r; return new RecordWithForkMap<>(recordEnvelope, forkOperator.forkDataRecord(workUnitState, recordEnvelope.getRecord())); } else if (r instanceof ControlMessage) { return new RecordWithForkMap<D>((ControlMessage<D>) r, branches); } else { throw new IllegalStateException("Expected RecordEnvelope or ControlMessage."); } }); if (activeForks > 1) { forkedStream = forkedStream.share(); } List<RecordStreamWithMetadata<D, S>> forkStreams = Lists.newArrayList(); boolean mustCopy = mustCopy(forkedSchemas); for(int i = 0; i < forkedSchemas.size(); i++) { if (forkedSchemas.get(i)) { final int idx = i; Flowable<StreamEntity<D>> thisStream = forkedStream.filter(new ForkFilter<>(idx)).map(RecordWithForkMap::getRecordCopyIfNecessary); forkStreams.add(inputStream.withRecordStream(thisStream, mustCopy ? (GlobalMetadata<S>) CopyHelper.copy(inputStream.getGlobalMetadata()) : inputStream.getGlobalMetadata())); } else { forkStreams.add(null); } } return new ForkedStream<>(forkStreams); }  ForkedStream<D, S> forkStream(RecordStreamWithMetadata<D, S> inputStream, ForkOperator<S, D> forkOperator, WorkUnitState workUnitState);  }