TcpEchoServer.java 文件源码

public static void main(final String[] args) {
    final int port = 8181;
    RxNetty.createTcpServer(port, PipelineConfigurators.textOnlyConfigurator(),
                            new ConnectionHandler<String, String>() {
                                @Override
                                public Observable<Void> handle(
                                        final ObservableConnection<String, String> connection) {
                                    System.out.println("New client connection established.");
                                    connection.writeAndFlush("Welcome! \n\n");
                                    return connection.getInput().flatMap(new Func1<String, Observable<Void>>() {
                                        @Override
                                        public Observable<Void> call(String msg) {
                                            System.out.println("onNext: " + msg);
                                            msg = msg.trim();
                                            if (!msg.isEmpty()) {
                                                return connection.writeAndFlush("echo => " + msg + '\n');
                                            } else {
                                                return COMPLETED_OBSERVABLE;
                                            }
                                        }
                                    });
                                }
                            }).startAndWait();
}