@SuppressWarnings( {"rawtypes", "unchecked"}) @Test public void testExecuteSource() { TestJob provider = new TestJob(); Job<Integer> job = provider.getJobInstance(); List<StageConfig<?, ?>> stages = job.getStages(); PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); int serverPort = portSelector.acquirePort(); WorkerPublisher producer = new WorkerPublisherRemoteObservable(serverPort, null, Observable.just(1), null); BehaviorSubject<Integer> workersInStageOneObservable = BehaviorSubject.create(1); StageExecutors.executeSource(0, job.getSource(), stages.get(0), producer, new Context(), workersInStageOneObservable); Iterator<Integer> iter = RemoteObservable.connect(new ConnectToObservable.Builder<Integer>() .host("localhost") .slotId("0") .port(serverPort) .decoder(Codecs.integer()) .build()) .getObservable() .toBlocking() .getIterator(); Assert.assertEquals(0, iter.next().intValue()); Assert.assertEquals(1, iter.next().intValue()); Assert.assertEquals(4, iter.next().intValue()); Assert.assertEquals(9, iter.next().intValue()); }
@SuppressWarnings( {"rawtypes", "unchecked"}) public static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage, WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable) { WorkerConsumer sourceConsumer = new WorkerConsumer() { @Override public Observable start(StageConfig stage) { Index index = new Index(workerIndex, totalWorkerAtStageObservable); source.getSourceFunction().init(context, index); Observable<Observable<?>> sourceObservable = (Observable) source.getSourceFunction().call(context, new Index(workerIndex, totalWorkerAtStageObservable)); return MantisMarker.sourceOut(sourceObservable); } @Override public void stop() {} }; executeIntermediate(sourceConsumer, stage, publisher, context); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage, WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable) { WorkerConsumer sourceConsumer = new WorkerConsumer() { @Override public Observable start(StageConfig stage) { Index index = new Index(workerIndex, totalWorkerAtStageObservable); source.getSourceFunction().init(context, index); Observable<Observable<?>> sourceObservable = (Observable) source.getSourceFunction().call(context, new Index(workerIndex, totalWorkerAtStageObservable)); return MantisMarker.sourceOut(sourceObservable); } @Override public void stop() {} }; executeIntermediate(sourceConsumer, stage, publisher, context); } }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage, WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable) { WorkerConsumer sourceConsumer = new WorkerConsumer() { @Override public Observable start(StageConfig stage) { Index index = new Index(workerIndex, totalWorkerAtStageObservable); source.getSourceFunction().init(context, index); Observable<Observable<?>> sourceObservable = (Observable) source.getSourceFunction().call(context, new Index(workerIndex, totalWorkerAtStageObservable)); return MantisMarker.sourceOut(sourceObservable); } @Override public void stop() {} }; executeIntermediate(sourceConsumer, stage, publisher, context); } private  StageExecutors(); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public static void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage, WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable) { WorkerConsumer sourceConsumer = new WorkerConsumer() { @Override public Observable start(StageConfig stage) { Index index = new Index(workerIndex, totalWorkerAtStageObservable); source.getSourceFunction().init(context, index); Observable<Observable<?>> sourceObservable = (Observable) source.getSourceFunction().call(context, new Index(workerIndex, totalWorkerAtStageObservable)); return MantisMarker.sourceOut(sourceObservable); } @Override public void stop() {} }; executeIntermediate(sourceConsumer, stage, publisher, context); } 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 void executeSource(final int workerIndex, final SourceHolder source, final StageConfig stage, WorkerPublisher publisher, final Context context, final Observable<Integer> totalWorkerAtStageObservable) { WorkerConsumer sourceConsumer = new WorkerConsumer() { @Override public Observable start(StageConfig stage) { Index index = new Index(workerIndex, totalWorkerAtStageObservable); source.getSourceFunction().init(context, index); Observable<Observable<?>> sourceObservable = (Observable) source.getSourceFunction().call(context, new Index(workerIndex, totalWorkerAtStageObservable)); return MantisMarker.sourceOut(sourceObservable); } @Override public void stop() {} }; executeIntermediate(sourceConsumer, stage, publisher, context); } 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);  }