### Question:
PipelineVerticle extends ConfigurableVerticle<PipelineOptions> implements SignalEmitter, SignalReceiver { private Completable undeploy() { return vertx.rxUndeploy(pumpDeployment.getValue()) .doOnComplete(() -> info("Shutting down processors")) .andThen(defer(this::undeployProcessors)) .onErrorComplete() .andThen(vertx.rxUndeploy(sinkDeployment.getValue())); }  @Override ValidationResult validate(JsonObject config); @Override PipelineOptions readConfiguration(JsonObject config); @Override Completable configure(PipelineOptions config); @Override Completable shutdown(); @Override String controlChannel(); @Override Completable onSignal(Signal signal); @Override void emitSignal(Signal signal); @Override ExchangeOptions exchange(); @Override EventBus eventBus();  }### Answer:
@Test @DisplayName("PipelineVerticle is able to start simple pipeline") void testPipelineVerticleStartsSimpleFlow(Vertx vertx, VertxTestContext ctx) { DeploymentOptions options = getDeploymentOptions("simple"); vertx.deployVerticle(PIPELINE_VERTICLE, options, ctx.succeeding(id -> vertx.setTimer(2_000L, timer -> vertx.undeploy(id, ctx.succeeding(v -> ctx.completeNow()))) ) ); }
@Test @DisplayName("PipelineVerticle is able to start multi instance / processors with file output in /tmp/output.json") void testPipelineVerticleStartsMultiInstanceMultiProcessorWithFileSink(Vertx vertx, VertxTestContext ctx) throws InterruptedException { DeploymentOptions options = getDeploymentOptions("multi-instance-multi-processor"); vertx.deployVerticle(PIPELINE_VERTICLE, options, ctx.succeeding(id -> vertx.setTimer(2_000L, timer -> vertx.undeploy(id, ctx.succeeding(v -> ctx.completeNow()))) ) ); ctx.awaitCompletion(5, TimeUnit.SECONDS); }
@Test @DisplayName("Logger processors....") void testLoggerProcessors(Vertx vertx, VertxTestContext ctx) throws InterruptedException { DeploymentOptions options = getDeploymentOptions("logger-test"); vertx.deployVerticle(PIPELINE_VERTICLE, options, ctx.succeeding(id -> vertx.setTimer(2_000L, timer -> vertx.undeploy(id, ctx.succeeding(v -> ctx.completeNow()))) ) ); ctx.awaitCompletion(5, TimeUnit.SECONDS); }
@Test @DisplayName("Data extractor maps fields") void testDataExtractorProcessor(Vertx vertx, VertxTestContext ctx) throws InterruptedException { DeploymentOptions options = getDeploymentOptions("data-extractor-test"); vertx.deployVerticle(PIPELINE_VERTICLE, options, ctx.succeeding(id -> vertx.setTimer(2_000L, timer -> vertx.undeploy(id, ctx.succeeding(v -> ctx.completeNow()))) ) ); ctx.awaitCompletion(5, TimeUnit.SECONDS); }