@Test public void testSucceededFuture(TestContext context) { Vertx vertx = Vertx.newInstance(rule.vertx()); ServiceDiscovery discovery = ServiceDiscovery.create(vertx); MicroserviceClientFactory factory = new MicroserviceClientFactory(vertx, discovery); Async async = context.async(); factory.createClient("name", Future.succeededFuture()) .subscribe(client -> { context.fail(); }, e -> { assertThat(e).isInstanceOf(IllegalArgumentException.class); async.complete(); }); }
@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);  }