Java 类io.reactivex.netty.channel.ConnectionHandler 实例源码

public UdpServer<DatagramPacket, DatagramPacket> createServer() {
    UdpServer<DatagramPacket, DatagramPacket> server = RxNetty.createUdpServer(port, new ConnectionHandler<DatagramPacket, DatagramPacket>() {
        @Override
        public Observable<Void> handle(final ObservableConnection<DatagramPacket, DatagramPacket> newConnection) {
            return newConnection.getInput().flatMap(new Func1<DatagramPacket, Observable<Void>>() {
                @Override
                public Observable<Void> call(final DatagramPacket received) {
                    return Observable.interval(delay, TimeUnit.MILLISECONDS).take(1).flatMap(new Func1<Long, Observable<Void>>() {
                        @Override
                        public Observable<Void> call(Long aLong) {
                            InetSocketAddress sender = received.sender();
                            System.out.println("Received datagram. Sender: " + sender);
                            ByteBuf data = newConnection.getChannel().alloc().buffer(WELCOME_MSG_BYTES.length);
                            data.writeBytes(WELCOME_MSG_BYTES);
                            return newConnection.writeAndFlush(new DatagramPacket(data, sender));
                        }
                    });
                }
            });
        }
    });
    System.out.println("UDP hello server started at port: " + port);
    return server;
}

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();
}

public RxServer(ServerBootstrap bootstrap, int port, final PipelineConfigurator<I, O> pipelineConfigurator,
                final ConnectionHandler<I, O> connectionHandler) {
    if (null == bootstrap) {
        throw new NullPointerException("Bootstrap can not be null.");
    }
    this.bootstrap = bootstrap;
    this.port = port;
    this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            RxRequiredConfigurator<I, O> requiredConfigurator = new RxRequiredConfigurator<I, O>(connectionHandler,
                                                                                                 errorHandler);
            PipelineConfigurator<I, O> configurator;
            if (null == pipelineConfigurator) {
                configurator = requiredConfigurator;
            } else {
                configurator = new PipelineConfiguratorComposite<I, O>(pipelineConfigurator, requiredConfigurator);
            }
            configurator.configureNewPipeline(ch.pipeline());
        }
    });

    serverStateRef = new AtomicReference<ServerState>(ServerState.Created);
}

@Inject
public void setInjector(Injector injector) {
    ServerConfig config = injector.getInstance(serverConfigKey);

    ConnectionHandler<I, O> connectionHandler = injector.getInstance(connectionHandlerKey);

    ServerBuilder<I, O> builder = RxNetty.newTcpServerBuilder(config.getPort(), connectionHandler);

    if (injector.getExistingBinding(pipelineConfiguratorKey) != null) {
        builder.appendPipelineConfigurator(injector.getInstance(pipelineConfiguratorKey));
    }

    if (injector.getExistingBinding(metricEventsListenerFactoryKey) != null) {
        builder.withMetricEventsListenerFactory(injector.getInstance(metricEventsListenerFactoryKey));
    }

    server = builder.build().start();
    logger.info("Starting server {} on port {}...", nameAnnotation.value(), server.getServerPort());
}

@Inject
@SuppressWarnings("unchecked")
public void setInjector(Injector injector) {
    KaryonWebSocketsModule.WebSocketsServerConfig config = (KaryonWebSocketsModule.WebSocketsServerConfig) injector.getInstance(serverConfigKey);

    ConnectionHandler<I, O> connectionHandler = injector.getInstance(connectionHandlerKey);

    WebSocketServerBuilder<I, O> builder = RxNetty.newWebSocketServerBuilder(config.getPort(), connectionHandler)
            .withMessageAggregator(config.isMessageAggregator());

    if (injector.getExistingBinding(pipelineConfiguratorKey) != null) {
        builder.appendPipelineConfigurator(injector.getInstance(pipelineConfiguratorKey));
    }

    if (injector.getExistingBinding(metricEventsListenerFactoryKey) != null) {
        builder.withMetricEventsListenerFactory(injector.getInstance(metricEventsListenerFactoryKey));
    }

    server = builder.build().start();
    logger.info("Starting WebSockets server {} on port {}...", nameAnnotation.value(), server.getServerPort());
}

public void start() {
    int port;
    try {
        port = choosePort();
    } catch (SocketException e) {
        throw new RuntimeException("Error choosing point", e);
    }

    server = RxNetty.createUdpServer(port, new ConnectionHandler<DatagramPacket, DatagramPacket>() {
        @Override
        public Observable<Void> handle(final ObservableConnection<DatagramPacket, DatagramPacket> newConnection) {
            return newConnection.getInput().flatMap(new Func1<DatagramPacket, Observable<Void>>() {
                @Override
                public Observable<Void> call(final DatagramPacket received) {
                    return Observable.interval(timeout, TimeUnit.MILLISECONDS).take(1).flatMap(new Func1<Long, Observable<Void>>() {
                        @Override
                        public Observable<Void> call(Long aLong) {
                            InetSocketAddress sender = received.sender();
                            LOG.info("Received datagram. Sender: " + sender);
                            ByteBuf data = newConnection.getChannel().alloc().buffer(WELCOME_MSG_BYTES.length);
                            data.writeBytes(WELCOME_MSG_BYTES);
                            return newConnection.writeAndFlush(new DatagramPacket(data, sender));
                        }
                    });
                }
            });
        }
    });

    server.start();

    LOG.info("UDP hello server started at port: " + port);
}

public static void main(String[] args) {
    RxNetty.createTcpServer(8181, PipelineConfigurators.textOnlyConfigurator(),
                            new ConnectionHandler<String, String>() {
                                @Override
                                public Observable<Void> handle(ObservableConnection<String, String> newConnection) {
                                    return startEventStream(newConnection);
                                }
                            }).startAndWait();
}

protected AbstractServerBuilder(int port, ConnectionHandler<I, O> connectionHandler, ServerBootstrap bootstrap) {
    if (null == connectionHandler) {
        throw new IllegalArgumentException("Connection handler can not be null");
    }
    if (null == bootstrap) {
        throw new IllegalArgumentException("Server bootstrap can not be null");
    }
    serverBootstrap = bootstrap;
    this.port = port;
    this.connectionHandler = connectionHandler;
    serverChannelClass = NioServerSocketChannel.class;
}

@Before
public void setUp() throws Exception {
    server = RxNetty.createTcpServer(PORT, new ConnectionHandler<ByteBuf, ByteBuf>() {
        @Override
        public Observable<Void> handle(ObservableConnection<ByteBuf, ByteBuf> newConnection) {
            return Observable.error(new IllegalStateException("I always throw an error."));
        }
    });
}

public TcpRxServerProvider(String name, Class<I> iType, Class<O> oType) {
    nameAnnotation = Names.named(name);

    connectionHandlerKey = keyFor(ConnectionHandler.class, iType, oType, nameAnnotation);
    pipelineConfiguratorKey = Key.get(PipelineConfigurator.class, nameAnnotation);
    metricEventsListenerFactoryKey = Key.get(MetricEventsListenerFactory.class, nameAnnotation);
    serverConfigKey = Key.get(ServerConfig.class, nameAnnotation);
}

public WebSocketsRxServerProvider(String name, Class<I> iType, Class<O> oType) {
    nameAnnotation = Names.named(name);

    connectionHandlerKey = keyFor(ConnectionHandler.class, iType, oType, nameAnnotation);
    pipelineConfiguratorKey = Key.get(PipelineConfigurator.class, nameAnnotation);
    metricEventsListenerFactoryKey = Key.get(MetricEventsListenerFactory.class, nameAnnotation);
    serverConfigKey = Key.get(AbstractServerModule.ServerConfig.class, nameAnnotation);
}

public static void main(final String[] args) {
    RxServer<TextWebSocketFrame, TextWebSocketFrame> webSocketServer = RxNetty.newWebSocketServerBuilder(
            8888,
            new ConnectionHandler<TextWebSocketFrame, TextWebSocketFrame>() {
                @Override
                public Observable<Void> handle(final ObservableConnection<TextWebSocketFrame, TextWebSocketFrame> connection) {
                    return connection.getInput().flatMap(new Func1<WebSocketFrame, Observable<Void>>() {
                        @Override
                        public Observable<Void> call(WebSocketFrame wsFrame) {
                            TextWebSocketFrame textFrame = (TextWebSocketFrame) wsFrame;
                            System.out.println("Got message: " + textFrame.text());
                            return connection.writeAndFlush(new TextWebSocketFrame(textFrame.text().toUpperCase()));
                        }
                    });
                }
            }
    ).build();
    Karyon.forWebSocketServer(
            webSocketServer,
            new KaryonBootstrapModule(),
            new ArchaiusBootstrapModule("websocket-echo-server"),
            // KaryonEurekaModule.asBootstrapModule(), /* Uncomment if you need eureka */
            Karyon.toBootstrapModule(KaryonWebAdminModule.class),
            ShutdownModule.asBootstrapModule(),
            KaryonServoModule.asBootstrapModule())
            .startAndWaitTillShutdown();
}

public ConnectionLifecycleHandler(ConnectionHandler<I, O> connectionHandler, ObservableAdapter observableAdapter,
                                  ErrorHandler errorHandler) {
    this.connectionHandler = connectionHandler;
    this.observableAdapter = observableAdapter;
    this.errorHandler = null == errorHandler ? new DefaultErrorHandler() : errorHandler;
}

public ConnectionLifecycleHandler(ConnectionHandler<I, O> connectionHandler, ObservableAdapter observableAdapter) {
    this(connectionHandler, observableAdapter, null);
}

public RxRequiredConfigurator(final ConnectionHandler<I, O> connectionHandler) {
    this(connectionHandler, null);
}

public RxRequiredConfigurator(final ConnectionHandler<I, O> connectionHandler, ErrorHandler errorHandler) {
    this.connectionHandler = connectionHandler;
    this.errorHandler = errorHandler;
}

public ServerBuilder(int port, ConnectionHandler<I, O> connectionHandler) {
    super(port, connectionHandler);
}

public ServerBuilder(int port, ConnectionHandler<I, O> connectionHandler, ServerBootstrap bootstrap) {
    super(port, connectionHandler, bootstrap);
}

public RxServer(ServerBootstrap bootstrap, int port, final ConnectionHandler<I, O> connectionHandler) {
    this(bootstrap, port, null, connectionHandler);
}

protected AbstractServerBuilder(int port, ConnectionHandler<I, O> connectionHandler) {
    this(port, connectionHandler, new ServerBootstrap());
}

public static <I, O> RxServer<I, O> createTcpServer(final int port, PipelineConfigurator<I, O> pipelineConfigurator,
                                                    ConnectionHandler<I, O> connectionHandler) {
    return new ServerBuilder<I, O>(port, connectionHandler).pipelineConfigurator(pipelineConfigurator).build();
}

public static RxServer<ByteBuf, ByteBuf> createTcpServer(final int port,
                                                         ConnectionHandler<ByteBuf, ByteBuf> connectionHandler) {
    return new ServerBuilder<ByteBuf, ByteBuf>(port, connectionHandler).build();
}

protected KaryonTcpModule(String moduleName, Class<I> iType, Class<O> oType) {
    super(moduleName, iType, oType);
    connectionHandlerKey = keyFor(ConnectionHandler.class, iType, oType, nameAnnotation);
    serverKey = keyFor(RxServer.class, iType, oType, nameAnnotation);
}

public LinkedBindingBuilder<ConnectionHandler<I, O>> bindConnectionHandler() {
    return bind(connectionHandlerKey);
}

protected KaryonWebSocketsModule(String moduleName, Class<I> iType, Class<O> oType) {
    super(moduleName, iType, oType);
    connectionHandlerKey = keyFor(ConnectionHandler.class, iType, oType, nameAnnotation);
    serverKey = keyFor(RxServer.class, iType, oType, nameAnnotation);
}

public LinkedBindingBuilder<ConnectionHandler<I, O>> bindConnectionHandler() {
    return bind(connectionHandlerKey);
}

/**
 * Creates a new {@link KaryonServer} that has a single TCP server instance which delegates all connection
 * handling to {@link ConnectionHandler}.
 * The {@link RxServer} is created using {@link RxNetty#newTcpServerBuilder(int, ConnectionHandler)}
 *
 * @param port Port for the server.
 * @param handler Connection Handler
 * @param modules Additional modules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forTcpConnectionHandler(int port, ConnectionHandler<ByteBuf, ByteBuf> handler,
                                                   Module... modules) {
    return forTcpConnectionHandler(port, handler, toBootstrapModule(modules));
}

/**
 * Creates a new {@link KaryonServer} that has a single TCP server instance which delegates all connection
 * handling to {@link ConnectionHandler}.
 * The {@link RxServer} is created using {@link RxNetty#newTcpServerBuilder(int, ConnectionHandler)}
 *
 * @param port Port for the server.
 * @param handler Connection Handler
 * @param bootstrapModules Additional bootstrapModules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forTcpConnectionHandler(int port, ConnectionHandler<ByteBuf, ByteBuf> handler,
                                                   BootstrapModule... bootstrapModules) {
    RxServer<ByteBuf, ByteBuf> server = RxNetty.newTcpServerBuilder(port, handler).build();
    return new RxNettyServerBackedServer(server, bootstrapModules);
}

/**
 * Creates a new {@link KaryonServer} that has a single UDP server instance which delegates all connection
 * handling to {@link ConnectionHandler}.
 * The {@link RxServer} is created using {@link RxNetty#newUdpServerBuilder(int, ConnectionHandler)}
 *
 * @param port Port for the server.
 * @param handler Connection Handler
 * @param modules Additional modules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forUdpConnectionHandler(int port, ConnectionHandler<ByteBuf, ByteBuf> handler,
                                                   Module... modules) {
    return forUdpConnectionHandler(port, handler, toBootstrapModule(modules));
}

/**
 * Creates a new {@link KaryonServer} that has a single UDP server instance which delegates all connection
 * handling to {@link ConnectionHandler}.
 * The {@link RxServer} is created using {@link RxNetty#newUdpServerBuilder(int, ConnectionHandler)}
 *
 * @param port Port for the server.
 * @param handler Connection Handler
 * @param bootstrapModules Additional bootstrapModules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forUdpConnectionHandler(int port, ConnectionHandler<ByteBuf, ByteBuf> handler,
                                                   BootstrapModule... bootstrapModules) {
    UdpServer<ByteBuf, ByteBuf> server = RxNetty.newUdpServerBuilder(port, handler).build();
    return new RxNettyServerBackedServer(server, bootstrapModules);
}