@Test public void ensureClientIsCreatedBasedOnServiceDiscovery(TestContext context) { Vertx vertx = Vertx.newInstance(rule.vertx()); ServiceDiscovery discovery = spy(ServiceDiscovery.create(vertx)); Async async = context.async(); discovery.publish(HttpEndpoint.createRecord("name", "localhost"), ar -> { assertThat(ar.succeeded()).isTrue(); MicroserviceClientFactory factory = new MicroserviceClientFactory(vertx, discovery); Future<Void> future = spy(Future.future()); factory.createClient("name", future).subscribe(client -> { verify(discovery).getRecordObservable(eq(new JsonObject().put("name", "name"))); async.complete(); }, context::fail); }); }
@Nonnull public Single<HttpClient> createClient(String name, Future<Void> closed) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(closed, "closed"); if (closed.isComplete()) { return Single.error(new IllegalArgumentException("Given Future is already closed")); } return Single.create(subscriber -> { CircuitBreaker breaker = getBreaker(name); Future<HttpClient> executed = breaker.execute(future -> { discovery .getRecordObservable(new JsonObject().put("name", name)) .map(discovery::getReference) .flatMap(reference -> { HttpClient client = new HttpClient(reference.get()); closed.setHandler(ar -> { reference.release(); }); return Observable.just(client); }) .toSingle() .subscribe(future::complete, future::fail); }); executed.setHandlerObservable().toSingle().subscribe(subscriber); }); }
MicroserviceClientFactory { @Nonnull public Single<HttpClient> createClient(String name, Future<Void> closed) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(closed, "closed"); if (closed.isComplete()) { return Single.error(new IllegalArgumentException("Given Future is already closed")); } return Single.create(subscriber -> { CircuitBreaker breaker = getBreaker(name); Future<HttpClient> executed = breaker.execute(future -> { discovery .getRecordObservable(new JsonObject().put("name", name)) .map(discovery::getReference) .flatMap(reference -> { HttpClient client = new HttpClient(reference.get()); closed.setHandler(ar -> { reference.release(); }); return Observable.just(client); }) .toSingle() .subscribe(future::complete, future::fail); }); executed.setHandlerObservable().toSingle().subscribe(subscriber); }); } }
MicroserviceClientFactory { @Nonnull public Single<HttpClient> createClient(String name, Future<Void> closed) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(closed, "closed"); if (closed.isComplete()) { return Single.error(new IllegalArgumentException("Given Future is already closed")); } return Single.create(subscriber -> { CircuitBreaker breaker = getBreaker(name); Future<HttpClient> executed = breaker.execute(future -> { discovery .getRecordObservable(new JsonObject().put("name", name)) .map(discovery::getReference) .flatMap(reference -> { HttpClient client = new HttpClient(reference.get()); closed.setHandler(ar -> { reference.release(); }); return Observable.just(client); }) .toSingle() .subscribe(future::complete, future::fail); }); executed.setHandlerObservable().toSingle().subscribe(subscriber); }); }  }
MicroserviceClientFactory { @Nonnull public Single<HttpClient> createClient(String name, Future<Void> closed) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(closed, "closed"); if (closed.isComplete()) { return Single.error(new IllegalArgumentException("Given Future is already closed")); } return Single.create(subscriber -> { CircuitBreaker breaker = getBreaker(name); Future<HttpClient> executed = breaker.execute(future -> { discovery .getRecordObservable(new JsonObject().put("name", name)) .map(discovery::getReference) .flatMap(reference -> { HttpClient client = new HttpClient(reference.get()); closed.setHandler(ar -> { reference.release(); }); return Observable.just(client); }) .toSingle() .subscribe(future::complete, future::fail); }); executed.setHandlerObservable().toSingle().subscribe(subscriber); }); }  @Nonnull Single<HttpClient> createClient(String name, Future<Void> closed); }
MicroserviceClientFactory { @Nonnull public Single<HttpClient> createClient(String name, Future<Void> closed) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(closed, "closed"); if (closed.isComplete()) { return Single.error(new IllegalArgumentException("Given Future is already closed")); } return Single.create(subscriber -> { CircuitBreaker breaker = getBreaker(name); Future<HttpClient> executed = breaker.execute(future -> { discovery .getRecordObservable(new JsonObject().put("name", name)) .map(discovery::getReference) .flatMap(reference -> { HttpClient client = new HttpClient(reference.get()); closed.setHandler(ar -> { reference.release(); }); return Observable.just(client); }) .toSingle() .subscribe(future::complete, future::fail); }); executed.setHandlerObservable().toSingle().subscribe(subscriber); }); }  @Nonnull Single<HttpClient> createClient(String name, Future<Void> closed);  }