EventSenderImpl extends AbstractDownstreamSender { @Override public Future<ProtonDelivery> sendAndWaitForOutcome(final Message message) { return send(message); } protected  EventSenderImpl(
            final HonoConnection con,
            final ProtonSender sender,
            final String tenantId,
            final String targetAddress,
            final SendMessageSampler sampler); @Override String getEndpoint(); static Future<DownstreamSender> create(
            final HonoConnection con,
            final String tenantId,
            final SendMessageSampler sampler,
            final Handler<String> remoteCloseHook); @Override Future<ProtonDelivery> sendAndWaitForOutcome(final Message message); @Override Future<ProtonDelivery> sendAndWaitForOutcome(final Message message, final SpanContext parent);  }
@Test public void testSendAndWaitForOutcomeFailsOnLackOfCredit() { when(sender.sendQueueFull()).thenReturn(Boolean.TRUE); final DownstreamSender messageSender = new EventSenderImpl(connection, sender, "tenant", "event/tenant", SendMessageSampler.noop()); final Message event = ProtonHelper.message("event/tenant", "hello"); final Future<ProtonDelivery> result = messageSender.sendAndWaitForOutcome(event); assertFalse(result.succeeded()); verify(sender, never()).send(any(Message.class), VertxMockSupport.anyHandler()); }
@Test public void testSendAndWaitForOutcomeMarksMessageAsDurable() { when(sender.sendQueueFull()).thenReturn(Boolean.FALSE); final DownstreamSender messageSender = new EventSenderImpl(connection, sender, "tenant", "telemetry/tenant", SendMessageSampler.noop()); when(sender.send(any(Message.class), VertxMockSupport.anyHandler())).thenReturn(mock(ProtonDelivery.class)); final Message msg = ProtonHelper.message("telemetry/tenant/deviceId", "some payload"); messageSender.sendAndWaitForOutcome(msg); verify(sender).send(any(Message.class), VertxMockSupport.anyHandler()); assertTrue(msg.isDurable()); }