Java 类io.reactivex.netty.protocol.http.server.HttpServerBuilder 实例源码

private HttpServer<ByteBuf, ByteBuf> createServer(final int listenPort) {
    HttpServer<ByteBuf, ByteBuf> server = new HttpServerBuilder<ByteBuf, ByteBuf>(listenPort, (
            request, response) -> request
            .getContent()
            .compose(new RequestTransformer())
            .filter(bidreq -> bidreq != null)
            .compose(new ExternalRequestTransformer())
            .compose(new ExternalRequestExecutor(client))
            .flatMap(
                    resp -> {
                        response.setStatus(resp == null ? HttpResponseStatus.NO_CONTENT
                                : HttpResponseStatus.OK);
                        response.writeString(resp == null ? "" : resp.toString());
                        return response.close();
                    })).build();
    LOG.info("Gateway started...");
    return server;
}

@BeforeClass
public static void init() {
    port = new Random().nextInt(1000) + 4000;
    HttpServerBuilder<ByteBuf, ByteBuf> builder 
        = new HttpServerBuilder<ByteBuf, ByteBuf>(new ServerBootstrap().group(new NioEventLoopGroup(10, new RxServerThreadFactory())), port, new RequestProcessor());
    server = builder.build();
    // server = RxNetty.createHttpServer(port, new RequestProcessor());

    server.start();
}

@SuppressWarnings("rawtypes")
@Inject
public void setInjector(Injector injector) {
    HttpServerConfig config = (HttpServerConfig) injector.getInstance(serverConfigKey);

    RequestHandler router = injector.getInstance(routerKey);

    GovernatorHttpInterceptorSupport<I, O> interceptorSupport = injector.getInstance(interceptorSupportKey);
    interceptorSupport.finish(injector);
    HttpRequestHandler<I, O> httpRequestHandler = new HttpRequestHandler<I, O>(router, interceptorSupport);

    HttpServerBuilder<I, O> builder = KaryonTransport.newHttpServerBuilder(config.getPort(), httpRequestHandler);

    if (config.requiresThreadPool()) {
        builder.withRequestProcessingThreads(config.getThreadPoolSize());
    }

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

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

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

public static HttpServer<ByteBuf, ByteBuf> createHttpServer(int port, RequestHandler<ByteBuf, ByteBuf> requestHandler) {
    return new HttpServerBuilder<ByteBuf, ByteBuf>(port, requestHandler).build();
}

public static <I, O> HttpServer<I, O> createHttpServer(int port,
                                                       RequestHandler<I, O> requestHandler,
                                                       PipelineConfigurator<HttpServerRequest<I>, HttpServerResponse<O>> configurator) {
    return new HttpServerBuilder<I, O>(port, requestHandler).pipelineConfigurator(configurator).build();
}

public static <I, O> HttpServerBuilder<I, O> newHttpServerBuilder(int port, RequestHandler<I, O> router) {
    return RxContexts.newHttpServerBuilder(port, new HttpRequestHandler<I, O>(router), RxContexts.DEFAULT_CORRELATOR);
}

public static <I, O> HttpServerBuilder<I, O> newHttpServerBuilder(int port, HttpRequestHandler<I, O> requestHandler) {
    return RxContexts.newHttpServerBuilder(port, requestHandler, RxContexts.DEFAULT_CORRELATOR);
}