### Question:
ApplicationInactivityListener { @EventListener public void on(TopologyEvents.ApplicationInactivityTimeout evt) { String clientId = evt.clientId(); String context = evt.clientStreamIdentification().getContext(); streamClosers.forEach((connectionType, streamCloser) -> { Set<String> streamIds = clientIdRegistry.streamIdsFor(clientId, connectionType); for (String streamId : streamIds) { ClientStreamIdentification streamIdentification = new ClientStreamIdentification(context, streamId); streamCloser.forceDisconnection(clientId, streamIdentification); } }); } @Autowired  ApplicationInactivityListener(ClientIdRegistry clientIdRegistry,
                                         CommandService commandService,
                                         QueryService queryService);  ApplicationInactivityListener(Map<ConnectionType, StreamCloser> streamClosers,
                                         ClientIdRegistry clientIdRegistry); @EventListener void on(TopologyEvents.ApplicationInactivityTimeout evt);  }### Answer:
@Test public void testCloseStreams() { ClientIdRegistry registry = new DefaultClientIdRegistry(); registry.register(commandStream, clientId, COMMAND); registry.register(queryStream, clientId, QUERY); AtomicBoolean commandStreamClosed = new AtomicBoolean(); AtomicBoolean queryStreamClosed = new AtomicBoolean(); Map<ConnectionType, StreamCloser> streamClosers = new HashMap<>(); streamClosers.put(COMMAND, ((client, streamIdentification) -> commandStreamClosed.set(true))); streamClosers.put(QUERY, ((client, streamIdentification) -> queryStreamClosed.set(true))); ApplicationInactivityListener testSubject = new ApplicationInactivityListener(streamClosers, registry); ClientStreamIdentification platformStream = new ClientStreamIdentification(context, this.platformStream); testSubject.on(new TopologyEvents.ApplicationInactivityTimeout(platformStream, component, clientId)); assertTrue(commandStreamClosed.get()); assertTrue(queryStreamClosed.get()); }
@Test public void testCloseCommandStream() { ClientIdRegistry registry = new DefaultClientIdRegistry(); registry.register(commandStream, clientId, COMMAND); AtomicBoolean commandStreamClosed = new AtomicBoolean(); AtomicBoolean queryStreamClosed = new AtomicBoolean(); Map<ConnectionType, StreamCloser> streamClosers = new HashMap<>(); streamClosers.put(COMMAND, ((client, streamIdentification) -> commandStreamClosed.set(true))); streamClosers.put(QUERY, ((client, streamIdentification) -> queryStreamClosed.set(true))); ApplicationInactivityListener testSubject = new ApplicationInactivityListener(streamClosers, registry); ClientStreamIdentification platformStream = new ClientStreamIdentification(context, this.platformStream); testSubject.on(new TopologyEvents.ApplicationInactivityTimeout(platformStream, component, clientId)); assertTrue(commandStreamClosed.get()); assertFalse(queryStreamClosed.get()); }
@Test public void testCloseNoStream() { ClientIdRegistry registry = new DefaultClientIdRegistry(); AtomicBoolean commandStreamClosed = new AtomicBoolean(); AtomicBoolean queryStreamClosed = new AtomicBoolean(); Map<ConnectionType, StreamCloser> streamClosers = new HashMap<>(); streamClosers.put(COMMAND, ((client, streamIdentification) -> commandStreamClosed.set(true))); streamClosers.put(QUERY, ((client, streamIdentification) -> queryStreamClosed.set(true))); ApplicationInactivityListener testSubject = new ApplicationInactivityListener(streamClosers, registry); ClientStreamIdentification platformStream = new ClientStreamIdentification(context, this.platformStream); testSubject.on(new TopologyEvents.ApplicationInactivityTimeout(platformStream, component, clientId)); assertFalse(commandStreamClosed.get()); assertFalse(queryStreamClosed.get()); }