Receiver extends AbstractApplicationClient { @PostConstruct Future<CompositeFuture> start() { return clientFactory.connect() .compose(con -> { clientFactory.addReconnectListener(this::createConsumer); return createConsumer(con); }) .onComplete(this::handleCreateConsumerStatus); }    }
@Test public void testTelemetryStart(final VertxTestContext context) { receiver.messageType = "telemetry"; receiver.start().onComplete( context.succeeding(result -> { context.verify(() -> { assertNotNull(result.list()); assertEquals(result.size(), 1); }); context.completeNow(); })); }
@Test public void testEventStart(final VertxTestContext context) { receiver.messageType = "event"; receiver.start().onComplete( context.succeeding(result -> { context.verify(() -> { assertNotNull(result.list()); assertEquals(result.size(), 1); }); context.completeNow(); })); }
@Test public void testDefaultStart(final VertxTestContext context) { receiver.messageType = "all"; receiver.start().onComplete( context.succeeding(result -> { context.verify(() -> { assertNotNull(result.list()); assertEquals(result.size(), 2); }); context.completeNow(); })); }
@Test public void testInvalidTypeStart(final VertxTestContext context) { receiver.messageType = "xxxxx"; receiver.start().onComplete( context.failing(result -> context.completeNow())); }