import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * <p>Publish a service once. The combination of name and address should be
 * unique. If the name and address are already in use, this method will
 * complete successfully.</p>
 * <p><b>Note:</b> For every published service, a cluster-wide counter will
 * be created with the name
 * <code>{@link Service#COUNTER_PREFIX} + name + ":" + address</code>.</p>
 * @param name the service name
 * @param discovery the service discovery that should be used to publish the
 * service
 * @param vertx the Vert.x instance
 * @param address the service address on the event bus
 * @return an single emitting one item when the service has been registered
 */
static Single<Void> publishOnce(String name, String address,
    ServiceDiscovery discovery, Vertx vertx) {
  Record record = new Record()
      .setName(name)
      .setLocation(new JsonObject().put(Record.ENDPOINT, address));

  if (name == null || name.trim().isEmpty()) {
    return Single.error(new IllegalArgumentException("Missing service name"));
  }
  if (address == null || address.trim().isEmpty()) {
    return Single.error(new IllegalArgumentException("Missing endpoint address"));
  }

  return publishOnce(record, discovery, vertx);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Test if a service is really published only once
 * @param context the test context
 */
@Test
public void publishOnce(TestContext context) {
  Vertx vertx = new Vertx(rule.vertx());
  Async async = context.async();

  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);
  Service.publishOnce("A", "A", discovery, vertx)
    .flatMap(v -> Service.publishOnce("A", "A", discovery, vertx))
    .flatMap(v -> Service.publishOnce("A", "B", discovery, vertx))
    .flatMap(v -> {
      return discovery.rxGetRecords(record -> true);
    })
    .doAfterTerminate(discovery::close)
    .subscribe(recordList -> {
      List<Record> lA = recordList.stream()
          .filter(r -> r.getName().equals("A"))
          .collect(Collectors.toList());
      context.assertEquals(2, lA.size());
      async.complete();
    }, context::fail);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Test if a service can be discovered
 * @param context the test context
 */
@Test
public void discover(TestContext context) {
  Vertx vertx = new Vertx(rule.vertx());
  Async async = context.async();

  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);
  Service.publishOnce("A", "a", discovery, vertx)
    .flatMap(v -> Service.publishOnce("A", "b", discovery, vertx))
    .flatMapObservable(v -> Service.discover("A", discovery, vertx))
    .count()
    .doOnTerminate(discovery::close)
    .subscribe(count -> {
      context.assertEquals(2, count);
      async.complete();
    }, context::fail);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Test if a message can be sent and if a consumer receives the message.
 * This method does not test if the message can be sent to only one
 * consumer (instead of all, see {@link #broadcast(TestContext)}),
 * because it's not possible to check if a consumer will not receive a
 * message before the asynchronous test ends.
 * @param context the test context
 */
@Test
public void send(TestContext context) {
  Vertx vertx = new Vertx(rule.vertx());

  Async a = context.async();

  String data = "special data";

  vertx.eventBus().consumer("a", h -> {
    context.assertEquals(data, h.body());
    a.complete();
  });

  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);
  Service.publishOnce("A", "a", discovery, vertx)
    .flatMapObservable(v -> Service.discover("A", discovery, vertx))
    .doOnTerminate(discovery::close)
    .subscribe(service -> {
      service.broadcast(data);
      service.send(data);
    }, context::fail);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Test if a service can be published an unpublished again
 * @param context the test context
 */
@Test
public void unpublish(TestContext context) {
  Vertx vertx = new Vertx(rule.vertx());
  Async async = context.async();

  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);
  Service.publishOnce("A", "a", discovery, vertx)
    .flatMapObservable(v -> Service.discover("A", discovery, vertx))
    .count()
    .doOnNext(count -> {
      context.assertEquals(1, count);
    })
    .flatMap(v -> Service.discover("A", discovery, vertx))
    .flatMapSingle(service -> service.unpublish(discovery))
    .flatMap(v -> Service.discover("A", discovery, vertx))
    .count()
    .doOnTerminate(discovery::close)
    .subscribe(count -> {
      context.assertEquals(0, count);
      async.complete();
    }, context::fail);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start(Future<Void> future) throws Exception {

    Router router = Router.router(vertx);

    circuit = CircuitBreaker.create("circuit-breaker", vertx, new CircuitBreakerOptions()
        .setFallbackOnFailure(true)
        .setMaxFailures(3)
        .setResetTimeout(5000)
        .setTimeout(1000)
    );

    router.get("/").handler(this::getShoppingList);

    ServiceDiscovery.create(vertx, discovery -> {

        Single<WebClient> s1 = HttpEndpoint.rxGetWebClient(discovery,
            rec -> rec.getName().equals("shopping-backend"));
        Single<WebClient> s2 = HttpEndpoint.rxGetWebClient(discovery,
            rec -> rec.getName().equals("pricer-service"));

        Single.zip(s1, s2, (x, y) -> {
            shopping = x;
            pricer = y;
            return vertx.createHttpServer()
                .requestHandler(router::accept)
                .listen(8080);
        }).subscribe();
    });

}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() {
    circuit = CircuitBreaker.create("circuit-breaker", vertx,
        new CircuitBreakerOptions()
            .setFallbackOnFailure(true)
            .setMaxFailures(3)
            .setResetTimeout(5000)
            .setTimeout(1000)
    );


    Router router = Router.router(vertx);
    router.route("/health").handler(rc ->
        rc.response().end("OK"));
    router.route("/").handler(this::getShoppingList);

    ServiceDiscovery.create(vertx, discovery -> {
        // Get pricer-service
        Single<WebClient> s1 = HttpEndpoint.rxGetWebClient(discovery, svc -> svc.getName().equals("pricer-service"));

        // Get shopping-backend
        Single<WebClient> s2 = HttpEndpoint.rxGetWebClient(discovery, svc -> svc.getName().equals("shopping-backend"));

        Single.zip(s1, s2, (p, s) -> {
            pricer = p;
            shopping = s;

            return vertx.createHttpServer()
                .requestHandler(router::accept)
                .listen(8080);
        }).subscribe();

        // When both are done...

    });
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() {
    circuit = CircuitBreaker.create("circuit-breaker", vertx,
        new CircuitBreakerOptions()
            .setFallbackOnFailure(true)
            .setMaxFailures(3)
            .setResetTimeout(5000)
            .setTimeout(1000)
    );

    Router router = Router.router(vertx);
    router.route("/health").handler(rc -> rc.response().end("OK"));
    router.route("/").handler(this::getShoppingList);

    ServiceDiscovery.create(vertx, discovery -> {
        // Get pricer-service
        Single<WebClient> s1 = HttpEndpoint.rxGetWebClient(discovery,
            svc -> svc.getName().equals("pricer-service"));

        // Get shopping-backend
        Single<WebClient> s2 = HttpEndpoint.rxGetWebClient(discovery,
            svc -> svc.getName().equals("shopping-backend"));

        // When both are done...
        Single.zip(s1, s2, (p, s) -> {
            pricer = p;
            shopping = s;

            return vertx.createHttpServer()
                .requestHandler(router::accept)
                .listen(8080);
        })
            .subscribe();
    });


}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() throws Exception {
  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);

  io.vertx.rxjava.servicediscovery.types.EventBusService.getServiceProxyWithJsonFilter(
    discovery,
    new JsonObject().put("service.interface", io.vertx.servicediscovery.service.HelloService.class.getName()),
    HelloService.class, // service interface
    ar -> {
      if (ar.failed()) {
        vertx.eventBus().send("result",
          new JsonObject().put("status", "ko").put("message", ar.cause().getMessage()));
      } else {
        HelloService hello = ar.result();
        hello.hello(new JsonObject().put("name", "vert.x"), result -> {
          if (result.failed()) {
            vertx.eventBus().send("result", new JsonObject()
              .put("status", "ko")
              .put("message", result.cause().getMessage()));
          } else {
            vertx.eventBus().send("result", new JsonObject()
              .put("status", "ok")
              .put("message", result.result()));

            ServiceDiscovery.releaseServiceObject(discovery, hello);
          }
        });
      }
    });
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
private synchronized JsonArray getBindings(ServiceDiscovery discovery) {
  JsonArray array = new JsonArray();
  for (ServiceReference ref : discovery.bindings()) {
    array.add(ref.toString());
  }
  return array;
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() throws Exception {
  discovery = ServiceDiscovery.create(vertx, new ServiceDiscoveryOptions().setBackendConfiguration(config()));
  JsonObject cbOptions = config().getJsonObject("circuit-breaker") != null ?
    config().getJsonObject("circuit-breaker") : new JsonObject();
  circuitBreaker = CircuitBreaker.create(cbOptions.getString("name", "circuit-breaker"), vertx,
    new CircuitBreakerOptions()
      .setMaxFailures(cbOptions.getInteger("maxFailures", 5))
      .setTimeout(cbOptions.getLong("timeout", 10000L))
      .setFallbackOnFailure(true)
      .setResetTimeout(cbOptions.getLong("resetTimeout", 30000L))
  );
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Discover all published services by name
 * @param name the service name
 * @param discovery the service discovery that should be used to discover the
 * services
 * @param vertx the Vert.x instance
 * @return all services published under the given the name
 * @see #publishOnce(Record, ServiceDiscovery, Vertx)
 */
static Observable<Service> discover(String name, ServiceDiscovery discovery,
    Vertx vertx) {
  return discovery.rxGetRecords(r -> r.getName().equals(name))
      .flatMapObservable(Observable::from)
      .map(record -> {
        String endpoint = record.getLocation().getString(Record.ENDPOINT);
        Service s = new DefaultService(name, endpoint,
            record.getRegistration(), vertx);
        return s;
      });
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Test if a message can be sent to all consumers
 * @param context the test context
 */
@Test
public void broadcast(TestContext context) {
  Vertx vertx = new Vertx(rule.vertx());

  Async aC = context.async();
  Async bC = context.async();
  Async cC = context.async();
  Async dC = context.async();

  String data = "special data";

  vertx.eventBus().consumer("a", h -> {
    context.assertEquals(data, h.body());
    aC.complete();
  });
  vertx.eventBus().consumer("a", h -> {
    context.assertEquals(data, h.body());
    bC.complete();
  });
  vertx.eventBus().consumer("b", h -> {
    context.assertEquals(data, h.body());
    cC.complete();
  });
  vertx.eventBus().consumer("b", h -> {
    context.assertEquals(data, h.body());
    dC.complete();
  });

  ServiceDiscovery discovery = ServiceDiscovery.create(vertx);
  Service.publishOnce("A", "a", discovery, vertx)
    .flatMap(v -> Service.publishOnce("A", "b", discovery, vertx))
    .flatMapObservable(v -> Service.discover("A", discovery, vertx))
    .doOnTerminate(discovery::close)
    .subscribe(service -> {
      service.broadcast(data);
    }, context::fail);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() throws Exception {
  discovery = ServiceDiscovery.create(vertx, new ServiceDiscoveryOptions().setBackendConfiguration(config()));
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public Single<Void> unpublish(ServiceDiscovery discovery) {
  return discovery.rxUnpublish(registrationId);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
@Override
public void start() {
  super.start();
  vertx = Vertx.newInstance(super.vertx);
  discovery = ServiceDiscovery.newInstance(super.discovery);
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Create a shopping checkout service instance
 * @param vertx 
 * @param discovery 
 * @return 
 */
public static CheckoutService createService(Vertx vertx, ServiceDiscovery discovery) { 
  CheckoutService ret = CheckoutService.newInstance(io.vertx.blueprint.microservice.cart.CheckoutService.createService(vertx.getDelegate(), discovery.getDelegate()));
  return ret;
}
 

import io.vertx.rxjava.servicediscovery.ServiceDiscovery; //导入依赖的package包/类
/**
 * Unpublish this service
 * @param discovery the service discovery where this service is registered
 * @return an single that emits one item when the operation has finished
 */
Single<Void> unpublish(ServiceDiscovery discovery);