Java 类io.vertx.core.AbstractVerticle 实例源码

public static <VERTICLE extends AbstractVerticle> boolean blockDeploy(Vertx vertx,
    Class<VERTICLE> cls,
    DeploymentOptions options) throws InterruptedException {
  Holder<Boolean> result = new Holder<>();

  CountDownLatch latch = new CountDownLatch(1);
  vertx.deployVerticle(cls.getName(), options, ar -> {
    result.value = ar.succeeded();

    if (ar.failed()) {
      LOGGER.error("deploy vertx failed, cause ", ar.cause());
    }

    latch.countDown();
  });

  latch.await();

  return result.value;
}

@Test
public void testInit() {
  boolean status = false;
  try {
    new MockUp<VertxUtils>() {
      @Mock
      public Vertx init(VertxOptions vertxOptions) {
        return null;
      }

      @Mock
      public <VERTICLE extends AbstractVerticle> boolean blockDeploy(Vertx vertx, Class<VERTICLE> cls,
          DeploymentOptions options) throws InterruptedException {
        return true;
      }
    };
    instance.init();
  } catch (Exception e) {
    status = true;
  }
  Assert.assertFalse(status);
}

@Test
// Regression test for ISS-73: undeployment of a verticle with unassigned consumer fails
public void testUndeployUnassignedConsumer(TestContext ctx) throws Exception {
  Properties config = kafkaCluster.useTo().getConsumerProperties("testUndeployUnassignedConsumer_consumer",
    "testUndeployUnassignedConsumer_consumer", OffsetResetStrategy.EARLIEST);
  config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

  Async async = ctx.async(1);
  vertx.deployVerticle(new AbstractVerticle() {
    @Override
    public void start() throws Exception {
      KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config);
      vertx.setTimer(20, record -> {
        // Very rarely, this throws a AlreadyUndedeployed error
        vertx.undeploy(context.deploymentID(), ctx.asyncAssertSuccess(ar -> {
          async.complete();
        }));
      });
    }
  }, ctx.asyncAssertSuccess());
}

/**
 * Deploy the given verticle.
 * 
 * @param vertx
 *            Vertex instance which should be deployed into
 * @param config
 *            Verticle configuration
 * @param verticle
 *            Verticle which should be deployed
 * @param worker
 *            Flag which indicates whether the verticle should be deployed as worker verticle
 * @return Deployment Id
 * @throws Exception
 */
public static String deployAndWait(Vertx vertx, JsonObject config, AbstractVerticle verticle, boolean worker)
        throws Exception {
    CompletableFuture<String> fut = new CompletableFuture<>();
    DeploymentOptions options = new DeploymentOptions();
    if (config != null) {
        options = new DeploymentOptions(new JsonObject().put("config", config));
    }
    options.setWorker(worker);
    vertx.deployVerticle(verticle, options, handler -> {
        if (handler.succeeded()) {
            String deploymentId = handler.result();
            if (log.isInfoEnabled()) {
                log.info("Deployed verticle {" + verticle.getClass().getName() + "} => " + deploymentId);
            }
            fut.complete(deploymentId);
        } else {
            Throwable error = handler.cause();
            log.error("Error:", error);
            fut.completeExceptionally(error);
        }
    });
    return fut.get(DEFAULT_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
}

/**
 * Executes the postConstruct using Reflection. This solves the issue that you can extend from a
 * VxmsEndpoint or use the static invocation of an AbstractVerticle.
 *
 * @param router the http router handler
 * @param startFuture the vert.x start future
 * @param registrationObject the object to execute postConstruct
 */
private static void executePostConstruct(AbstractVerticle registrationObject, Router router,
    final Future<Void> startFuture) {
  final Stream<ReflectionExecutionWrapper> reflectionExecutionWrapperStream = Stream
      .of(new ReflectionExecutionWrapper("postConstruct",
              registrationObject, new Object[]{router, startFuture}, startFuture),
          new ReflectionExecutionWrapper("postConstruct",
              registrationObject, new Object[]{startFuture, router}, startFuture),
          new ReflectionExecutionWrapper("postConstruct",
              registrationObject, new Object[]{startFuture}, startFuture));
  final Optional<ReflectionExecutionWrapper> methodWrapperToInvoke = reflectionExecutionWrapperStream
      .filter(ReflectionExecutionWrapper::isPresent).findFirst();
  methodWrapperToInvoke.ifPresent(ReflectionExecutionWrapper::invoke);
  if(!methodWrapperToInvoke.isPresent() && !startFuture.isComplete()) {
    startFuture.complete();
  }

}

/**
 * Resolves discovery annotation in AbstractVerticles
 *
 * @param service the service where to resolve the annotations
 * @param kubeConfig the kubernetes config
 */
public static void resolveBeanAnnotations(AbstractVerticle service, Config kubeConfig) {
  final JsonObject env = service.config();
  final List<Field> serviceNameFields = findServiceFields(service);
  if (!env.getBoolean("kube.offline", false)) { // online
    final DefaultKubernetesClient client =
        new DefaultKubernetesClient(kubeConfig); // TODO be aware of OpenShiftClient
    if (!serviceNameFields.isEmpty()) {
      findServiceEntryAndSetValue(service, serviceNameFields, env, client);
    } else {
      // TODO check and handle Endpoints & Pods
    }
  } else {
    // resolve properties offline
    if (!serviceNameFields.isEmpty()) {
      resolveServicesOffline(service, serviceNameFields, env);
    } else {
      // TODO check and handle Endpoints & Pods
    }
  }
}

@Source(translate = false)
public static void asyncAssertSuccess_02(Vertx vertx, TestContext context) {
  AtomicBoolean started = new AtomicBoolean();
  Async async = context.async();
  vertx.deployVerticle(new AbstractVerticle() {
    public void start() throws Exception {
      started.set(true);
    }
  }, ar -> {
    if (ar.succeeded()) {
      context.assertTrue(started.get());
      async.complete();
    } else {
      context.fail(ar.cause());
    }
  });

  // Can be replaced by

  vertx.deployVerticle("my.verticle", context.asyncAssertSuccess(id -> {
    context.assertTrue(started.get());
  }));
}

/**
 * @return the Java {@link Verticle} adapter for this Groovy Verticle
 */
public Verticle asJavaVerticle() {
  return new AbstractVerticle() {

    @Override
    public void start(Future<Void> startFuture) throws Exception {
      GroovyVerticle.this.vertx = super.vertx;
      GroovyVerticle.this.context = super.context;
      GroovyVerticle.this.start(startFuture);
    }

    @Override
    public void stop(Future<Void> stopFuture) throws Exception {
      GroovyVerticle.this.stop(stopFuture);
    }
  };
}

public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    NetworkUtils.init();
    System.out.println("===================Test start===================");
    vertx.deployVerticle(new AbstractVerticle() {
        @Override
        public void start() throws Exception {
            NetworkUtils.asyncPostString("http://breo.turing.asia/gzwx/smartBox/poll", System.out::println);
        }
    });

}

@Test
public void builderWithConfig(){
    RestKiqrServerVerticle.Builder builder = RestKiqrServerVerticle.Builder.serverBuilder(new KStreamBuilder(), new Properties());

    AbstractVerticle serverVerticle = builder.withOptions(new HttpServerOptions().setPort(4711)).build();
    assertThat(serverVerticle, is(instanceOf(RestKiqrServerVerticle.class)));

    RestKiqrServerVerticle server = (RestKiqrServerVerticle) serverVerticle;
    assertThat(server.serverOptions.getPort(), is(equalTo(4711)));

}

@Test
public void builderWithPort(){
    RestKiqrServerVerticle.Builder builder = RestKiqrServerVerticle.Builder.serverBuilder(new KStreamBuilder(), new Properties());

    AbstractVerticle serverVerticle = builder.withPort(4711).build();
    assertThat(serverVerticle, is(instanceOf(RestKiqrServerVerticle.class)));

    RestKiqrServerVerticle server = (RestKiqrServerVerticle) serverVerticle;
    assertThat(server.serverOptions.getPort(), is(equalTo(4711)));

}

@Test
public void testCleanupInProducer(TestContext ctx) throws Exception {
  Properties config = kafkaCluster.useTo().getProducerProperties("the_producer");
  config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

  AtomicReference<KafkaProducer<String, String>> producerRef = new AtomicReference<>();
  AtomicReference<String> deploymentRef = new AtomicReference<>();
  vertx.deployVerticle(new AbstractVerticle() {
    @Override
    public void start() throws Exception {
      KafkaProducer<String, String> producer = KafkaProducer.create(vertx, config);
      producerRef.set(producer);
      producer.write(KafkaProducerRecord.create("the_topic", "the_value"), ctx.asyncAssertSuccess());
    }
  }, ctx.asyncAssertSuccess(deploymentRef::set));

  Async async = ctx.async();

  kafkaCluster.useTo().consumeStrings("the_topic", 1, 10, TimeUnit.SECONDS, () -> {
    vertx.undeploy(deploymentRef.get(), ctx.asyncAssertSuccess(v -> {
      Thread.getAllStackTraces().forEach((t, s) -> {
        if (t.getName().contains("kafka-producer-network-thread")) {
          ctx.fail("Was expecting the producer to be closed");
        }
      });
      async.complete();
    }));
  });
}

@Test
public void testCleanupInConsumer(TestContext ctx) throws Exception {
  String topicName = "testCleanupInConsumer";
  Properties config = kafkaCluster.useTo().getConsumerProperties("testCleanupInConsumer_consumer",
    "testCleanupInConsumer_consumer", OffsetResetStrategy.EARLIEST);
  config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

  Async async = ctx.async(2);
  Async produceLatch = ctx.async();
  vertx.deployVerticle(new AbstractVerticle() {
    @Override
    public void start(Future<Void> fut) throws Exception {
      KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config);
      consumer.handler(record -> {
        // Very rarely, this throws a AlreadyUndedeployed error
        vertx.undeploy(context.deploymentID(), ctx.asyncAssertSuccess(ar -> {
          try {
            // Race condition? Without a sleep, test fails sometimes
            Thread.sleep(10);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          Thread.getAllStackTraces().forEach((t, s) -> {
            if (t.getName().contains("vert.x-kafka-consumer-thread")) {
              ctx.fail("Was expecting the consumer to be closed");
            }
          });
          async.countDown();
        }));
      });
      consumer.assign(new TopicPartition(topicName, 0), fut);
    }
  }, ctx.asyncAssertSuccess(v ->  produceLatch.complete()
  ));
  produceLatch.awaitSuccess(10000);
  kafkaCluster.useTo().produce("testCleanupInConsumer_producer", 100,
    new StringSerializer(), new StringSerializer(), async::countDown,
    () -> new ProducerRecord<>(topicName, "the_value"));
}

public ClusterDiscovery(String name, List<ApiDefinition> definitions, AtomicInteger seq) {
  vertx
          .deployVerticle(new AbstractVerticle() {
            @Override
            public void start() throws Exception {
              apiDiscovery = ApiDiscovery.create(vertx, new ApiDiscoveryOptions().setName(name));
              for (ApiDefinition definition : definitions) {
                apiDiscovery.publish(definition,
                                     ar -> seq.incrementAndGet());
              }
            }
          });
}

private static AbstractVerticle fakeEurekaVerticle() {
  return new AbstractVerticle() {

    @Override
    public void start(Future<Void> startFuture) throws Exception {
      final Router router = Router.router(vertx);

      router.get("/eureka/v2/apps").handler(context -> {
        final HttpServerResponse response = context.response();

        vertx.fileSystem()
             .readFile("eureka-data.xml", res -> {
               if (res.succeeded()) {
                 response.putHeader(HttpHeaders.CONTENT_TYPE, "application/xml")
                         .end(res.result());
               } else {
                 response.setStatusCode(500)
                         .end("Error while reading eureka data from classpath");
               }
             });
      });

      vertx.createHttpServer(new HttpServerOptions())
           .requestHandler(router::accept)
           .listen(FAKE_EUREKA_SERVER_PORT, res -> {
             if (res.succeeded()) {
               startFuture.complete();
             } else {
               startFuture.fail(res.cause());
             }
           });
    }
  };
}

/**
 * Initialize an existing Vert.x instance as a Vxms endpoint so you can use an Verticle (Extending
 * AbstractVerticle) as a fully functional Vxms Endpoint). Caution: this start methods completes
 * the start future when it's ready, so it must be the last initialisation (order and callback
 * wise)
 * @param startFuture the Start Future from Vert.x
 * @param registrationObject the AbstractVerticle to register (mostly this)
 */
public static void start(final Future<Void> startFuture, AbstractVerticle registrationObject) {
  final Vertx vertx = registrationObject.getVertx();
  final JsonObject config = vertx.getOrCreateContext().config();
  vertx.eventBus()
      .consumer(ConfigurationUtil.getServiceName(config, registrationObject.getClass()) + "-info",
          VxmsEndpoint::info);
  initEndpoint(startFuture, registrationObject, new VxmsShared(vertx, new LocalData(vertx)));
}

private static void initExtensions(HttpServer server, Router router,
    AbstractVerticle registrationObject, VxmsShared vxmsShared) {
  initDiscoveryxtensions(registrationObject);
  initWebSocketExtensions(server, registrationObject,vxmsShared);
  initRESTExtensions(router, registrationObject, vxmsShared);
  initEventBusExtensions(registrationObject, vxmsShared);
}

/**
 * starts the HTTP Endpoint
 *
 * @param startFuture the vertx start future
 * @param server the vertx server
 * @param topRouter the router object
 */
private static void initHTTPEndpoint(AbstractVerticle registrationObject,
    Future<Void> startFuture,HttpServer server,
    Router topRouter) {
  server.requestHandler(topRouter::accept).listen(status -> {
    if (status.succeeded()) {
      executePostConstruct(registrationObject, topRouter, startFuture);
      startFuture.setHandler(result -> logStartfuture(startFuture));
    } else {
      startFuture.fail(status.cause());
      startFuture.setHandler(result -> logStartfuture(startFuture));
    }
  });

}

private static void initRESTExtensions(Router router, AbstractVerticle registrationObject,
    VxmsShared vxmsShared) {
  // check for REST extension
  Optional.
      ofNullable(getRESTSPI()).
      ifPresent(resthandlerSPI -> {
        resthandlerSPI
            .initRESTHandler(vxmsShared, router, registrationObject);
        log("start REST extension");
      });
}

private static void initEventBusExtensions(AbstractVerticle registrationObject,
    VxmsShared vxmsShared) {
  // check for REST extension
  Optional.
      ofNullable(getEventBusSPI()).
      ifPresent(eventbusHandlerSPI -> {
        eventbusHandlerSPI
            .initEventHandler(vxmsShared, registrationObject);
        log("start event-bus extension");
      });
}

private static void initWebSocketExtensions(HttpServer server,
    AbstractVerticle registrationObject, VxmsShared vxmsShared) {
  // check for websocket extension
  final Vertx vertx = vxmsShared.getVertx();
  final JsonObject config = vertx.getOrCreateContext().config();
  Optional.
      ofNullable(getWebSocketSPI()).
      ifPresent(webSockethandlerSPI -> {
        webSockethandlerSPI
            .registerWebSocketHandler(server, vertx, config, registrationObject);
        log("start websocket extension");
      });
}

private static void initDiscoveryxtensions(AbstractVerticle registrationObject) {
  // check for REST extension
  Optional.
      ofNullable(getDiscoverySPI()).
      ifPresent(discoverySPI -> {
       discoverySPI.initDiscovery(registrationObject);
        log("start discovery extension");
      });
}

@Override
public void initDiscovery(AbstractVerticle service) {
  final JsonObject config = service.config();
  final Vertx vertx = service.getVertx();
  if (!service.getClass().isAnnotationPresent(K8SDiscovery.class))
    return;
  final K8SDiscovery annotation = service.getClass().getAnnotation(K8SDiscovery.class);
  final String customClientClassName =
      ConfigurationUtil.getStringConfiguration(
          config,
          CUSTOM_CLIENT_CONFIGURATION,
          annotation.customClientConfiguration().getCanonicalName());
  final CustomClientConfig custConf = getCustomConfiguration(customClientClassName);
  final Config customConfiguration = custConf.createCustomConfiguration(vertx);
  if (customConfiguration == null) {
    final String master_url =
        ConfigurationUtil.getStringConfiguration(config, MASTER_URL, annotation.master_url());
    final String namespace =
        ConfigurationUtil.getStringConfiguration(config, NAMESPACE, annotation.namespace());
    final Config kubeConfig =
        new ConfigBuilder().withMasterUrl(master_url).withNamespace(namespace).build();

    updateKubeConfig(kubeConfig, config, annotation);
    // 1.) Check from K8SDiscovery Annotation
    // 1.1) read properties and from Annotation or from configuration
    // 2.) init KubernetesClient
    KubeDiscovery.resolveBeanAnnotations(service, kubeConfig);
  } else {
    updateKubeConfig(customConfiguration, config, annotation);
    KubeDiscovery.resolveBeanAnnotations(service, customConfiguration);
  }
}

public void initDiscovery(AbstractVerticle service, Config kubeConfig) {
  final JsonObject config = service.config();
  if (!service.getClass().isAnnotationPresent(K8SDiscovery.class))
    return;
  final K8SDiscovery annotation = service.getClass().getAnnotation(K8SDiscovery.class);
  updateKubeConfig(kubeConfig, config, annotation);
  // 1.) Check from K8SDiscovery Annotation
  // 1.1) read properties and from Annotation or from configuration
  // 2.) init KubernetesClient
  KubeDiscovery.resolveBeanAnnotations(service, kubeConfig);
}

@Override
public CommandRegistry registerCommands(List<Command> commands, Handler<AsyncResult<List<Command>>> doneHandler) {
  if (closed) {
    throw new IllegalStateException();
  }
  vertx.deployVerticle(new AbstractVerticle() {

    @Override
    public void start() throws Exception {
      Map<String, CommandRegistration> newReg = new HashMap<>();
      for (Command command : commands) {
        String name = command.name();
        if (commandMap.containsKey(name)) {
          throw new Exception("Command " + name + " already registered");
        }
        CommandRegistration registration = new CommandRegistration(command, deploymentID());
        newReg.put(name, registration);
      }
      commandMap.putAll(newReg);
    }

    @Override
    public void stop() throws Exception {
      String deploymentId = deploymentID();
      commandMap.values().removeIf(reg -> deploymentId.equals(reg.deploymendID));
    }
  }, ar -> {
    if (ar.succeeded()) {
      List<Command> regs = commandMap.values().
          stream().
          filter(reg -> ar.result().equals(reg.deploymendID)).
          map(reg -> reg.command).
          collect(Collectors.toList());
      doneHandler.handle(Future.succeededFuture(regs));
    } else {
      doneHandler.handle(Future.failedFuture(ar.cause()));
    }
  });
  return this;
}

@Test
public void testUndeployInVerticleContext(TestContext context) {
  CommandRegistry registry = CommandRegistry.getShared(vertx);
  Async async = context.async();
  AtomicReference<String> ref = new AtomicReference<>();
  vertx.deployVerticle(new AbstractVerticle() {
    @Override
    public void start(Future<Void> startFuture) throws Exception {
      CommandBuilder command = CommandBuilder.command("hello");
      command.processHandler(process -> {
      });
      registry.registerCommand(command.build(vertx), ar -> {
        if (ar.succeeded()) {
          startFuture.complete();
        } else {
          startFuture.fail(ar.cause());
        }
      });
    }
  }, context.asyncAssertSuccess(id -> {
    ref.set(id);
    async.complete();
  }));
  async.awaitSuccess(5000);
  vertx.undeploy(ref.get(), context.asyncAssertSuccess(v -> {
    context.assertEquals(Collections.emptyList(), registry.commands());
  }));
}

/**
 * Kicks off simple server generation.
 * @param tables the cassandra table meta data
 * @throws IOException thrown if we fail to write out to disk
 */
public static void generate(Collection<TableMetadata> tables) throws IOException {
    String namespaceToUse = MetaData.instance.getNamespace();

    TypeSpec.Builder serverBuilder = TypeSpec.classBuilder("Server")
            .addModifiers(Modifier.PUBLIC);

    serverBuilder.superclass(AbstractVerticle.class);

    addMemberVars(namespaceToUse, serverBuilder);

    serverBuilder.addMethod(getStartupMethod());
    serverBuilder.addMethod(getIsInitializerFunc());
    serverBuilder.addMethod(getBuildApi(tables));
    serverBuilder.addMethod(getStartServer());
    serverBuilder.addMethod(getStopMethod());

    serverBuilder.addJavadoc(GeneratorHelper.getJavaDocHeader("Simple server that registers all {@link " +
            ClassName.get(RestApi.class) + "} for CRUD operations." +
            "\n\nto build: ./gradlew clean shadowJar\n" +
            "to run: java -jar build/libs/[project-name]-fat.jar -conf [your_conf.json]", MetaData.instance.getUpdateTime()));

    JavaFile javaFile = JavaFile.builder(namespaceToUse, serverBuilder.build()).build();

    Disk.outputFile(javaFile);

    //setup the logback file the server needs to run
    Disk.outputFile(LogbackTemplate.TEMPLATE, "src/main/resources/logback.xml");
    // output a default conf file
    Disk.outputFile(ConfTemplate.TEMPLATE, "conf.json");
}

@Test
public void testVerticleMetrics() throws Exception {
  int verticles = 5;
  CountDownLatch latch = new CountDownLatch(verticles);
  AtomicReference<String> ref = new AtomicReference<>();
  for (int i = 0; i < 5; i++) {
    vertx.deployVerticle(new AbstractVerticle() {}, ar -> {
      assertTrue(ar.succeeded());
      ref.set(ar.result()); // just use the last deployment id to test undeploy metrics below
      latch.countDown();
    });
  }

  awaitLatch(latch);

  JsonObject metrics = metricsService.getMetricsSnapshot(vertx);
  assertNotNull(metrics);
  assertFalse(metrics.isEmpty());

  assertCount(metrics.getJsonObject("vertx.verticles"), (long) verticles);

  vertx.undeploy(ref.get(), ar -> {
    assertTrue(ar.succeeded());
    assertCount(metricsService.getMetricsSnapshot(vertx).getJsonObject("vertx.verticles"), (long) verticles - 1);
    testComplete();
  });

  await();
}

/**
 * Initialize a Spring Context for given Verticle instance. A Verticle MUST be annotated with {@link SpringVerticle}
 * @param verticle The Verticle Instance where to start the Spring Context
 */
public static void initSpring(AbstractVerticle verticle) {
    final Class<?> currentVerticleClass = verticle.getClass();
    final Vertx vertx = verticle.getVertx();
    if(!currentVerticleClass.isAnnotationPresent(SpringVerticle.class)) {
        throw new InvalidParameterException("no @SpringVerticle annotation found");
    }
    final SpringVerticle annotation = currentVerticleClass.getAnnotation(SpringVerticle.class);
    final Class<?> springConfigClass = annotation.springConfig();

    // Create the parent context
    final GenericApplicationContext genericApplicationContext = getGenericApplicationContext(
        vertx.getClass().getClassLoader());

    // 1. Create a new context for each verticle and use the specified spring configuration class if possible
    AnnotationConfigApplicationContext annotationConfigApplicationContext = getAnnotationConfigApplicationContext(
        springConfigClass, genericApplicationContext);

    // 2. Register the Vertx instance as a singleton in spring context
    annotationConfigApplicationContext.getBeanFactory().registerSingleton(vertx.getClass().getSimpleName(), vertx);

    // 3. Register a bean definition for this verticle
    annotationConfigApplicationContext.getBeanFactory().registerSingleton(verticle.getClass().getSimpleName(), verticle);

    // 4. Add a bean factory post processor to avoid configuration issues
    addPostprocessorAndUpdateContext(currentVerticleClass, annotationConfigApplicationContext);

    // 5. perform autowiring
    annotationConfigApplicationContext.getAutowireCapableBeanFactory().autowireBean(verticle);

}

public static <T extends AbstractVerticle> void deployVerticle(Vertx vertx, Class<T> cls, int instanceCount) {
  DeploymentOptions options = new DeploymentOptions().setInstances(instanceCount);

  vertx.deployVerticle(cls.getName(), options);
}

@Override
public AbstractVerticle build() {
    AbstractVerticle runtimeVerticle = super.build();
    return new RestKiqrServerVerticle(httpServerOptions != null ? httpServerOptions : new HttpServerOptions(), runtimeVerticle);
}

protected RestKiqrServerVerticle(HttpServerOptions serverOptions, AbstractVerticle runtimeVerticle) {
    this.serverOptions = serverOptions;
    this.runtimeVerticle = runtimeVerticle;
}

public MockedRuntimeHttpServerVerticle(HttpServerOptions serverOptions, AbstractVerticle runtimeVerticle) {
    super(serverOptions, runtimeVerticle);
}

private Future deployVerticles(AbstractVerticle... verticles) {


        Stream<AbstractVerticle> stream = Arrays.stream(verticles);

        List<Future> futures = stream
                .map(verticle -> {

                    Future<String> future = Future.<String>future();

                    vertx.deployVerticle(verticle, future.completer());

                    return future;
                })
                .map(future -> (Future) future)
                .collect(toList());

        return CompositeFuture.all(futures);

    }

/**
 * Return a Map of mandatory verticles.
 * 
 * @return
 */
private List<AbstractVerticle> getMandatoryVerticleClasses() {
    List<AbstractVerticle> verticles = new ArrayList<>();
    verticles.add(restVerticle);
    return verticles;
}

/**
 * Get the map of mandatory worker verticle classes
 * 
 * @return
 */
private List<AbstractVerticle> getMandatoryWorkerVerticleClasses() {
    List<AbstractVerticle> verticles = new ArrayList<>();
    verticles.add(jobWorkerVerticle);
    return verticles;
}

@Override
public void initEventHandler(VxmsShared vxmsShared, AbstractVerticle service) {
  EventInitializer.initEventbusHandling(vxmsShared, service);
}

void registerWebSocketHandler(HttpServer server, Vertx vertx, JsonObject config,
AbstractVerticle service);

private static void initNoHTTPEndpoint(AbstractVerticle registrationObject, Future<Void> startFuture,
    Router topRouter) {
  startFuture.setHandler(result -> logStartfuture(startFuture));
  executePostConstruct(registrationObject, topRouter, startFuture);
}

@Override
public void initRESTHandler(VxmsShared vxmsShared, Router router, AbstractVerticle service) {
  RESTInitializer.initRESTHandler(vxmsShared, router, service);
}