@SuppressWarnings( {"rawtypes", "unchecked"}) @Test public void testExecuteIntermediatStage() throws InterruptedException { TestJob provider = new TestJob(); Job<Integer> job = provider.getJobInstance(); List<StageConfig<?, ?>> stages = job.getStages(); PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); final int publishPort = portSelector.acquirePort(); final int consumerPort = portSelector.acquirePort(); RemoteRxServer server1 = RemoteObservable.serve(consumerPort, Observable.range(0, 10), Codecs.integer()); server1.start(); EndpointInjector staticEndpoints = new EndpointInjector() { @Override public Observable<EndpointChange> deltas() { return Observable.create(new OnSubscribe<EndpointChange>() { @Override public void call(Subscriber<? super EndpointChange> subscriber) { subscriber.onNext(new EndpointChange(EndpointChange.Type.add, new Endpoint("localhost", consumerPort, "1"))); subscriber.onCompleted(); } }); } }; WorkerConsumer consumer = new WorkerConsumerRemoteObservable(null, staticEndpoints); WorkerPublisher producer = new WorkerPublisherRemoteObservable(publishPort, null, Observable.just(1), null); StageExecutors.executeIntermediate(consumer, stages.get(1), producer, new Context()); Iterator<Integer> iter = RemoteObservable.connect(new ConnectToObservable.Builder<Integer>() .host("localhost") .slotId("0") .port(publishPort) .decoder(Codecs.integer()) .build()) .getObservable() .toBlocking() .getIterator(); Assert.assertEquals(0, iter.next().intValue()); Assert.assertEquals(2, iter.next().intValue()); Assert.assertEquals(4, iter.next().intValue()); }
@SuppressWarnings( {"rawtypes", "unchecked"}) public static <T, R> void executeIntermediate(WorkerConsumer consumer, final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context) { if (consumer == null) { throw new IllegalArgumentException("consumer cannot be null"); } if (stage == null) { throw new IllegalArgumentException("stage cannot be null"); } if (publisher == null) { throw new IllegalArgumentException("producer cannot be null"); } Observable<?> toSink = null; if (stage instanceof ScalarToScalar) { ScalarToScalar scalarStage = (ScalarToScalar) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToScalarStage(scalarStage, source, context); } else if (stage instanceof ScalarToKey) { ScalarToKey scalarStage = (ScalarToKey) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToKeyStage(scalarStage, source, context); } else if (stage instanceof ScalarToGroup) { ScalarToGroup scalarStage = (ScalarToGroup) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToGroupStage(scalarStage, source, context); } else if (stage instanceof KeyToKey) { KeyToKey keyToKey = (KeyToKey) stage; Observable<Observable<GroupedObservable<String, T>>> source = consumer.start(keyToKey); toSink = setupKeyToKeyStage(keyToKey, source, context); } else if (stage instanceof GroupToGroup) { GroupToGroup groupToGroup = (GroupToGroup) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToGroup); toSink = setupGroupToGroupStage(groupToGroup, source, context); } else if (stage instanceof KeyToScalar) { KeyToScalar scalarToKey = (KeyToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(scalarToKey); toSink = setupKeyToScalarStage(scalarToKey, source, context); } else if (stage instanceof GroupToScalar) { GroupToScalar groupToScalar = (GroupToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToScalar); toSink = setupGroupToScalarStage(groupToScalar, source, context); } publisher.start(stage, toSink); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static <T, R> void executeIntermediate(WorkerConsumer consumer, final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context) { if (consumer == null) { throw new IllegalArgumentException("consumer cannot be null"); } if (stage == null) { throw new IllegalArgumentException("stage cannot be null"); } if (publisher == null) { throw new IllegalArgumentException("producer cannot be null"); } Observable<?> toSink = null; if (stage instanceof ScalarToScalar) { ScalarToScalar scalarStage = (ScalarToScalar) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToScalarStage(scalarStage, source, context); } else if (stage instanceof ScalarToKey) { ScalarToKey scalarStage = (ScalarToKey) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToKeyStage(scalarStage, source, context); } else if (stage instanceof ScalarToGroup) { ScalarToGroup scalarStage = (ScalarToGroup) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToGroupStage(scalarStage, source, context); } else if (stage instanceof KeyToKey) { KeyToKey keyToKey = (KeyToKey) stage; Observable<Observable<GroupedObservable<String, T>>> source = consumer.start(keyToKey); toSink = setupKeyToKeyStage(keyToKey, source, context); } else if (stage instanceof GroupToGroup) { GroupToGroup groupToGroup = (GroupToGroup) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToGroup); toSink = setupGroupToGroupStage(groupToGroup, source, context); } else if (stage instanceof KeyToScalar) { KeyToScalar scalarToKey = (KeyToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(scalarToKey); toSink = setupKeyToScalarStage(scalarToKey, source, context); } else if (stage instanceof GroupToScalar) { GroupToScalar groupToScalar = (GroupToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToScalar); toSink = setupGroupToScalarStage(groupToScalar, source, context); } publisher.start(stage, toSink); } }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static <T, R> void executeIntermediate(WorkerConsumer consumer, final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context) { if (consumer == null) { throw new IllegalArgumentException("consumer cannot be null"); } if (stage == null) { throw new IllegalArgumentException("stage cannot be null"); } if (publisher == null) { throw new IllegalArgumentException("producer cannot be null"); } Observable<?> toSink = null; if (stage instanceof ScalarToScalar) { ScalarToScalar scalarStage = (ScalarToScalar) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToScalarStage(scalarStage, source, context); } else if (stage instanceof ScalarToKey) { ScalarToKey scalarStage = (ScalarToKey) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToKeyStage(scalarStage, source, context); } else if (stage instanceof ScalarToGroup) { ScalarToGroup scalarStage = (ScalarToGroup) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToGroupStage(scalarStage, source, context); } else if (stage instanceof KeyToKey) { KeyToKey keyToKey = (KeyToKey) stage; Observable<Observable<GroupedObservable<String, T>>> source = consumer.start(keyToKey); toSink = setupKeyToKeyStage(keyToKey, source, context); } else if (stage instanceof GroupToGroup) { GroupToGroup groupToGroup = (GroupToGroup) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToGroup); toSink = setupGroupToGroupStage(groupToGroup, source, context); } else if (stage instanceof KeyToScalar) { KeyToScalar scalarToKey = (KeyToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(scalarToKey); toSink = setupKeyToScalarStage(scalarToKey, source, context); } else if (stage instanceof GroupToScalar) { GroupToScalar groupToScalar = (GroupToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToScalar); toSink = setupGroupToScalarStage(groupToScalar, source, context); } publisher.start(stage, toSink); } private  StageExecutors(); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static <T, R> void executeIntermediate(WorkerConsumer consumer, final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context) { if (consumer == null) { throw new IllegalArgumentException("consumer cannot be null"); } if (stage == null) { throw new IllegalArgumentException("stage cannot be null"); } if (publisher == null) { throw new IllegalArgumentException("producer cannot be null"); } Observable<?> toSink = null; if (stage instanceof ScalarToScalar) { ScalarToScalar scalarStage = (ScalarToScalar) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToScalarStage(scalarStage, source, context); } else if (stage instanceof ScalarToKey) { ScalarToKey scalarStage = (ScalarToKey) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToKeyStage(scalarStage, source, context); } else if (stage instanceof ScalarToGroup) { ScalarToGroup scalarStage = (ScalarToGroup) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToGroupStage(scalarStage, source, context); } else if (stage instanceof KeyToKey) { KeyToKey keyToKey = (KeyToKey) stage; Observable<Observable<GroupedObservable<String, T>>> source = consumer.start(keyToKey); toSink = setupKeyToKeyStage(keyToKey, source, context); } else if (stage instanceof GroupToGroup) { GroupToGroup groupToGroup = (GroupToGroup) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToGroup); toSink = setupGroupToGroupStage(groupToGroup, source, context); } else if (stage instanceof KeyToScalar) { KeyToScalar scalarToKey = (KeyToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(scalarToKey); toSink = setupKeyToScalarStage(scalarToKey, source, context); } else if (stage instanceof GroupToScalar) { GroupToScalar groupToScalar = (GroupToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToScalar); toSink = setupGroupToScalarStage(groupToScalar, source, context); } publisher.start(stage, toSink); } private  StageExecutors(); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSingleStageJob(final SourceHolder source, final StageConfig stage,
                                             final SinkHolder sink, final PortSelector portSelector, RxMetrics rxMetrics,
                                             final Context context, Action0 sinkObservableTerminatedCallback,
                                             final int workerIndex,
                                             final Observable<Integer> totalWorkerAtStageObservable,
                                             final Action0 onSinkSubscribe, final Action0 onSinkUnsubscribe,
                                             Action0 observableOnCompleteCallback, Action1<Throwable> observableOnErrorCallback); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage,
                                     WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeIntermediate(WorkerConsumer consumer,
                                                  final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSink(WorkerConsumer consumer, StageConfig stage, SinkHolder sink,
                                   PortSelector portSelector, RxMetrics rxMetrics, Context context,
                                   Action0 sinkObservableCompletedCallback,
                                   final Action0 onSinkSubscribe, final Action0 onSinkUnsubscribe,
                                   Action0 observableOnCompleteCallback, Action1<Throwable> observableOnErrorCallback); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static <T, R> void executeIntermediate(WorkerConsumer consumer, final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context) { if (consumer == null) { throw new IllegalArgumentException("consumer cannot be null"); } if (stage == null) { throw new IllegalArgumentException("stage cannot be null"); } if (publisher == null) { throw new IllegalArgumentException("producer cannot be null"); } Observable<?> toSink = null; if (stage instanceof ScalarToScalar) { ScalarToScalar scalarStage = (ScalarToScalar) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToScalarStage(scalarStage, source, context); } else if (stage instanceof ScalarToKey) { ScalarToKey scalarStage = (ScalarToKey) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToKeyStage(scalarStage, source, context); } else if (stage instanceof ScalarToGroup) { ScalarToGroup scalarStage = (ScalarToGroup) stage; Observable<Observable<T>> source = consumer.start(scalarStage); toSink = setupScalarToGroupStage(scalarStage, source, context); } else if (stage instanceof KeyToKey) { KeyToKey keyToKey = (KeyToKey) stage; Observable<Observable<GroupedObservable<String, T>>> source = consumer.start(keyToKey); toSink = setupKeyToKeyStage(keyToKey, source, context); } else if (stage instanceof GroupToGroup) { GroupToGroup groupToGroup = (GroupToGroup) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToGroup); toSink = setupGroupToGroupStage(groupToGroup, source, context); } else if (stage instanceof KeyToScalar) { KeyToScalar scalarToKey = (KeyToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(scalarToKey); toSink = setupKeyToScalarStage(scalarToKey, source, context); } else if (stage instanceof GroupToScalar) { GroupToScalar groupToScalar = (GroupToScalar) stage; Observable<Observable<MantisGroup<String, T>>> source = consumer.start(groupToScalar); toSink = setupGroupToScalarStage(groupToScalar, source, context); } publisher.start(stage, toSink); } private  StageExecutors(); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSingleStageJob(final SourceHolder source, final StageConfig stage,
                                             final SinkHolder sink, final PortSelector portSelector, RxMetrics rxMetrics,
                                             final Context context, Action0 sinkObservableTerminatedCallback,
                                             final int workerIndex,
                                             final Observable<Integer> totalWorkerAtStageObservable,
                                             final Action0 onSinkSubscribe, final Action0 onSinkUnsubscribe,
                                             Action0 observableOnCompleteCallback, Action1<Throwable> observableOnErrorCallback); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage,
                                     WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeIntermediate(WorkerConsumer consumer,
                                                  final StageConfig<T, R> stage, WorkerPublisher publisher, final Context context); @SuppressWarnings( {"rawtypes", "unchecked"}) static void executeSink(WorkerConsumer consumer, StageConfig stage, SinkHolder sink,
                                   PortSelector portSelector, RxMetrics rxMetrics, Context context,
                                   Action0 sinkObservableCompletedCallback,
                                   final Action0 onSinkSubscribe, final Action0 onSinkUnsubscribe,
                                   Action0 observableOnCompleteCallback, Action1<Throwable> observableOnErrorCallback);  }