DownstreamSenderFactoryImpl extends AbstractHonoClientFactory implements DownstreamSenderFactory { @Override public final Future<DownstreamSender> getOrCreateTelemetrySender(final String tenantId) { Objects.requireNonNull(tenantId); return connection.isConnected(getDefaultConnectionCheckTimeout()) .compose(v -> connection.executeOnContext(result -> { clientFactory.getOrCreateClient( AddressHelper.getTargetAddress(TelemetryConstants.TELEMETRY_ENDPOINT, tenantId, null, connection.getConfig()), () -> TelemetrySenderImpl.create(connection, tenantId, samplerFactory.create(TelemetryConstants.TELEMETRY_ENDPOINT), onSenderClosed -> { clientFactory.removeClient(AddressHelper.getTargetAddress(TelemetryConstants.TELEMETRY_ENDPOINT, tenantId, null, connection.getConfig())); }), result); })); } DownstreamSenderFactoryImpl(final HonoConnection connection, final SendMessageSampler.Factory samplerFactory); @Override final Future<DownstreamSender> getOrCreateTelemetrySender(final String tenantId); @Override final Future<DownstreamSender> getOrCreateEventSender(final String tenantId);  }
@Test public void testGetTelemetrySenderFailsIfInvokedConcurrently(final VertxTestContext ctx) { final Promise<ProtonSender> sender = Promise.promise(); when(connection.createSender(anyString(), any(ProtonQoS.class), VertxMockSupport.anyHandler())) .thenReturn(sender.future()); final Future<DownstreamSender> result = factory.getOrCreateTelemetrySender("telemetry/tenant"); assertThat(result.isComplete()).isFalse(); factory.getOrCreateTelemetrySender("telemetry/tenant").onComplete(ctx.failing(t -> { ctx.verify(() -> { assertThat(t).isInstanceOf(ServerErrorException.class); verify(vertx, times(CachingClientFactory.MAX_CREATION_RETRIES)).setTimer(anyLong(), notNull()); }); })); sender.complete(mock(ProtonSender.class)); ctx.verify(() -> assertThat(result.isComplete()).isTrue()); ctx.completeNow(); }