@SuppressWarnings( {"rawtypes", "unchecked"}) @Test public void testExecuteSink() throws InterruptedException { TestJob provider = new TestJob(); Job<Integer> job = provider.getJobInstance(); List<StageConfig<?, ?>> stages = job.getStages(); PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); 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(); } }); } }; Action0 noOpAction = new Action0() { @Override public void call() {} }; Action1<Throwable> noOpError = new Action1<Throwable>() { @Override public void call(Throwable t) {} }; WorkerConsumer consumer = new WorkerConsumerRemoteObservable(null, staticEndpoints); StageExecutors.executeSink(consumer, stages.get(1), job.getSink(), new TestPortSelector(), new RxMetrics(), new Context(), noOpAction, null, null, noOpAction, noOpError); Iterator<Integer> iter = provider.getItemsWritten().iterator(); Assert.assertEquals(0, iter.next().intValue()); Assert.assertEquals(2, iter.next().intValue()); Assert.assertEquals(4, iter.next().intValue()); }
@SuppressWarnings( {"rawtypes", "unchecked"}) public 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) { WorkerPublisher sinkPublisher = new SinkPublisher(sink, portSelector, context, sinkObservableCompletedCallback, onSinkSubscribe, onSinkUnsubscribe, observableOnCompleteCallback, observableOnErrorCallback); executeIntermediate(consumer, stage, sinkPublisher, context); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public 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) { WorkerPublisher sinkPublisher = new SinkPublisher(sink, portSelector, context, sinkObservableCompletedCallback, onSinkSubscribe, onSinkUnsubscribe, observableOnCompleteCallback, observableOnErrorCallback); executeIntermediate(consumer, stage, sinkPublisher, context); } }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public 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) { WorkerPublisher sinkPublisher = new SinkPublisher(sink, portSelector, context, sinkObservableCompletedCallback, onSinkSubscribe, onSinkUnsubscribe, observableOnCompleteCallback, observableOnErrorCallback); executeIntermediate(consumer, stage, sinkPublisher, context); } private  StageExecutors(); }
StageExecutors { @SuppressWarnings( {"rawtypes", "unchecked"}) public 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) { WorkerPublisher sinkPublisher = new SinkPublisher(sink, portSelector, context, sinkObservableCompletedCallback, onSinkSubscribe, onSinkUnsubscribe, observableOnCompleteCallback, observableOnErrorCallback); executeIntermediate(consumer, stage, sinkPublisher, 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 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) { WorkerPublisher sinkPublisher = new SinkPublisher(sink, portSelector, context, sinkObservableCompletedCallback, onSinkSubscribe, onSinkUnsubscribe, observableOnCompleteCallback, observableOnErrorCallback); executeIntermediate(consumer, stage, sinkPublisher, 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);  }