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

项目：MarketData 文件：StaticServer.java

public HttpServer<ByteBuf, ByteBuf> createServer() {
    try {
        final String ideBasePath = new File(".").getCanonicalPath();
        final File rootDirectory;
        if (ideBasePath.contains(rootDir)) {
            // for eclipse
            rootDirectory = new File(".");
        } else {
            // for intellij
            rootDirectory = new File(rootDir);
        }
        return RxNetty.createHttpServer(port,
                RequestHandlerWithErrorMapper.from(
                        new LocalDirectoryRequestHandler(rootDirectory),
                        new FileErrorResponseMapper()));
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

项目：RxNettyProxy 文件：ProxyServer.java

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

项目：hawkular-apm 文件：NettyNoResponseHttpITest.java

@Override
public void init() {
    server = HttpServer.newServer()
            .enableWireLogging(LogLevel.DEBUG)
            .start((req, resp) -> {
                if (req.getHeader(Constants.HAWKULAR_APM_TRACEID) == null) {
                    return resp.setStatus(HttpResponseStatus.BAD_REQUEST);
                }
                if (req.getHttpMethod() == HttpMethod.POST
                        || req.getHttpMethod() == HttpMethod.PUT) {
                    req.getContent().subscribe(bb -> System.out.println("DATA = " + bb.toString()));
                }
                resp.setStatus(HttpResponseStatus.OK);
                return resp;
            }
            );

    super.init();
}

项目：hawkular-apm 文件：NettyHttpITest.java

@Override
public void init() {
    server = HttpServer.newServer()
            .enableWireLogging(LogLevel.DEBUG)
            .start((req, resp) -> {
                if (req.getHeader(Constants.HAWKULAR_APM_TRACEID) == null) {
                    return resp.setStatus(HttpResponseStatus.BAD_REQUEST);
                }
                if (req.getHttpMethod() == HttpMethod.POST
                        || req.getHttpMethod() == HttpMethod.PUT) {
                    req.getContent().subscribe(bb -> System.out.println("DATA = " + bb.toString()));
                }
                return resp.writeString(Observable.just(HELLO_WORLD));
            });
    super.init();
}

项目：mesos-rxjava 文件：MesosClientIntegrationTest.java

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mesos5xxResponse() throws Throwable {
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because subscribe will 500");
    } catch (Mesos5xxException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(500);
    } finally {
        server.shutdown();
    }
}

项目：mesos-rxjava 文件：MesosClientIntegrationTest.java

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mismatchContentType() throws Throwable {
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "application/json");
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because of content type mismatch");
    } catch (MesosException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(200);
        assertThat(ctx.getMessage()).isEqualTo("Response had Content-Type \"application/json\" expected \"text/plain;charset=utf-8\"");
    } finally {
        server.shutdown();
    }
}

项目：triathlon 文件：RxTestServer.java

public HttpServer<ByteBuf, ByteBuf> createServer() {
    HttpServer<ByteBuf, ByteBuf> server = RxNetty.newHttpServerBuilder(port, new RequestHandler<ByteBuf, ByteBuf>() {
        @Override
        public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {
            if (request.getPath().contains("/v2/apps")) {
                if (request.getHttpMethod().equals(HttpMethod.POST)) {
                    return handleTest(request, response);
                }
            }
            response.setStatus(HttpResponseStatus.NOT_FOUND);
            return response.close();
        }
    }).pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpServerConfigurator()).enableWireLogging(LogLevel.ERROR).build();

    System.out.println("RxTetstServer server started...");
    return server;
}

项目：Artemis 文件：KaryonPluginModule.java

@Provides
public HttpServer providesKaryonTransport() {
    SimpleUriRouter simpleUriRouter = new SimpleUriRouter();
    simpleUriRouter.addUri("/foo", new RequestHandler() {
        @Override
        public Observable<Void> handle(HttpServerRequest request, HttpServerResponse response) {
            response.writeAndFlush("Hello World", StringTransformer.DEFAULT_INSTANCE);
            response.setStatus(HttpResponseStatus.OK);
            return Observable.empty();
        }

        @Override
        public Observable<Void> handle(Object input, Object output) {
            return Observable.empty();
        }
    });
    return KaryonTransport.newHttpServer(8888, simpleUriRouter);
}

项目：ReactiveLab 文件：ClientServer.java

public static void main(String[] args) {

        /**
         * Start our HTTP server.
         */
        HttpServer<ByteBuf, ByteBuf> server = startServer(8088);

        /**
         * Submit the request.
         */
        createRequest("localhost", server.getServerPort())
                /* Block till you get the response. In a real world application, one should not be blocked but chained
                 * into a response to the caller. */
                .toBlocking()
                /**
                 * Print each content of the response.
                 */
                .forEach(System.out::println);
    }

项目：ReactiveLab 文件：ClientServer.java

public static HttpServer<ByteBuf, ByteBuf> startServer(int port) {

        /**
         * Creates an HTTP server which returns "Hello World!" responses.
         */
        return RxNetty.createHttpServer(port,
                                        /*
                                         * HTTP Request handler for RxNetty where you control what you write as the
                                         * response for each and every request the server receives.
                                         */
                                        (request, response) -> {
                                            /**
                                             * In a real server, you would be writing different responses based on the
                                             * URI of the request.
                                             * This example just returns a "Hello World!!" string unconditionally.
                                             */
                                            return response.writeStringAndFlush("Hello World!!");
                                        })
                      .start();
    }

项目：ribbon 文件：RxMovieServer.java

public HttpServer<ByteBuf, ByteBuf> createServer() {
    HttpServer<ByteBuf, ByteBuf> server = RxNetty.newHttpServerBuilder(port, new RequestHandler<ByteBuf, ByteBuf>() {
        @Override
        public Observable<Void> handle(HttpServerRequest<ByteBuf> request, final HttpServerResponse<ByteBuf> response) {
            if (request.getPath().contains("/users")) {
                if (request.getHttpMethod().equals(HttpMethod.GET)) {
                    return handleRecommendationsByUserId(request, response);
                } else {
                    return handleUpdateRecommendationsForUser(request, response);
                }
            }
            if (request.getPath().contains("/recommendations")) {
                return handleRecommendationsBy(request, response);
            }
            if (request.getPath().contains("/movies")) {
                return handleRegisterMovie(request, response);
            }
            response.setStatus(HttpResponseStatus.NOT_FOUND);
            return response.close();
        }
    }).pipelineConfigurator(PipelineConfigurators.<ByteBuf, ByteBuf>httpServerConfigurator()).enableWireLogging(LogLevel.ERROR).build();

    System.out.println("RxMovie server started...");
    return server;
}

项目：tusRx 文件：TestServer.java

public TestServer() {
    Options options = getOptions();
    RequestHandler<ByteBuf, ByteBuf> requestHandler = new TusRxRequestHandler(
            options, new RxNettyRequestHandlerFactory(options, new UploaderPool(options.getRootDir())));
    server = HttpServer.newServer();

    CompletableFuture<Void> serverFuture = CompletableFuture.runAsync(() -> server.start(requestHandler));
    serverFuture.thenAccept(voyd -> server.awaitShutdown());
    serverFuture.join();
}

项目：tusRx 文件：TestServer.java

public static void main(String args[]) {
    Options options = getOptions();
    RequestHandler<ByteBuf, ByteBuf> requestHandler = new TusRxRequestHandler(
            options, new RxNettyRequestHandlerFactory(options, new UploaderPool(options.getRootDir())));
    HttpServer<ByteBuf, ByteBuf> server = HttpServer.newServer(8080).start(requestHandler);

    server.awaitShutdown();
}

项目：MarketData 文件：RxNettyEventBroadcaster.java

public HttpServer<ByteBuf, ServerSentEvent> createServer() {
    if (flaky) {
        events = SubscriptionLimiter
                    .limitSubscriptions(1,initializeEventStream());
    } else {
        events  = initializeEventStream();
    }
    return super.createServer();
}

项目：MarketData 文件：RxNettyRequestReplyServer.java

public HttpServer<ByteBuf, ByteBuf> createServer() {
    return RxNetty.createHttpServer(port, (request, response) -> {

        HttpRequest httpRequest = new HttpRequest(request.getQueryParameters());
        String content = getResponseContent(httpRequest);
        return response.writeStringAndFlush(content);
    });
}

项目：MarketData 文件：RxNettyEventServer.java

public HttpServer<ByteBuf, ServerSentEvent> createServer() {
    HttpServer<ByteBuf, ServerSentEvent> server = RxNetty.createHttpServer(port,
            (request, response) -> {
                response.getHeaders().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
                response.getHeaders().set(CACHE_CONTROL, "no-cache");
                response.getHeaders().set(CONNECTION, "keep-alive");
                response.getHeaders().set(CONTENT_TYPE, "text/event-stream");
                return getIntervalObservable(request, response);
            }, PipelineConfigurators.<ByteBuf>serveSseConfigurator());
    System.out.println("HTTP Server Sent Events server started...");
    return server;
}

项目：rxnetty-nio-test 文件：RxJavaNioTest.java

@Test
public void nioTest() throws Exception {
  HttpServer<ByteBuf, ByteBuf> server = getServer();

  TestSubscriber<String> ts = new TestSubscriber<>();

  long start = System.currentTimeMillis();

  // we use 10 since the default rxnetty thread pool size is 8
  // you could also shrink the pool down for the same effect
  // but I couldn't be bothered finding the settings
  Observable.range(1, 10)
    // flatMap runs async Observables concurrently
    .flatMap(i ->
      HttpClient.newClient(server.getServerAddress())
        .createGet("/" + i)
        .flatMap(response ->
          response.getContent()
            .map(bytes ->
              bytes.toString(Charset.defaultCharset()) + " " +
                "[response received on " + Thread.currentThread().getName() +
                " at " + (System.currentTimeMillis() - start) + "]"
            )
        )
    )
    .doOnNext(System.out::println)
    .subscribe(ts);

  ts.awaitTerminalEvent();

  server.shutdown();
}

项目：rxnetty-nio-test 文件：RxJavaNioTest.java

private HttpServer<ByteBuf, ByteBuf> getServer() {
  return HttpServer.newServer()
      .start((request, response) -> {
        String requestThread = Thread.currentThread().getName();
        // we need tp flush on each so that the response will 'stream'
        return response.writeStringAndFlushOnEach(
          // one numbered reply per second
          Observable.interval(1, 1, TimeUnit.SECONDS, Schedulers.io())
            .take(4)
            .map(i ->
              getResponseValue(request, requestThread, i)
            )
        );
      });
}

项目：netty-cookbook 文件：NettyRxJavaServer.java

public static void main(String... args) throws InterruptedException {
    HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(8080, 
            (HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) -> {
        System.out.println("Server => Request: " + request.getPath());            
        try {
            if ("/error".equals(request.getPath())) {
                throw new RuntimeException("forced error");
            }
            response.setStatus(HttpResponseStatus.OK);
            response.writeString("Path Requested =>: " + request.getPath() + '\n');
            return response.close();
        } catch (Throwable e) {
            System.err.println("Server => Error [" + request.getPath() + "] => " + e);
            response.setStatus(HttpResponseStatus.BAD_REQUEST);
            response.writeString("Error 500: Bad Request\n");
            return response.close();
        }
    });

    server.startAndWait();

    RxNetty.createHttpGet("http://localhost:8080/")
           .flatMap(response -> response.getContent())
           .map(data -> "Client => " + data.toString(Charset.defaultCharset()))
           .toBlocking().forEach(System.out::println);

    RxNetty.createHttpGet("http://localhost:8080/error")
           .flatMap(response -> response.getContent())
           .map(data -> "Client => " + data.toString(Charset.defaultCharset()))
           .toBlocking().forEach(System.out::println);

    RxNetty.createHttpGet("http://localhost:8080/data")
           .flatMap(response -> response.getContent())
           .map(data -> "Client => " + data.toString(Charset.defaultCharset()))
           .toBlocking().forEach(System.out::println);

    //server.shutdown();
}

项目：mesos-rxjava 文件：MesosClientIntegrationTest.java

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mesos4xxResponse() throws Throwable {
    final String errorMessage = "Error message that should come from the server";
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.BAD_REQUEST);
        final byte[] msgBytes = errorMessage.getBytes(StandardCharsets.UTF_8);
        response.getHeaders().setHeader("Content-Length", msgBytes.length);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        response.writeBytes(msgBytes);
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because subscribe will 400");
    } catch (Mesos4xxException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(400);
        assertThat(ctx.getMessage()).isEqualTo(errorMessage);
    } finally {
        server.shutdown();
    }
}

项目：mesos-rxjava 文件：MesosClientBackpressureIntegrationTest.java

@Test
@Ignore
public void testBurstyObservable_missingBackpressureException() throws Throwable {
    final String subscribedMessage = "{\"type\": \"SUBSCRIBED\",\"subscribed\": {\"framework_id\": {\"value\":\"12220-3440-12532-2345\"},\"heartbeat_interval_seconds\":15.0}";
    final String heartbeatMessage = "{\"type\":\"HEARTBEAT\"}";
    final byte[] hmsg = heartbeatMessage.getBytes(StandardCharsets.UTF_8);
    final byte[] hbytes = String.format("%d\n", heartbeatMessage.getBytes().length).getBytes(StandardCharsets.UTF_8);

    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        writeRecordIOMessage(response, subscribedMessage);
        for (int i = 0; i < 20000; i++) {
            response.writeBytes(hbytes);
            response.writeBytes(hmsg);
        }
        return response.flush();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClientForStreaming(uri).build();

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up due to backpressure");
    } catch (MissingBackpressureException e) {
        // expected
        e.printStackTrace();
        assertThat(e.getMessage()).isNullOrEmpty();
    } finally {
        server.shutdown();
    }
}

项目：mesos-rxjava 文件：MesosClientBackpressureIntegrationTest.java

@Test
public void testBurstyObservable_unboundedBufferSucceeds() throws Throwable {
    msgNo = 0;
    final int numMessages = 20000;
    final String subscribedMessage = "{\"type\": \"SUBSCRIBED\",\"subscribed\": {\"framework_id\": {\"value\":\"12220-3440-12532-2345\"},\"heartbeat_interval_seconds\":15.0}";
    final String heartbeatMessage = "{\"type\":\"HEARTBEAT\"}";
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        writeRecordIOMessage(response, subscribedMessage);
        for (int i = 0; i < numMessages; i++) {
            writeRecordIOMessage(response, heartbeatMessage);
        }
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClientForStreaming(uri)
            .onBackpressureBuffer()
            .build();

    try {
        client.openStream().await();
    } finally {
        // 20000 heartbeats PLUS 1 subscribe
        assertEquals("All heartbeats received (plus the subscribed)", 1 + numMessages, msgNo);
        server.shutdown();
    }
}

项目：reactivesocket-websocket-rxnetty 文件：ClientServerTest.java

@BeforeClass
public static void setup() {
    ReactiveSocketWebSocketServer serverHandler = ReactiveSocketWebSocketServer.create(
            requestResponsePayload -> {
                String requestResponse = byteToString(requestResponsePayload.getData());
                if (requestResponse.startsWith("h")) {
                    return Single.just(utf8EncodedPayloadData(requestResponse + " world"));
                } else if ("test".equals(requestResponse)) {
                    return Single.just(utf8EncodedPayloadData("test response"));
                } else {
                    return Single.error(new RuntimeException("Not Found"));
                }
            } ,
            requestStreamPayload -> {
                String requestStream = byteToString(requestStreamPayload.getData());
                return Observable.just(requestStream, "world").map(n -> utf8EncodedPayloadData(n));
            } , null, null, null);

    server = HttpServer.newServer()
            // .clientChannelOption(ChannelOption.AUTO_READ, true)
            // .enableWireLogging(LogLevel.ERROR)
            .start((req, resp) -> {
                return resp.acceptWebSocketUpgrade(serverHandler::acceptWebsocket);
            });

    client = HttpClient.newClient("localhost", server.getServerPort()).enableWireLogging(LogLevel.ERROR)
            .createGet("/rs")
            .requestWebSocketUpgrade()
            .flatMap(WebSocketResponse::getWebSocketConnection)
            .map(ReactiveSocketWebSocketClient::create)
            .toBlocking().single();

    client.connect()
            .subscribe(v -> {
            } , t -> t.printStackTrace());
}

项目：reactivesocket-websocket-rxnetty 文件：ReactiveSocketWebSocketServerTest.java

@Test
public void test() {
    // create protocol with handlers
    ReactiveSocketWebSocketServer handler = ReactiveSocketWebSocketServer.create(
            requestResponsePayload -> {
                String requestResponse = byteToString(requestResponsePayload.getData()); 
                return Single.just(utf8EncodedPayloadData("hello" + requestResponse));
            } ,
            requestStreamPayload -> {
                String requestStream = byteToString(requestStreamPayload.getData());
                return just("a_" + requestStream, "b_" + requestStream).map(n -> utf8EncodedPayloadData(n));
            } , null, null, null);

    // start server with protocol
    HttpServer<ByteBuf, ByteBuf> server = HttpServer.newServer();
    int port = server.getServerPort();
    server.start((request, response) -> {
        return response.acceptWebSocketUpgrade(handler::acceptWebsocket);
    });

    // TODO send actual requests
    HttpClient.newClient("localhost", server.getServerPort())
            .createGet("/")
            .requestWebSocketUpgrade();

    server.shutdown();
}

项目：ocelli 文件：HttpExampleUtils.java

protected static SocketAddress startServer(long latencyMillis) {
    return HttpServer.newServer()
                     .start((request, response) -> {
                         return Observable.timer(latencyMillis, TimeUnit.MILLISECONDS)
                                          .flatMap(aTick -> response.addHeader("X-Instance",
                                                                               response.unsafeNettyChannel()
                                                                                       .localAddress())
                                                                    .setStatus(HttpResponseStatus.OK));
                     })
                     .getServerAddress();
}

项目：ocelli 文件：HttpExampleUtils.java

protected static SocketAddress startServer(HttpResponseStatus cannedStatus) {
    return HttpServer.newServer()
                     .start((request, response) -> {
                         return response.addHeader("X-Instance", response.unsafeNettyChannel().localAddress())
                                        .setStatus(cannedStatus);
                     })
                     .getServerAddress();
}

项目：spring-cloud-netflix 文件：TurbineStreamConfiguration.java

@Bean
@SuppressWarnings("deprecation")
public HttpServer<ByteBuf, ServerSentEvent> aggregatorServer() {
    // multicast so multiple concurrent subscribers get the same stream
    Observable<Map<String, Object>> publishedStreams = StreamAggregator
            .aggregateGroupedStreams(hystrixSubject().groupBy(
                    data -> InstanceKey.create((String) data.get("instanceId"))))
            .doOnUnsubscribe(() -> log.info("Unsubscribing aggregation."))
            .doOnSubscribe(() -> log.info("Starting aggregation")).flatMap(o -> o)
            .publish().refCount();
    Observable<Map<String, Object>> ping = Observable.timer(1, 10, TimeUnit.SECONDS)
            .map(count -> Collections.singletonMap("type", (Object) "Ping")).publish()
            .refCount();
    Observable<Map<String, Object>> output = Observable.merge(publishedStreams, ping);

    this.turbinePort = this.properties.getPort();

    if (this.turbinePort <= 0) {
        this.turbinePort = SocketUtils.findAvailableTcpPort(40000);
    }

    HttpServer<ByteBuf, ServerSentEvent> httpServer = RxNetty
            .createHttpServer(this.turbinePort, (request, response) -> {
                log.info("SSE Request Received");
                response.getHeaders().setHeader("Content-Type", "text/event-stream");
                return output.doOnUnsubscribe(
                        () -> log.info("Unsubscribing RxNetty server connection"))
                        .flatMap(data -> response.writeAndFlush(new ServerSentEvent(
                                null,
                                Unpooled.copiedBuffer("message",
                                        StandardCharsets.UTF_8),
                                Unpooled.copiedBuffer(JsonUtility.mapToJson(data),
                                        StandardCharsets.UTF_8))));
            }, serveSseConfigurator());
    return httpServer;
}

项目：ReactiveLab 文件：AbstractMiddleTierService.java

public HttpServer<ByteBuf, ServerSentEvent> createServer(int port) {
    System.out.println("Start " + getClass().getSimpleName() + " on port: " + port);

    // declare handler chain (wrapped in Hystrix)
    // TODO create a better way of chaining these (related https://github.com/ReactiveX/RxNetty/issues/232 and https://github.com/ReactiveX/RxNetty/issues/202)
    HystrixMetricsStreamHandler<ByteBuf, ServerSentEvent> handlerChain 
      = new HystrixMetricsStreamHandler<>(metrics, "/hystrix.stream", 1000, (request, response) -> {
        try {
            long startTime = System.currentTimeMillis();
            return handleRequest(request, response)
                    .doOnCompleted(() -> System.out.println("Response => " + request.getPath() + " Time => " + (int) (System.currentTimeMillis() - startTime) + "ms"))
                    .doOnCompleted(() -> metrics.getRollingPercentile().addValue((int) (System.currentTimeMillis() - startTime)))
                    .doOnCompleted(() -> metrics.getRollingNumber().add(Metrics.EventType.SUCCESS, 1))
                    .doOnError(t -> metrics.getRollingNumber().add(Metrics.EventType.FAILURE, 1));
        } catch (Throwable e) {
            e.printStackTrace();
            System.err.println("Server => Error [" + request.getPath() + "] => " + e);
            response.setStatus(HttpResponseStatus.BAD_REQUEST);
            return response.writeStringAndFlush("data: Error 500: Bad Request\n" + e.getMessage() + "\n");
        }
    });

    return RxNetty.createHttpServer(port, (request, response) -> {
        // System.out.println("Server => Request: " + request.getPath());
            return handlerChain.handle(request, response);
        }, PipelineConfigurators.<ByteBuf> serveSseConfigurator());
}

项目：karyon 文件：KaryonHttpModule.java

protected KaryonHttpModule(String moduleName, Class<I> iType, Class<O> oType) {
    super(moduleName, iType, oType);

    routerKey = keyFor(RequestHandler.class, iType, oType, nameAnnotation);
    interceptorSupportKey = keyFor(GovernatorHttpInterceptorSupport.class, iType, oType, nameAnnotation);
    httpServerKey = keyFor(HttpServer.class, iType, oType, nameAnnotation);
}

项目：karyon 文件：KaryonHttpModule.java

@Override
protected void configure() {
    configureServer();

    bind(serverConfigKey).toInstance(serverConfigBuilder.build());
    bind(interceptorSupportKey).toInstance(interceptorSupportInstance);

    MapBinder.newMapBinder(binder(), String.class, RxServer.class).addBinding(nameAnnotation.value()).toProvider(
            new HttpRxServerProvider<I, O, HttpServer<I, O>>(nameAnnotation.value(), iType, oType)
    ).asEagerSingleton();
}

项目：karyon 文件：Karyon.java

/**
 * Creates a new {@link KaryonServer} that has a single HTTP server instance which delegates all request
 * handling to {@link RequestHandler}.
 * The {@link HttpServer} is created using {@link KaryonTransport#newHttpServer(int, HttpRequestHandler)}
 *
 * @param port Port for the server.
 * @param handler Request Handler
 * @param bootstrapModules Additional bootstrapModules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forRequestHandler(int port, final RequestHandler<ByteBuf, ByteBuf> handler,
                                             BootstrapModule... bootstrapModules) {
    HttpServer<ByteBuf, ByteBuf> httpServer =
            KaryonTransport.newHttpServer(port, new RequestHandler<ByteBuf, ByteBuf>() {
                @Override
                public Observable<Void> handle(HttpServerRequest<ByteBuf> request,
                                               HttpServerResponse<ByteBuf> response) {
                    return handler.handle(request, response);
                }
            });
    return new RxNettyServerBackedServer(httpServer, bootstrapModules);
}

项目：Artemis 文件：KaryonService.java

@Inject
public KaryonService(HttpServer httpServer) {
    this.httpServer = httpServer;
}

项目：ReactiveLab 文件：ClientServerWithDiscovery.java

public static void main(String[] args) throws Exception {

        final int eurekaReadServerPort = 7005;
        final int eurekaWriteServerPort = 7006;

        /**
         * Starts an embedded eureka server with the defined read and write ports.
         */
        startEurekaServer(eurekaReadServerPort, eurekaWriteServerPort);

        /**
         * Create eureka client with the same read and write ports for the embedded eureka server.
         */
        EurekaClient eurekaClient = createEurekaClient(eurekaReadServerPort, eurekaWriteServerPort);

        /**
         * Reuse {@link ClientServer} example to start an RxNetty server on the passed port.
         */
        HttpServer<ByteBuf, ByteBuf> server = ClientServer.startServer(8089);

        /**
         * Register the server started above with eureka using a unique virtual IP address (VIP).
         * Eureka uses VIPs to group homogeneous instances of a service together, so that they can be used by clients,
         * interchangeably.
         */
        String vipAddress = "mock_server-" + server.getServerPort();
        registerWithEureka(server.getServerPort(), eurekaClient, vipAddress);

        /**
         * Retrieve the instance information of the registered server from eureka.
         * This is to demonstrate how to use eureka to fetch information about any server in your deployment.
         * In order to fetch information from eureka, one MUST know the VIP address of the server before hand.
         */
        InstanceInfo serverInfo = getServerInfo(eurekaClient, vipAddress);

        /**
         * Retrieve IPAddress and port information from the instance information returned from eureka.
         */
        Host host = getServerHostAndPort(serverInfo);

        /**
         * Reuse {@link ClientServer} example to create an HTTP request to the server retrieved from eureka.
         */
        ClientServer.createRequest(host.getIpAddress(), host.getPort())
                    /* Block till you get the response. In a real world application, one should not be blocked but chained
                     * into a response to the caller. */
                    .toBlocking()
                    /**
                     * Print each content of the response.
                     */
                    .forEach(System.out::println);
    }

项目：ReactiveLab 文件：ClientServerWithResiliencePatterns.java

public static void main(String[] args) throws Exception {

        final int eurekaReadServerPort = 7008;
        final int eurekaWriteServerPort = 7010;

        /**
         * Starts an embedded eureka server with the defined read and write ports.
         */
        ClientServerWithDiscovery.startEurekaServer(eurekaReadServerPort, eurekaWriteServerPort);

        /**
         * Create eureka client with the same read and write ports for the embedded eureka server.
         */
        EurekaClient eurekaClient = ClientServerWithDiscovery.createEurekaClient(eurekaReadServerPort,
                                                                                 eurekaWriteServerPort);

        /**
         * Reuse {@link ClientServer} example to start an RxNetty server on the passed port.
         */
        HttpServer<ByteBuf, ByteBuf> server = ClientServer.startServer(8089);

        /**
         * Register the server started above with eureka using a unique virtual IP address (VIP).
         * Eureka uses VIPs to group homogeneous instances of a service together, so that they can be used by clients,
         * interchangeably.
         */
        String vipAddress = "mock_server-" + server.getServerPort();
        ClientServerWithDiscovery.registerWithEureka(server.getServerPort(), eurekaClient, vipAddress);

        /**
         * Using the eureka client, create an Ocelli Host event stream.
         * Ocelli, uses this host stream to know about the available hosts.
         */
        Observable<MembershipEvent<Host>> eurekaHostSource = ClientServerWithLoadBalancer.createEurekaHostStream(
                eurekaClient, vipAddress);

        MyCommand myCommand = new MyCommand(eurekaHostSource);

        /**
         * This executes the request on the client (just as {@link ClientServerWithLoadBalancer} but using hystrix.
         */
        myCommand.toObservable()
                /* Block till you get the response. In a real world application, one should not be blocked but chained
                 * into a response to the caller. */
                .toBlocking()
                /**
                 * Print each content of the response.
                 */
                .forEach(System.out::println);

    }

项目：ReactiveLab 文件：ClientServerWithLoadBalancer.java

public static void main(String[] args) throws Exception {

        final int eurekaReadServerPort = 7007;
        final int eurekaWriteServerPort = 7008;

        /**
         * Starts an embedded eureka server with the defined read and write ports.
         */
        ClientServerWithDiscovery.startEurekaServer(eurekaReadServerPort, eurekaWriteServerPort);

        /**
         * Create eureka client with the same read and write ports for the embedded eureka server.
         */
        EurekaClient eurekaClient = ClientServerWithDiscovery.createEurekaClient(eurekaReadServerPort,
                                                                                 eurekaWriteServerPort);

        /**
         * Reuse {@link ClientServer} example to start an RxNetty server on the passed port.
         */
        HttpServer<ByteBuf, ByteBuf> server = ClientServer.startServer(8089);

        /**
         * Register the server started above with eureka using a unique virtual IP address (VIP).
         * Eureka uses VIPs to group homogeneous instances of a service together, so that they can be used by clients,
         * interchangeably.
         */
        String vipAddress = "mock_server-" + server.getServerPort();
        ClientServerWithDiscovery.registerWithEureka(server.getServerPort(), eurekaClient, vipAddress);

        /**
         * Using the eureka client, create an Ocelli Host event stream.
         * Ocelli, uses this host stream to know about the available hosts.
         */
        Observable<MembershipEvent<Host>> eurekaHostSource = createEurekaHostStream(eurekaClient, vipAddress);

        /**
         * Instead of directly using the host and port from eureka as in example {@link ClientServerWithDiscovery},
         * choose a host from the load balancer.
         */
        createRequestFromLB(eurekaHostSource)
                /* Block till you get the response. In a real world application, one should not be blocked but chained
                 * into a response to the caller. */
                .toBlocking()
                /**
                 * Print each content of the response.
                 */
                .forEach(System.out::println);
    }

项目：RxNetty 文件：RxNetty.java

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

项目：RxNetty 文件：RxNetty.java

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

项目：karyon 文件：KaryonTransport.java

public static <I, O> HttpServer<I, O> newHttpServer(int port, RequestHandler<I, O> router) {
    return newHttpServerBuilder(port, router).build();
}

项目：karyon 文件：KaryonTransport.java

public static <I, O> HttpServer<I, O> newHttpServer(int port, HttpRequestHandler<I, O> requestHandler) {
    return newHttpServerBuilder(port, requestHandler).build();
}

项目：karyon 文件：Karyon.java

/**
 * Creates a new {@link KaryonServer} which combines lifecycle of the passed {@link HttpServer} with
 * it's own lifecycle.
 *
 * @param server HTTP server
 * @param modules Additional bootstrapModules if any.
 *
 * @return {@link KaryonServer} which is to be used to start the created server.
 */
public static KaryonServer forHttpServer(HttpServer<?, ?> server, Module... modules) {
    return forHttpServer(server, toBootstrapModule(modules));
}