AmqpServiceBase extends AbstractServiceBase<T> { protected void onRemoteConnectionOpen(final ProtonConnection connection) { connection.setContainer(String.format("%s-%s:%d", getServiceName(), getBindAddress(), getPort())); setRemoteConnectionOpenHandler(connection); }  @Autowired @Qualifier(Constants.QUALIFIER_AMQP) @Override void setConfig(final T configuration); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue(); @Autowired(required = false) void addEndpoints(final List<AmqpEndpoint> definedEndpoints); final void addEndpoint(final AmqpEndpoint ep); @Autowired(required = false) void setSaslAuthenticatorFactory(final ProtonSaslAuthenticatorFactory factory); final void setAuthorizationService(final AuthorizationService authService); @Override Future<Void> startInternal(); @Override final Future<Void> stopInternal(); @Override void registerReadinessChecks(final HealthCheckHandler handler); @Override void registerLivenessChecks(final HealthCheckHandler handler);  }
@Test @SuppressWarnings({"rawtypes", "unchecked"}) public void testServerCallsPublishEventOnClientDisconnect() { final Handler<ProtonConnection> publishConnectionClosedEvent = mock(Handler.class); final AmqpServiceBase<ServiceConfigProperties> server = createServer(null, publishConnectionClosedEvent); final ProtonConnection con = newConnection(Constants.PRINCIPAL_ANONYMOUS); server.onRemoteConnectionOpen(con); final ArgumentCaptor<Handler> closeHandlerCaptor = ArgumentCaptor.forClass(Handler.class); verify(con).disconnectHandler(closeHandlerCaptor.capture()); closeHandlerCaptor.getValue().handle(con); verify(publishConnectionClosedEvent).handle(any(ProtonConnection.class)); }