import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
private Single<HttpServer> configureTheHTTPServer() {

        //TODO
        //----

        Router router = Router.router(vertx);
        router.get("/").handler(this::retrieveOperations);
        router.get("/health").handler(rc -> {
            if (ready) {
                rc.response().end("Ready");
            } else {
                // Service not yet available
                rc.response().setStatusCode(503).end();
            }
        });
        return vertx.createHttpServer().requestHandler(router::accept).rxListen(8080);
        //----

//        return ready;
    }
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void delayFlowable(HttpServer server) {
  server.requestHandler(request -> {
    if (request.method() == HttpMethod.POST) {

      // Stop receiving buffers
      request.pause();

      checkAuth(res -> {

        // Now we can receive buffers again
        request.resume();

        if (res.succeeded()) {
          Flowable<Buffer> flowable = request.toFlowable();
          flowable.subscribe(buff -> {
            // Get buffers
          });
        }
      });
    }
  });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void completable(HttpServer server) {

    // Obtain a completable that performs the actual close on subscribe
    Completable single = server.rxClose();

    // Subscribe to bind the server
    single.
      subscribe(
        () -> {
          // Server is closed
        },
        failure -> {
          // Server closed but encoutered issue
        }
      );
  }
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Test
public void testGet() {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.response().setChunked(true).end("some_content"));
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Single<HttpResponse<Buffer>> single = client
              .get(8080, "localhost", "/the_uri")
              .as(BodyCodec.buffer())
              .rxSend();
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> {
          Buffer body = resp.body();
          assertEquals("some_content", body.toString("UTF-8"));
          complete();
        }, this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Test
public void testPost() {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.bodyHandler(buff -> {
    assertEquals("onetwothree", buff.toString());
    req.response().end();
  }));
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Observable<Buffer> stream = Observable.just(Buffer.buffer("one"), Buffer.buffer("two"), Buffer.buffer("three"));
      Single<HttpResponse<Buffer>> single = client
              .post(8080, "localhost", "/the_uri")
              .rxSendStream(stream);
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> complete(), this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Test
public void testResponseMissingBody() throws Exception {
  int times = 5;
  waitFor(times);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.response().setStatusCode(403).end());
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Single<HttpResponse<Buffer>> single = client
              .get(8080, "localhost", "/the_uri")
              .rxSend();
      for (int i = 0; i < times; i++) {
        single.subscribe(resp -> {
          assertEquals(403, resp.statusCode());
          assertNull(resp.body());
          complete();
        }, this::fail);
      }
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Test
public void testResponseBodyAsAsJsonMapped() throws Exception {
  JsonObject expected = new JsonObject().put("cheese", "Goat Cheese").put("wine", "Condrieu");
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestStream().handler(req -> req.response().end(expected.encode()));
  try {
    server.listen(ar -> {
      client = WebClient.wrap(vertx.createHttpClient(new HttpClientOptions()));
      Single<HttpResponse<WineAndCheese>> single = client
              .get(8080, "localhost", "/the_uri")
              .as(BodyCodec.json(WineAndCheese.class))
              .rxSend();
      single.subscribe(resp -> {
        assertEquals(200, resp.statusCode());
        assertEquals(new WineAndCheese().setCheese("Goat Cheese").setWine("Condrieu"), resp.body());
        testComplete();
      }, this::fail);
    });
    await();
  } finally {
    server.close();
  }
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Override
public void start() {
    /** 1.Call router hub to mount commont **/
    final Axis<Router> routerAxiser = Fn.poolThread(Pool.ROUTERS,
            () -> Instance.instance(RouterAxis.class));
    /** 2.Call route hub to mount defined **/
    final Axis<Router> axiser = Fn.poolThread(Pool.EVENTS,
            () -> Instance.instance(EventAxis.class));

    /** 3.Get the default HttpServer Options **/
    ZeroAtomic.RX_OPTS.forEach((port, option) -> {
        /** 3.1.Single server processing **/
        final HttpServer server = this.vertx.createHttpServer(option);
        /** 3.2. Build router with current option **/
        final Router router = Router.router(this.vertx);

        routerAxiser.mount(router);
        axiser.mount(router);

        /** 3.3. Listen for router on the server **/
        final Single<HttpServer> result =
                server.requestHandler(router::accept).rxListen();
        /** 3.4. Log output **/
        {
            result.subscribe((rxServer) -> {
                recordServer(option, router);
            });
        }
    });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    WebClient client = WebClient.create(vertx);
    HttpServer server = vertx.createHttpServer();
    server
        // 1 - Transform the sequence of request into a stream
        .requestStream().toFlowable()
        // 2 - For each request, call the twitter API
        .flatMapCompletable(req ->
            client.getAbs("https://twitter.com/vertx_project")
                .rxSend()
                // 3 - Extract the body as string
                .map(HttpResponse::bodyAsString)
                // 4 - In case of a failure
                .onErrorReturn(t -> "Cannot access the twitter " +
                    "feed: " + t.getMessage())
                // 5 - Write the response
                .doOnSuccess(res -> req.response().end(res))
                // 6 - Just transform the restul into a completable
                .toCompletable()
        )
        // 7 - Never forget to subscribe to a reactive type,
        // or nothing happens
        .subscribe();

    server.listen(8080);
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
/**
 * Starts the verticle asynchronously. The the initialization is completed, it calls
 * `complete()` on the given {@link Future} object. If something wrong happens,
 * `fail` is called.
 *
 * @param future the future to indicate the completion
 */
@Override
public void start(Future<Void> future) {
    // creates the jdbc client.

    ServiceDiscovery.create(vertx, discovery -> {

        // Discover and configure the database.
        Single<JDBCClient> jdbc = JDBCDataSource.rxGetJDBCClient(discovery,
            svc -> svc.getName().equals("audit-database"),
            getDatabaseConfiguration()
        ).doOnSuccess(jdbcClient -> this.jdbc = jdbcClient);

        // TODO
        // ----

        Single<JDBCClient> databaseReady = jdbc
            .flatMap(client -> initializeDatabase(client, true));
        Single<HttpServer> httpServerReady = configureTheHTTPServer();
        Single<MessageConsumer<JsonObject>> messageConsumerReady = retrieveThePortfolioMessageSource();

        Single<MessageConsumer<JsonObject>> readySingle = Single.zip(databaseReady, httpServerReady,
            messageConsumerReady, (db, http, consumer) -> consumer);

        // ----

        // signal a verticle start failure
        readySingle.doOnSuccess(consumer -> {
            // on success we set the handler that will store message in the database
            consumer.handler(message -> storeInDatabase(message.body()));
        }).subscribe(consumer -> {
            // complete the verticle start with a success
            future.complete();
            ready = true;
        }, future::fail);
    });


}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
private Single<Record> publish(HttpServer server) {
    // 1 - Create a service record using `io.vertx.reactivex.servicediscovery.types.HttpEndpoint.createRecord`.
    // This record define the service name ("greetings"), the host ("localhost"), the server port and the root ("/")
    // TODO

    // 2 - Call the rxPublish method with the created record and return the resulting single
    // TODO
    return null; // To be removed
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Override
public void start(Future<Void> future) throws Exception {
    // Get the stream of messages sent on the "market" address
    vertx.eventBus().<JsonObject>consumer(GeneratorConfigVerticle.ADDRESS).toFlowable()
        // TODO Extract the body of the message using `.map(msg -> {})`
        // ----
        //
        // ----
        // TODO For each message, populate the `quotes` map with the received quote. Use `.doOnNext(json -> {})`
        // Quotes are json objects you can retrieve from the message body
        // The map is structured as follows: name -> quote
        // ----
        //
        // ----
        .subscribe();

    HttpServer server = vertx.createHttpServer();
    server.requestStream().toFlowable()
        .doOnNext(request -> {
            HttpServerResponse response = request.response()
                .putHeader("content-type", "application/json");

            // TODO Handle the HTTP request
            // The request handler returns a specific quote if the `name` parameter is set, or the whole map if none.
            // To write the response use: `response.end(content)`
            // If the name is set but not found, you should return 404 (use response.setStatusCode(404)).
            // To encode a Json object, use the `encorePrettily` method
            // ----

            // Remove this line
            response.end(Json.encodePrettily(quotes));
            
            // ----
        })
    .subscribe();

    server.rxListen(config().getInteger("http.port", 8080))
        .toCompletable()
        .subscribe(CompletableHelper.toObserver(future));
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
private HttpServer createHttpServer() {
  JsonObject serverOptions = configuration.getServerOptions();

  return serverOptions.isEmpty()
      ? vertx.createHttpServer()
      : vertx.createHttpServer(new HttpServerOptions(serverOptions));
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void websocketServer(HttpServer server) {
  Observable<ServerWebSocket> socketObservable = server.websocketStream().toObservable();
  socketObservable.subscribe(
      socket -> System.out.println("Web socket connect"),
      failure -> System.out.println("Should never be called"),
      () -> {
        System.out.println("Subscription ended or server closed");
      }
  );
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void httpServerRequestObservableUnmarshall(HttpServer server) {
  Observable<HttpServerRequest> requestObservable = server.requestStream().toObservable();
  requestObservable.subscribe(request -> {
    Observable<MyPojo> observable = request.
      toObservable().
      compose(io.vertx.reactivex.core.ObservableHelper.unmarshaller(MyPojo.class));
  });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Override
public void start(Future<Void> startFuture) throws Exception {
  HttpServer httpServer = vertx.createHttpServer();

  BookDatabaseService bookDatabaseService = createProxy(vertx.getDelegate(), config().getString(CONFIG_DB_EB_QUEUE));

  Router router = Router.router(vertx);

  router.route().handler(BodyHandler.create());

  router.route().handler(HTTPRequestValidationHandler.create().addExpectedContentType("application/json"));

  router.get(GET_BOOKS).handler(HTTPRequestValidationHandler.create()
    .addQueryParam("title", ParameterType.GENERIC_STRING, false)
    .addQueryParam("category", ParameterType.GENERIC_STRING, false)
    .addQueryParam("publicationDate", ParameterType.DATE, false))
    .handler(new GetBooksHandler(bookDatabaseService));

  router.post(ADD_NEW_BOOK).handler(new AddBookHandler(bookDatabaseService));

  router.delete(DELETE_BOOK_BY_ID).handler(HTTPRequestValidationHandler.create()
    .addPathParam("id", ParameterType.INT))
    .handler(new DeleteBookByIdHandler(bookDatabaseService));

  router.get(GET_BOOK_BY_ID).handler(HTTPRequestValidationHandler.create()
    .addPathParam("id", ParameterType.INT))
    .handler(new GetBookByIdHandler(bookDatabaseService));

  router.put(UPDATE_BOOK_BY_ID).handler(HTTPRequestValidationHandler.create()
    .addPathParam("id", ParameterType.INT))
    .handler(new UpsertBookByIdHandler(bookDatabaseService));

  router.route().failureHandler(new FailureHandler());

  int httpServerPort = config().getInteger(CONFIG_HTTP_SERVER_PORT, 8080);
  httpServer.requestHandler(router::accept)
    .rxListen(httpServerPort)
    .subscribe(server -> {
        LOGGER.info("HTTP server is running on port " + httpServerPort);
        startFuture.complete();
      },
      throwable -> {
        LOGGER.error("Fail to start a HTTP server ", throwable);
        startFuture.fail(throwable);
      });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
@Override
public void start(Future<Void> future) throws Exception {
    // Get the stream of messages sent on the "market" address
    vertx.eventBus().<JsonObject>consumer(GeneratorConfigVerticle.ADDRESS).toFlowable()
        // TODO Extract the body of the message using `.map(msg -> {})`
        // ----
        .map(Message::body)
        // ----
        // TODO For each message, populate the `quotes` map with the received quote. Use `.doOnNext(json -> {})`
        // Quotes are json objects you can retrieve from the message body
        // The map is structured as follows: name -> quote
        // ----
        .doOnNext(json -> {
            quotes.put(json.getString("name"), json); // 2
        })
        // ----
        .subscribe();

    HttpServer server = vertx.createHttpServer();
    server.requestStream().toFlowable()
        .doOnNext(request -> {
            HttpServerResponse response = request.response()
                .putHeader("content-type", "application/json");

            // TODO Handle the HTTP request
            // The request handler returns a specific quote if the `name` parameter is set, or the whole map if none.
            // To write the response use: `response.end(content)`
            // If the name is set but not found, you should return 404 (use response.setStatusCode(404)).
            // To encode a Json object, use the `encorePrettily` method
            // ----
            
            String company = request.getParam("name");
            if (company == null) {
                String content = Json.encodePrettily(quotes);
                response.end(content);
            } else {
                JsonObject quote = quotes.get(company);
                if (quote == null) {
                    response.setStatusCode(404).end();
                } else {
                    response.end(quote.encodePrettily());
                }
            }
            // ----
        })
        .subscribe();

    server.rxListen(config().getInteger("http.port", 8080))
        .toCompletable()
        .subscribe(CompletableHelper.toObserver(future));
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void httpServerRequest(HttpServer server) {
  Observable<HttpServerRequest> requestObservable = server.requestStream().toObservable();
  requestObservable.subscribe(request -> {
    // Process request
  });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
public void httpServerRequestObservable(HttpServer server) {
  Observable<HttpServerRequest> requestObservable = server.requestStream().toObservable();
  requestObservable.subscribe(request -> {
    Observable<Buffer> observable = request.toObservable();
  });
}
 

import io.vertx.reactivex.core.http.HttpServer; //导入依赖的package包/类
private Single<HttpServer> configureTheHTTPServer() {

        //TODO
        //----

        return Single
            .error(new UnsupportedOperationException("Not implemented yet"));
        //----

    }