AbstractProtocolAdapterBase extends AbstractServiceBase<T> { @Override protected final Future<Void> startInternal() { final Promise<Void> result = Promise.promise(); if (Strings.isNullOrEmpty(getTypeName())) { result.fail(new IllegalStateException("adapter does not define a typeName")); } else if (tenantClientFactory == null) { result.fail(new IllegalStateException("Tenant client factory must be set")); } else if (downstreamSenderFactory == null) { result.fail(new IllegalStateException("AMQP Messaging Network client must be set")); } else if (registrationClientFactory == null) { result.fail(new IllegalStateException("Device Registration client factory must be set")); } else if (credentialsClientFactory == null) { result.fail(new IllegalStateException("Credentials client factory must be set")); } else if (commandConsumerFactory == null) { result.fail(new IllegalStateException("Command & Control client factory must be set")); } else if (deviceConnectionClientFactory == null) { result.fail(new IllegalStateException("Device Connection client factory must be set")); } else { log.info("using ResourceLimitChecks [{}]", resourceLimitChecks.getClass().getName()); connectToService(tenantClientFactory, "Tenant service"); connectToService(downstreamSenderFactory, "AMQP Messaging Network"); connectToService(registrationClientFactory, "Device Registration service"); connectToService(credentialsClientFactory, "Credentials service"); if (deviceConnectionClientFactory instanceof ConnectionLifecycle) { connectToService((ConnectionLifecycle<?>) deviceConnectionClientFactory, "Device Connection service"); } connectToService( commandConsumerFactory, "Command & Control", this::onCommandConnectionLost, this::onCommandConnectionEstablished) .onComplete(c -> { if (c.succeeded()) { onCommandConnectionEstablished(c.result()); } }); commandTargetMapper.initialize(registrationClientFactory, deviceConnectionClientFactory); commandConsumerFactory.initialize(commandTargetMapper, deviceConnectionClientFactory); doStart(result); } return result.future(); }  @Autowired @Override void setConfig(final T configuration); @Qualifier(TenantConstants.TENANT_ENDPOINT) @Autowired final void setTenantClientFactory(final TenantClientFactory factory); final TenantClientFactory getTenantClientFactory(); @Qualifier(DeviceConnectionConstants.DEVICE_CONNECTION_ENDPOINT) @Autowired final void setDeviceConnectionClientFactory(final BasicDeviceConnectionClientFactory factory); @Deprecated(forRemoval = true) final DeviceConnectionClientFactory getDeviceConnectionClientFactory(); @Qualifier(Constants.QUALIFIER_MESSAGING) @Autowired final void setDownstreamSenderFactory(final DownstreamSenderFactory factory); final DownstreamSenderFactory getDownstreamSenderFactory(); @Qualifier(RegistrationConstants.REGISTRATION_ENDPOINT) @Autowired final void setRegistrationClientFactory(final RegistrationClientFactory factory); final RegistrationClientFactory getRegistrationClientFactory(); @Qualifier(CredentialsConstants.CREDENTIALS_ENDPOINT) @Autowired final void setCredentialsClientFactory(final CredentialsClientFactory factory); final CredentialsClientFactory getCredentialsClientFactory(); @Autowired(required = false) void setConnectionEventProducer(final ConnectionEventProducer connectionEventProducer); ConnectionEventProducer getConnectionEventProducer(); @Autowired final void setCommandConsumerFactory(final ProtocolAdapterCommandConsumerFactory factory); final ProtocolAdapterCommandConsumerFactory getCommandConsumerFactory(); @Autowired final void setCommandTargetMapper(final CommandTargetMapper commandTargetMapper); @Autowired(required = false) final void setResourceLimitChecks(final ResourceLimitChecks resourceLimitChecks); final void setConnectionLimitManager(final ConnectionLimitManager connectionLimitManager); @Override void registerReadinessChecks(final HealthCheckHandler handler); @Override void registerLivenessChecks(final HealthCheckHandler handler); static ErrorCondition getErrorCondition(final Throwable t); static ConnectionAttemptOutcome getOutcome(final Throwable e);  }
@SuppressWarnings("unchecked") @Test public void testStartInternalConnectsToServices(final VertxTestContext ctx) { final Handler<Void> startupHandler = mock(Handler.class); final Handler<Void> commandConnectionHandler = mock(Handler.class); final Handler<Void> commandConnectionLostHandler = mock(Handler.class); givenAnAdapterConfiguredWithServiceClients(startupHandler, commandConnectionHandler, commandConnectionLostHandler); adapter.startInternal().onComplete(ctx.succeeding(ok -> ctx.verify(() -> { verify(tenantClientFactory).connect(); verify(registrationClientFactory).connect(); verify(downstreamSenderFactory).connect(); verify(credentialsClientFactory).connect(); verify(commandConsumerFactory).connect(); verify(commandConsumerFactory).addDisconnectListener(any(DisconnectListener.class)); verify(commandConsumerFactory).addReconnectListener(any(ReconnectListener.class)); verify(startupHandler).handle(null); verify(commandConnectionHandler).handle(null); verify(commandConnectionLostHandler, never()).handle(null); ctx.completeNow(); }))); }
@SuppressWarnings("unchecked") @Test public void testCallbacksInvokedOnReconnect(final VertxTestContext ctx) { final Handler<Void> commandConnectionEstablishedHandler = mock(Handler.class); final Handler<Void> commandConnectionLostHandler = mock(Handler.class); givenAnAdapterConfiguredWithServiceClients(mock(Handler.class), commandConnectionEstablishedHandler, commandConnectionLostHandler); adapter.startInternal().onComplete(ctx.succeeding(ok -> ctx.verify(() -> { final ArgumentCaptor<DisconnectListener<HonoConnection>> disconnectHandlerCaptor = ArgumentCaptor.forClass(DisconnectListener.class); verify(commandConsumerFactory).addDisconnectListener(disconnectHandlerCaptor.capture()); final ArgumentCaptor<ReconnectListener<HonoConnection>> reconnectHandlerCaptor = ArgumentCaptor.forClass(ReconnectListener.class); verify(commandConsumerFactory).addReconnectListener(reconnectHandlerCaptor.capture()); disconnectHandlerCaptor.getValue().onDisconnect(mock(HonoConnection.class)); verify(commandConnectionLostHandler).handle(null); reconnectHandlerCaptor.getValue().onReconnect(mock(HonoConnection.class)); verify(commandConnectionEstablishedHandler, times(2)).handle(null); ctx.completeNow(); }))); }