watch-sse.java

private void watch(WebClient client, String token, String resourceVersion) {
    String path = "/api/v1/namespaces/" + namespace + "/services";
    JsonParser parser = JsonParser.newParser().objectValueMode()
      .handler(event -> onChunk(event.objectValue()));

    client.get(path)
      .addQueryParam("watch", "true")
      .addQueryParam("resourceVersion", resourceVersion)
      .as(BodyCodec.pipe(new WriteStream<Buffer>() {
        @Override
        public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
          return this;
        }

        @Override
        public Future<Void> write(Buffer data) {
          parser.write(data);
          return Future.succeededFuture();
        }

        @Override
        public void write(Buffer data, Handler<AsyncResult<Void>> handler) {
          parser.write(data);
          if (handler != null) {
            handler.handle(Future.succeededFuture());
          }
        }

        @Override
        public void end(Handler<AsyncResult<Void>> handler) {
          try {
            parser.end();
          } catch (Exception e) {
            handler.handle(Future.failedFuture(e));
            return;
          }
          handler.handle(Future.succeededFuture());
        }

        @Override
        public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
          return this;
        }

        @Override
        public boolean writeQueueFull() {
          return false;
        }

        @Override
        public WriteStream<Buffer> drainHandler(Handler<Void> handler) {
          return this;
        }
      }))
      .putHeader("Authorization", "Bearer " + token)
      .send(ar -> {
        if (ar.failed()) {
          LOGGER.error("Unable to setup the watcher on the service list", ar.cause());
        } else {
          LOGGER.info("Watching services from namespace " + namespace);
        }
      });
  }