package io.vertx.workshop.common;

import io.vertx.core.*;
import io.vertx.core.impl.ConcurrentHashSet;
import io.vertx.servicediscovery.Record;
import io.vertx.servicediscovery.ServiceDiscovery;
import io.vertx.servicediscovery.ServiceDiscoveryOptions;
import io.vertx.servicediscovery.types.EventBusService;
import io.vertx.servicediscovery.types.MessageSource;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

/**
 * An implementation of {@link Verticle} taking care of the discovery and publication of services.
 *
 * @author <a href="http://escoffier.me">Clement Escoffier</a>
 */
public class MicroServiceVerticle extends AbstractVerticle {

  protected ServiceDiscovery discovery;
  protected Set<Record> registeredRecords = new ConcurrentHashSet<>();

  @Override
  public void start() {
    discovery = ServiceDiscovery.create(vertx, new ServiceDiscoveryOptions().setBackendConfiguration(config()));
  }

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

  public void publishMessageSource(String name, String address, Handler<AsyncResult<Void>>
      completionHandler) {
    Record record = MessageSource.createRecord(name, address);
    publish(completionHandler, record);
  }

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

  private void publish(Handler<AsyncResult<Void>> completionHandler, Record record) {
    if (discovery == null) {
      try {
        start();
      } catch (Exception e) {
        throw new RuntimeException("Cannot create discovery service");
      }
    }

    discovery.publish(record, ar -> {
      if (ar.succeeded()) {
        registeredRecords.add(record);
        completionHandler.handle(Future.succeededFuture());
      } else {
        completionHandler.handle(Future.failedFuture(ar.cause()));
      }
    });
  }

  @Override
  public void stop(Future<Void> future) throws Exception {
    List<Future> futures = new ArrayList<>();
    for (Record record : registeredRecords) {
      Future<Void> unregistrationFuture = Future.future();
      futures.add(unregistrationFuture);
      discovery.unpublish(record.getRegistration(), unregistrationFuture.completer());
    }

    if (futures.isEmpty()) {
      discovery.close();
      future.complete();
    } else {
      CompositeFuture composite = CompositeFuture.all(futures);
      composite.setHandler(ar -> {
        discovery.close();
        if (ar.failed()) {
          future.fail(ar.cause());
        } else {
          future.complete();
        }
      });
    }
  }
}

@@ -303,16 +303,69 @@
 ervice;%0A
+import io.vertx.servicediscovery.types.HttpEndpoint;%0A
 import i
@@ -978,32 +978,271 @@
 onfig()));%0A  %7D%0A%0A
+  public void publishHttpEndpoint(String name, String host, int port, Handler%3CAsyncResult%3CVoid%3E%3E%0A      completionHandler) %7B%0A    Record record = HttpEndpoint.createRecord(name, host, port, %22/%22);%0A    publish(record, completionHandler);%0A  %7D%0A%0A
   public void pu
@@ -1444,32 +1444,40 @@
 s);%0A    publish(
+record, 
 completionHandle
@@ -1473,32 +1473,24 @@
 etionHandler
-, record
 );%0A  %7D%0A%0A  pu
@@ -1670,32 +1670,40 @@
 s);%0A    publish(
+record, 
 completionHandle
@@ -1699,32 +1699,24 @@
 etionHandler
-, record
 );%0A  %7D%0A%0A  pu
@@ -1942,16 +1942,24 @@
 publish(
+record, 
 completi
@@ -1971,16 +1971,8 @@
 dler
-, record
 );%0A 
@@ -1998,16 +1998,31 @@
 publish(
+Record record, 
 Handler%3C
@@ -2061,23 +2061,8 @@
 dler
-, Record record
 ) %7B%0A

Add a method to publish HTTP endpoint