@Test
public void testLookupWithNonMatchingQuery() throws UnsupportedEncodingException {
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");

  Record record1 = EventBusService.createRecord("Hello", "address", HelloService.class,
    new JsonObject().put("key", "foo"));
  Record record2 = EventBusService.createRecord("Hello-2", "address", HelloService.class,
    new JsonObject().put("key", "bar"));

  discovery.publish(record1, (r) -> {
  });
  discovery.publish(record2, (r) -> {
  });

  await().until(() -> record1.getRegistration() != null);
  await().until(() -> record2.getRegistration() != null);

  JsonArray services = given()
    .param("query", "{\"stuff\":\"*\"}")
    .get("/discovery")
    .asJsonArray();

  assertThat(services.size()).isEqualTo(0);
}
 

private void authUaaHandler(RoutingContext context) {
  if (context.user() != null) {
    JsonObject principal = context.user().principal();
    String username = null;  // TODO: Only for demo. Complete this in next version.
    // String username = KeycloakHelper.preferredUsername(principal);
    if (username == null) {
      context.response()
        .putHeader("content-type", "application/json")
        .end(new Account().setId("TEST666").setUsername("Eric").toString()); // TODO: no username should be an error
    } else {
      Future<AccountService> future = Future.future();
      EventBusService.getProxy(discovery, AccountService.class, future.completer());
      future.compose(accountService -> {
        Future<Account> accountFuture = Future.future();
        accountService.retrieveByUsername(username, accountFuture.completer());
        return accountFuture.map(a -> {
          ServiceDiscovery.releaseServiceObject(discovery, accountService);
          return a;
        });
      })
        .setHandler(resultHandlerNonEmpty(context)); // if user does not exist, should return 404
    }
  } else {
    context.fail(401);
  }
}
 

private Future<PortfolioService> getPortfolioService(ServiceDiscovery discovery) {
    Future<PortfolioService> future = Future.future();
    EventBusService.getServiceProxy(discovery,
        rec -> rec.getName().equalsIgnoreCase("portfolio"),
        PortfolioService.class,
        future);
    return future;
}
 

private Future<PortfolioService> getPortfolioService(ServiceDiscovery discovery) {
    Future<PortfolioService> future = Future.future();
    EventBusService.getServiceProxy(discovery,
        rec -> rec.getName().equalsIgnoreCase("portfolio"),
        PortfolioService.class,
        future);
    return future;
}
 

@Override
public void start(Future<Void> future) {
    super.start();

    // Get configuration
    config = ConfigFactory.load();

    String company = TraderUtils.pickACompany();
    int numberOfShares = TraderUtils.pickANumber();

    EventBus eventBus = vertx.eventBus();
    EventBusService.getProxy(discovery, PortfolioService.class, ar -> {
        if (ar.failed()) {
            System.out.println("Portfolio service could not be retrieved: " + ar.cause());
        } else {
            // Our services:
            PortfolioService portfolio = ar.result();
            MessageConsumer<JsonObject> marketConsumer = eventBus.consumer(config.getString("market.address"));

            // Listen to the market...
            marketConsumer.handler(message -> {
                JsonObject quote = message.body();
                TraderUtils.dumbTradingLogic(company, numberOfShares, portfolio, quote);
            });
        }
    });
}
 

public void example1(ServiceDiscovery discovery) {
  Record record = EventBusService.createRecord(
      "some-eventbus-service", // The service name
      "address", // the service address,
      MyService.class // the service interface
  );

  discovery.publish(record, ar -> {
    // ...
  });

}
 

public void example3(ServiceDiscovery discovery) {
  EventBusService.getProxy(discovery, MyService.class, ar -> {
    if (ar.succeeded()) {
      MyService service = ar.result();

      // Dont' forget to release the service
      ServiceDiscovery.releaseServiceObject(discovery, service);
    }
  });
}
 

public void example1(ServiceDiscovery discovery) {
  Record record = EventBusService.createRecord(
      "some-eventbus-service", // The service name
      "address", // the service address,
      "examples.MyService", // the service interface as string
      new JsonObject()
          .put("some-metadata", "some value")
  );

  discovery.publish(record, ar -> {
    // ...
  });
}
 

public void example3(ServiceDiscovery discovery) {
  EventBusService.getProxy(discovery, MyService.class, ar -> {
    if (ar.succeeded()) {
      MyService service = ar.result();

      // Dont' forget to release the service
      ServiceDiscovery.releaseServiceObject(discovery, service);
    }
  });
}
 

public void example31(ServiceDiscovery discovery) {
  EventBusService.getServiceProxyWithJsonFilter(discovery,
    new JsonObject().put("service.interface", "org.acme.MyService"), // The java interface
    MyService.class, // The expect client
    ar -> {
    if (ar.succeeded()) {
      MyService service = ar.result();

      // Dont' forget to release the service
      ServiceDiscovery.releaseServiceObject(discovery, service);
    }
  });
}
 

@Test
public void testThatWeGetTheTwoPublishedServicesWithMetadata() {
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");

  Record record1 = EventBusService.createRecord("Hello", "address", HelloService.class,
    new JsonObject().put("key", "foo"));
  Record record2 = EventBusService.createRecord("Hello-2", "address", HelloService.class,
    new JsonObject().put("key", "bar"));

  discovery.publish(record1, (r) -> {
  });
  discovery.publish(record2, (r) -> {
  });

  await().until(() -> record1.getRegistration() != null);
  await().until(() -> record2.getRegistration() != null);

  Restafari.Response response = get("/discovery");
  JsonArray services = new JsonArray(response.asString());

  assertThat(services.size()).isEqualTo(2);

  for (Object json : services) {
    Record rec = new Record((JsonObject) json);
    assertThat(rec.getStatus()).isEqualTo(Status.UP);
    assertThat(rec.getRegistration()).isNotNull();
    assertThat(rec.getName()).startsWith("Hello");
    assertThat(rec.getMetadata().getString("key")).isNotNull();

    get("/discovery/" + rec.getRegistration()).then().body("name", not(nullValue()));
  }
}
 

@Test
public void testLookupWithQuery() throws UnsupportedEncodingException {
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");

  Record record1 = EventBusService.createRecord("Hello", "address", HelloService.class,
    new JsonObject().put("key", "foo"));
  Record record2 = EventBusService.createRecord("Hello-2", "address", HelloService.class,
    new JsonObject().put("key", "bar"));

  discovery.publish(record1, (r) -> {
  });
  discovery.publish(record2, (r) -> {
  });

  await().until(() -> record1.getRegistration() != null);
  await().until(() -> record2.getRegistration() != null);


  JsonArray services =
    given()
      .param("query", "{\"name\":\"Hello\"}")
      .get("/discovery")
      .asJsonArray();

  assertThat(services.size()).isEqualTo(1);
}
 

/**
 * Fetch global counter of order from the cache infrastructure.
 *
 * @param key counter key (type)
 * @return async result of the counter
 */
private Future<Long> retrieveCounter(String key) {
  Future<Long> future = Future.future();
  EventBusService.getProxy(discovery, CounterService.class,
    ar -> {
      if (ar.succeeded()) {
        CounterService service = ar.result();
        service.addThenRetrieve(key, future.completer());
      } else {
        future.fail(ar.cause());
      }
    });
  return future;
}
 

private Future<ShoppingCart> getCurrentCart(String userId) {
  Future<ShoppingCartService> future = Future.future();
  EventBusService.getProxy(discovery, ShoppingCartService.class, future.completer());
  return future.compose(service -> {
    Future<ShoppingCart> cartFuture = Future.future();
    service.getShoppingCart(userId, cartFuture.completer());
    return cartFuture.compose(c -> {
      if (c == null || c.isEmpty())
        return Future.failedFuture(new IllegalStateException("Invalid shopping cart"));
      else
        return Future.succeededFuture(c);
    });
  });
}
 

/**
 * Save checkout cart event for current user.
 *
 * @param userId user id
 * @return async result
 */
private Future<Void> saveCheckoutEvent(String userId) {
  Future<ShoppingCartService> future = Future.future();
  EventBusService.getProxy(discovery, ShoppingCartService.class, future.completer());
  return future.compose(service -> {
    Future<Void> resFuture = Future.future();
    CartEvent event = CartEvent.createCheckoutEvent(userId);
    service.addCartEvent(event, resFuture.completer());
    return resFuture;
  });
}
 

@Override
public void start(Future<Void> future) {
  super.start();

  //----
  // Initialize the trader
  String company = TraderUtils.pickACompany();
  int numberOfShares = TraderUtils.pickANumber();
  System.out.println("Java compulsive trader configured for company " + company + " and shares: " + numberOfShares);

  // We need to retrieve two services, create two futures object that will get the services
  Future<MessageConsumer<JsonObject>> marketFuture = Future.future();
  Future<PortfolioService> portfolioFuture = Future.future();
  // Retrieve the services, use the "special" completed to assign the future
  MessageSource.getConsumer(discovery, new JsonObject().put("name", "market-data"), marketFuture);
  EventBusService.getProxy(discovery, PortfolioService.class, portfolioFuture);

  // When done (both services retrieved), execute the handler
  CompositeFuture.all(marketFuture, portfolioFuture).setHandler(ar -> {
    if (ar.failed()) {
      future.fail("One of the required service cannot " +
          "be retrieved: " + ar.cause());
    } else {
      // Our services:
      PortfolioService portfolio = portfolioFuture.result();
      MessageConsumer<JsonObject> marketConsumer = marketFuture.result();

      // Listen the market...
      marketConsumer.handler(message -> {
        JsonObject quote = message.body();
        TraderUtils.dumbTradingLogic(company, numberOfShares, portfolio, quote);
      });

      future.complete();
    }
  });
  // ----
}
 

@Test
public void testWithGroovyConsumer() {
  // Step 1 - register the service
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");
  Record record = EventBusService.createRecord("Hello", "address", HelloService.class);

  discovery.publish(record, (r) -> {
  });
  await().until(() -> record.getRegistration() != null);

  // Step 2 - register a consumer that get the result
  AtomicReference<JsonObject> result = new AtomicReference<>();
  vertx.eventBus().<JsonObject>consumer("result", message -> result.set(message.body()));

  // Step 3 - deploy the verticle
  vertx.deployVerticle("discovery/verticles/HelloServiceConsumer.groovy", ar -> {
    if (ar.failed()) {
      // Will fail anyway.
      ar.cause().printStackTrace();
    }
  });

  await().until(() -> result.get() != null);

  assertThat(result.get().getString("status")).isEqualTo("ok");
  assertThat(result.get().getString("message")).isEqualTo("stuff vert.x");
}
 

@Test
public void testWithGroovyConsumerWithJsonFilter() {
  // Step 1 - register the service
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");
  Record record = EventBusService.createRecord("Hello", "address", HelloService.class);

  discovery.publish(record, (r) -> {
  });
  await().until(() -> record.getRegistration() != null);

  // Step 2 - register a consumer that get the result
  AtomicReference<JsonObject> result = new AtomicReference<>();
  vertx.eventBus().<JsonObject>consumer("result", message -> result.set(message.body()));

  // Step 3 - deploy the verticle
  vertx.deployVerticle("discovery/verticles/HelloServiceConsumerWithJsonFilter.groovy", ar -> {
    if (ar.failed()) {
      // Will fail anyway.
      ar.cause().printStackTrace();
    }
  });

  await().until(() -> result.get() != null);

  assertThat(result.get().getString("status")).isEqualTo("ok");
  assertThat(result.get().getString("message")).isEqualTo("stuff vert.x");
}
 

public void publishEventBusService(String name, String address, Class serviceClass, Handler<AsyncResult<Void>>
        completionHandler) {
    Record record = EventBusService.createRecord(name, address, serviceClass);
    publish(record, completionHandler);
}
 

@Test
public void testServiceUsage() throws InterruptedException {
  List<JsonObject> usages = new ArrayList<>();


  vertx.eventBus().<JsonObject>consumer(ServiceDiscoveryOptions.DEFAULT_USAGE_ADDRESS,
    msg -> usages.add(msg.body()));

  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");

  Record record = new Record()
    .setName("Hello")
    .setMetadata(new JsonObject()
      .put("key", "A")
      .put("service.interface", HelloService.class.getName()))
    .setType(EventBusService.TYPE)
    .setLocation(new JsonObject().put(Record.ENDPOINT, "address"));
  discovery.publish(record, (r) -> {
  });
  await().until(() -> record.getRegistration() != null);

  ServiceReference reference = discovery.getReference(record);
  await().until(() -> usages.size() == 1);

  assertThat(usages.get(0).getJsonObject("record").getJsonObject("location").getString(Record.ENDPOINT))
    .isEqualToIgnoringCase("address");
  assertThat(usages.get(0).getString("type")).isEqualTo("bind");
  assertThat(usages.get(0).getString("id")).isNotNull().isNotEmpty();

  assertThat((HelloService) reference.cached()).isNull();
  assertThat((HelloService) reference.get()).isNotNull();
  assertThat((HelloService) reference.cached()).isNotNull();

  reference.release();
  Assertions.assertThat(discovery.bindings()).isEmpty();
  await().until(() -> usages.size() == 2);
  assertThat(usages.get(1).getJsonObject("record").getJsonObject("location").getString(Record.ENDPOINT))
    .isEqualToIgnoringCase("address");
  assertThat(usages.get(1).getString("type")).isEqualTo("release");
  assertThat(usages.get(1).getString("id")).isNotNull().isNotEmpty();

  // Check that even if we release the reference another time the service event is not send a second time.
  reference.release();
  Assertions.assertThat(discovery.bindings()).isEmpty();
  Thread.sleep(100);
  assertThat(usages).hasSize(2);
}
 

@Test
public void testExporter() {
  HelloService svc = new HelloServiceImpl("stuff");
  ProxyHelper.registerService(HelloService.class, vertx, svc, "address");
  Record record = new Record()
    .setName("Hello")
    .setType(EventBusService.TYPE)
    .setLocation(new JsonObject().put(Record.ENDPOINT, "address"))
    .setMetadata(new JsonObject().put("foo", "foo_value_1"));

  TestServiceExporter exporter = new TestServiceExporter();
  discovery.registerServiceExporter(exporter, new JsonObject());

  discovery.publish(record, (r) -> {
  });
  await().until(() -> exporter.state.size() > 0);
  String id = exporter.state.keySet().iterator().next();
  assertNotNull(id);
  Record exported = exporter.state.get(id);
  assertEquals("Hello", exported.getName());
  assertEquals(EventBusService.TYPE, exported.getType());
  assertEquals(Status.UP, exported.getStatus());
  assertEquals(new JsonObject().put(Record.ENDPOINT, "address"), exported.getLocation());
  assertEquals(new JsonObject().put("foo", "foo_value_1"), exported.getMetadata());

  AtomicBoolean updated = new AtomicBoolean();
  discovery.update(new Record(record).setMetadata(new JsonObject().put("foo", "foo_value_2")), ar -> updated.set(true));
  await().until(updated::get);
  assertNotSame(exporter.state.get(id), exported);
  exported = exporter.state.get(id);
  assertEquals("Hello", exported.getName());
  assertEquals(EventBusService.TYPE, exported.getType());
  assertEquals(Status.UP, exported.getStatus());
  assertEquals(new JsonObject().put(Record.ENDPOINT, "address"), exported.getLocation());
  assertEquals(new JsonObject().put("foo", "foo_value_2"), exported.getMetadata());

  AtomicBoolean removed = new AtomicBoolean();
  discovery.unpublish(id, ar -> removed.set(true));
  await().until(removed::get);
  assertEquals(Collections.emptyMap(), exporter.state);

  discovery.close();
  assertTrue(exporter.closed);
}
 

/**
 * Get product service from the service discovery infrastructure.
 *
 * @return async result of the service.
 */
private Future<ProductService> getProductService() {
  Future<ProductService> future = Future.future();
  EventBusService.getProxy(discovery, ProductService.class, future.completer());
  return future;
}
 

protected Future<Void> publishEventBusService(String name, String address, Class serviceClass) {
  Record record = EventBusService.createRecord(name, address, serviceClass);
  return publish(record);
}
 

protected Single<Void> publishEventBusService(String name, String address, Class serviceClass) {
  Record record = EventBusService.createRecord(name, address, serviceClass);
  return publish(record);
}
 

public void publishEventBusService(String name, String address, Class<?> serviceClass, Handler<AsyncResult<Void>>
    completionHandler) {
  Record record = EventBusService.createRecord(name, address, serviceClass);
  publish(record, completionHandler);
}