EventConsumerImpl extends AbstractConsumer implements MessageConsumer { public static Future<MessageConsumer> create( final HonoConnection con, final String tenantId, final BiConsumer<ProtonDelivery, Message> eventConsumer, final Handler<String> closeHook) { return create(con, tenantId, eventConsumer, true, closeHook); } private  EventConsumerImpl(final HonoConnection connection, final ProtonReceiver receiver); static Future<MessageConsumer> create(
            final HonoConnection con,
            final String tenantId,
            final BiConsumer<ProtonDelivery, Message> eventConsumer,
            final Handler<String> closeHook); static Future<MessageConsumer> create(
            final HonoConnection con,
            final String tenantId,
            final BiConsumer<ProtonDelivery, Message> eventConsumer,
            final boolean autoAccept,
            final Handler<String> closeHook);  }
@Test public void testCreateRegistersBiConsumerAsMessageHandler(final VertxTestContext ctx) { final BiConsumer<ProtonDelivery, Message> eventConsumer = (delivery, message) -> { ProtonHelper.released(delivery, true); }; final Source source = mock(Source.class); when(source.getAddress()).thenReturn("event/tenant"); final ProtonReceiver receiver = mock(ProtonReceiver.class); when(receiver.getSource()).thenReturn(source); when(receiver.getRemoteSource()).thenReturn(source); when(receiver.getRemoteQoS()).thenReturn(ProtonQoS.AT_LEAST_ONCE); when(connection.createReceiver( anyString(), any(ProtonQoS.class), any(ProtonMessageHandler.class), anyInt(), anyBoolean(), VertxMockSupport.anyHandler())).thenReturn(Future.succeededFuture(receiver)); EventConsumerImpl.create( connection, "tenant", eventConsumer, remoteDetach -> {}) .onComplete(ctx.succeeding(c -> { final ArgumentCaptor<ProtonMessageHandler> messageHandler = ArgumentCaptor.forClass(ProtonMessageHandler.class); ctx.verify(() -> verify(connection).createReceiver( eq("event/tenant"), eq(ProtonQoS.AT_LEAST_ONCE), messageHandler.capture(), anyInt(), eq(true), VertxMockSupport.anyHandler())); final ProtonDelivery delivery = mock(ProtonDelivery.class); final Message msg = mock(Message.class); messageHandler.getValue().handle(delivery, msg); ctx.verify(() -> verify(delivery).disposition(any(Released.class), eq(Boolean.TRUE))); ctx.completeNow(); })); }