in mantis-remote-observable/src/main/java/io/reactivex/mantis/remote/observable/RemoteObservable.java [496:589]
private static <T> Observable<T> createTcpConnectionToServer(final ConnectToObservable<T> params,
final RemoteUnsubscribe remoteUnsubscribe, final RxMetrics metrics,
final Action0 connectionDisconnectCallback, Observable<Integer> closeTrigger) {
final Decoder<T> decoder = params.getDecoder();
loadFastProperties();
return
RxNetty.createTcpClient(params.getHost(), params.getPort(), new PipelineConfiguratorComposite<RemoteRxEvent, List<RemoteRxEvent>>(
new PipelineConfigurator<RemoteRxEvent, List<RemoteRxEvent>>() {
@Override
public void configureNewPipeline(ChannelPipeline pipeline) {
if (enableNettyLogging) {
pipeline.addFirst(new LoggingHandler(LogLevel.ERROR)); // uncomment to enable debug logging
}
if (enableHeartBeating) {
pipeline.addLast("idleStateHandler", new IdleStateHandler(10, 2, 0));
pipeline.addLast("heartbeat", new HeartbeatHandler());
}
if (enableCompression) {
pipeline.addLast("gzipInflater", new JdkZlibEncoder(ZlibWrapper.GZIP));
pipeline.addLast("gzipDeflater", new JdkZlibDecoder(ZlibWrapper.GZIP));
}
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4)); // 4 bytes to encode length
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(maxFrameLength, 0, 4, 0, 4)); // max frame = half MB
}
}, new BatchedRxEventPipelineConfigurator()))
.connect()
// send subscription request, get input stream
.flatMap(new Func1<ObservableConnection<RemoteRxEvent, List<RemoteRxEvent>>, Observable<RemoteRxEvent>>() {
@Override
public Observable<RemoteRxEvent> call(final ObservableConnection<RemoteRxEvent, List<RemoteRxEvent>> connection) {
connection.writeAndFlush(RemoteRxEvent.subscribed(params.getName(), params.getSubscribeParameters())); // send subscribe event to server
remoteUnsubscribe.setConnection(connection);
return connection.getInput()
.lift(new DropOperator<RemoteRxEvent>("incoming_" + RemoteObservable.class.getCanonicalName() + "_createTcpConnectionToServer"))
.rebatchRequests(bufferSize <= 0 ? 1 : bufferSize);
}
})
.doOnCompleted(new Action0() {
@Override
public void call() {
// connection completed
logger.warn("Detected connection completed when trying to connect to host: " + params.getHost() + " port: " + params.getPort());
connectionDisconnectCallback.call();
}
})
.onErrorResumeNext(new Func1<Throwable, Observable<RemoteRxEvent>>() {
@Override
public Observable<RemoteRxEvent> call(Throwable t1) {
logger.warn("Detected connection error when trying to connect to host: " + params.getHost() + " port: " + params.getPort(), t1);
connectionDisconnectCallback.call();
// complete if error occurs
return Observable.empty();
}
})
.takeUntil(closeTrigger)
.map(new Func1<RemoteRxEvent, Notification<T>>() {
@Override
public Notification<T> call(RemoteRxEvent rxEvent) {
if (rxEvent.getType() == RemoteRxEvent.Type.next) {
metrics.incrementNextCount();
return Notification.createOnNext(decoder.decode(rxEvent.getData()));
} else if (rxEvent.getType() == RemoteRxEvent.Type.error) {
metrics.incrementErrorCount();
return Notification.createOnError(fromBytesToThrowable(rxEvent.getData()));
} else if (rxEvent.getType() == RemoteRxEvent.Type.completed) {
metrics.incrementCompletedCount();
return Notification.createOnCompleted();
} else {
throw new RuntimeException("RemoteRxEvent of type: " + rxEvent.getType() + ", not supported.");
}
}
})
.<T>dematerialize()
.doOnEach(new Observer<T>() {
@Override
public void onCompleted() {
logger.info("RemoteRxEvent: " + params.getName() + " onCompleted()");
}
@Override
public void onError(Throwable e) {
logger.error("RemoteRxEvent: " + params.getName() + " onError()", e);
}
@Override
public void onNext(T t) {
if (logger.isDebugEnabled()) {
logger.debug("RemoteRxEvent: " + params.getName() + " onNext(): " + t);
}
}
});
}