mantis-remote-observable/src/main/java/io/reactivex/mantis/remote/observable/RemoteObservable.java [377:405]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        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()
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



mantis-remote-observable/src/main/java/io/reactivex/mantis/remote/observable/RemoteObservable.java [501:530]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        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()
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -