import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchPage(String name, Handler<AsyncResult<JsonObject>> resultHandler) {
  dbClient.rxQueryWithParams(sqlQueries.get(SqlQuery.GET_PAGE), new JsonArray().add(name))
    .map(result -> {
      if (result.getNumRows() > 0) {
        JsonArray row = result.getResults().get(0);
        return new JsonObject()
          .put("found", true)
          .put("id", row.getInteger(0))
          .put("rawContent", row.getString(1));
      } else {
        return new JsonObject().put("found", false);
      }
    })
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchPageById(int id, Handler<AsyncResult<JsonObject>> resultHandler) {
  // tag::rx-execute-query-with-params[]
  Single<ResultSet> resultSet = dbClient.rxQueryWithParams(
    sqlQueries.get(SqlQuery.GET_PAGE_BY_ID), new JsonArray().add(id));
  // end::rx-execute-query-with-params[]
  resultSet
    .map(result -> {
      if (result.getNumRows() > 0) {
        JsonObject row = result.getRows().get(0);
        return new JsonObject()
          .put("found", true)
          .put("id", row.getInteger("ID"))
          .put("name", row.getString("NAME"))
          .put("content", row.getString("CONTENT"));
      } else {
        return new JsonObject().put("found", false);
      }
    })
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchPageById(int id, Handler<AsyncResult<JsonObject>> resultHandler) {
  Single<ResultSet> resultSet = dbClient.rxQueryWithParams(
    sqlQueries.get(SqlQuery.GET_PAGE_BY_ID), new JsonArray().add(id));
  resultSet
    .map(result -> {
      if (result.getNumRows() > 0) {
        JsonObject row = result.getRows().get(0);
        return new JsonObject()
          .put("found", true)
          .put("id", row.getInteger("ID"))
          .put("name", row.getString("NAME"))
          .put("content", row.getString("CONTENT"));
      } else {
        return new JsonObject().put("found", false);
      }
    })
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
private Observable<ObjectNode> observeGetObject(String endpoint) {
    ObservableHandler<HttpClientResponse> responseObservable = RxHelper.observableHandler();
    client.getNow(endpoint, responseObservable.toHandler());
    return responseObservable
        .flatMap(response -> {
            if (response.statusCode() == HTTP_OK) {
                ObservableHandler<Buffer> observable = RxHelper.observableHandler();
                response.bodyHandler(observable.toHandler());
                return observable;
            } else {
                throw new RuntimeException(String.format(
                        "Endpoint '%s' get returned %d status",
                        endpoint, response.statusCode()));
            }
        })
        .map(buffer -> Json.decodeValue(buffer.toString(UTF_8.name()), ObjectNode.class));
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
/**
 * Convert a chunk to a Elasticsearch document
 * @param chunk the chunk to convert
 * @param fallbackCRSString a string representing the CRS that should be used
 * to index the chunk if it does not specify a CRS itself (may be null if no
 * CRS is available as fallback)
 * @param parserOperator the operator used to parse the chunk stream into
 * stream events
 * @param indexerFactories a sequence of indexer factories that should be
 * used to index the chunk
 * @param <T> the type of the stream events created by <code>parserOperator</code>
 * @return an observable that will emit the document
 */
private <T extends StreamEvent> Observable<Map<String, Object>> chunkToDocument(
    ChunkReadStream chunk, String fallbackCRSString,
    Operator<T, Buffer> parserOperator,
    List<? extends IndexerFactory> indexerFactories) {
  List<StreamIndexer<T>> indexers = new ArrayList<>();
  indexerFactories.forEach(factory -> {
    @SuppressWarnings("unchecked")
    StreamIndexer<T> i = (StreamIndexer<T>)factory.createIndexer();
    if (fallbackCRSString != null && i instanceof CRSAware) {
      ((CRSAware)i).setFallbackCRSString(fallbackCRSString);
    }
    indexers.add(i);
  });
  
  return RxHelper.toObservable(chunk)
    .lift(parserOperator)
    .doOnNext(e -> indexers.forEach(i -> i.onEvent(e)))
    .last() // "wait" until the whole chunk has been consumed
    .map(e -> {
      // create the Elasticsearch document
      Map<String, Object> doc = new HashMap<>();
      indexers.forEach(i -> doc.putAll(i.getResult()));
      return doc;
    });
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Test
public void testStreamRX() {
  String sql = "SELECT ID, FNAME, LNAME FROM select_table ORDER BY ID";
  connection().queryStream(sql, onSuccess(res -> {

    final AtomicInteger cnt = new AtomicInteger(0);

    Observable<JsonArray> observable = RxHelper.toObservable(res);
    observable.subscribe(
      // handle one row
      row -> {
        assertEquals("doe", row.getString(res.column("lname")));
        cnt.incrementAndGet();
      },
      // it should not fail
      this::fail,
      () -> {
        assertEquals(2, cnt.get());
        testComplete();
      });
  }));

  await();
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public void start() throws Exception {
    //TODO: Fix a better way of configuration other than system properties?
    Integer port = Integer.getInteger("websocket.port", 5556);

    ObservableFuture<HttpServer> httpServerObservable = RxHelper.observableFuture();
    HttpServer httpServer = vertx.createHttpServer(new HttpServerOptions().setPort(port));
    httpServerObservable.subscribe(
            a -> log.info("Starting web socket listener..."),
            e -> log.error("Could not start web socket listener at port " + port, e),
            () -> log.info("Started web socket listener on port " + port)
    );

    Observable<Tup2<ServerWebSocket, Func1<Event, Boolean>>> eventObservable = EventObservable.convertFromWebSocketObservable(RxHelper.toObservable(httpServer.websocketStream()));
    eventObservable.subscribe(new EventToJsonAction(Riemann.getEvents(vertx), WebSocketFrameImpl::new), e -> {
        log.error(e);
        //TODO: Fix proper error handling
    });

    httpServer.listen(httpServerObservable.asHandler());
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Test
public void testScheduledTimer() {
  vertx.runOnContext(v -> {
    long startTime = System.currentTimeMillis();
    Context initCtx = Vertx.currentContext();
    Observable.timer(100, 100, TimeUnit.MILLISECONDS, io.vertx.rxjava.core.RxHelper.scheduler(vertx)).take(10).subscribe(new Observer<Long>() {
      public void onNext(Long value) {
        assertEquals(initCtx.getDelegate(), Vertx.currentContext().getDelegate());
      }

      public void onError(Throwable e) {
        fail("unexpected failure");
      }

      public void onCompleted() {
        long timeTaken = System.currentTimeMillis() - startTime;
        assertTrue("Was expecting to have time taken | " + timeTaken + " -  1000 | < 200", Math.abs(timeTaken - 1000) < 1000);
        testComplete();
      }
    });
  });
  await();
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Test
public void testObserverToHandler() throws Exception {
  AtomicInteger count = new AtomicInteger();
  Observer<Long> observer = new Observer<Long>() {
    @Override
    public void onCompleted() {
      assertEquals(1, count.get());
      testComplete();
    }

    @Override
    public void onError(Throwable e) {
      fail(e.getMessage());
    }

    @Override
    public void onNext(Long l) {
      count.incrementAndGet();
    }
  };
  vertx.setTimer(1, RxHelper.toHandler(observer));
  await();
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Path("4")
@GET
public void hello4(@Suspended final AsyncResponse asyncResponse,
	      // Inject the Vertx instance
	      @Context Vertx vertx){
	System.err.println("Creating client");
	HttpClientOptions options = new HttpClientOptions();
	options.setSsl(true);
	options.setTrustAll(true);
	options.setVerifyHost(false);
	HttpClient client = vertx.createHttpClient(options);
	ObservableHandler<HttpClientResponse> responseHandler = RxHelper.observableHandler();
	client.getNow(443,
			"www.google.com", 
			"/robots.txt", 
			responseHandler.toHandler());
	
	ObservableHandler<Buffer> bodyHandler = RxHelper.observableHandler();
	responseHandler.subscribe(resp -> {
		System.err.println("Got response");
		resp.bodyHandler(bodyHandler.toHandler());
	});
	
	bodyHandler.subscribe(body -> {
		System.err.println("Got body");
		asyncResponse.resume(Response.ok(body.toString()).build());
	});
	System.err.println("Created client");
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
WikiDatabaseServiceImpl(io.vertx.ext.jdbc.JDBCClient dbClient, HashMap<SqlQuery, String> sqlQueries, Handler<AsyncResult<WikiDatabaseService>> readyHandler) {
  this.dbClient = new JDBCClient(dbClient);
  this.sqlQueries = sqlQueries;

  getConnection()
    .flatMap(conn -> conn.rxExecute(sqlQueries.get(SqlQuery.CREATE_PAGES_TABLE)))
    .map(v -> this)
    .subscribe(RxHelper.toSubscriber(readyHandler));
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
// tag::rx-data-flow[]
public WikiDatabaseService fetchAllPages(Handler<AsyncResult<JsonArray>> resultHandler) {
  dbClient.rxQuery(sqlQueries.get(SqlQuery.ALL_PAGES))
    .flatMapObservable(res -> {  // <1>
      List<JsonArray> results = res.getResults();
      return Observable.from(results); // <2>
    })
    .map(json -> json.getString(0)) // <3>
    .sorted() // <4>
    .collect(JsonArray::new, JsonArray::add) // <5>
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService createPage(String title, String markdown, Handler<AsyncResult<Void>> resultHandler) {
  dbClient.rxUpdateWithParams(sqlQueries.get(SqlQuery.CREATE_PAGE), new JsonArray().add(title).add(markdown))
    .map(res -> (Void) null)
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService savePage(int id, String markdown, Handler<AsyncResult<Void>> resultHandler) {
  dbClient.rxUpdateWithParams(sqlQueries.get(SqlQuery.SAVE_PAGE), new JsonArray().add(markdown).add(id))
    .map(res -> (Void) null)
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService deletePage(int id, Handler<AsyncResult<Void>> resultHandler) {
  JsonArray data = new JsonArray().add(id);
  dbClient.rxUpdateWithParams(sqlQueries.get(SqlQuery.DELETE_PAGE), data)
    .map(res -> (Void) null)
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchAllPagesData(Handler<AsyncResult<List<JsonObject>>> resultHandler) { // <1>
  dbClient.rxQuery(sqlQueries.get(SqlQuery.ALL_PAGES_DATA))
    .map(ResultSet::getRows)
    .subscribe(RxHelper.toSubscriber(resultHandler));  // <2>
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchAllPages(Handler<AsyncResult<JsonArray>> resultHandler) {
  dbClient.rxQuery(sqlQueries.get(SqlQuery.ALL_PAGES))
    .flatMapObservable(res -> {
      List<JsonArray> results = res.getResults();
      return Observable.from(results);
    })
    .map(json -> json.getString(0))
    .sorted()
    .collect(JsonArray::new, JsonArray::add)
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}
 

import io.vertx.rx.java.RxHelper; //导入依赖的package包/类
@Override
public WikiDatabaseService fetchAllPagesData(Handler<AsyncResult<List<JsonObject>>> resultHandler) {
  dbClient.rxQuery(sqlQueries.get(SqlQuery.ALL_PAGES_DATA))
    .map(ResultSet::getRows)
    .subscribe(RxHelper.toSubscriber(resultHandler));
  return this;
}