WebSocketConsumersManager implements Runnable { public boolean removeConsumersForSessionHash(final String sessionHash) { synchronized (consumers) { for (final Map.Entry<ConsumerKey, ConsumerEntry> entry : consumers.entrySet()) { if (! entry.getKey().getSessionHash().equals(sessionHash)) { continue; } entry.getValue().requestStop(); return true; } } return false; } WebSocketConsumersManager(
        final WebKafkaConsumerFactory webKafkaConsumerFactory,
        final SimpMessagingTemplate messagingTemplate,
        final int maxConcurrentConsumers
    );  WebSocketConsumersManager(
        final WebKafkaConsumerFactory webKafkaConsumerFactory,
        final SimpMessagingTemplate messagingTemplate,
        final ThreadPoolExecutor threadPoolExecutor
    ); void addNewConsumer(
        final View view,
        final Collection<FilterDefinition> filters,
        final StartingPosition startingPosition,
        final SessionIdentifier sessionIdentifier
    ); void removeConsumersForSessionId(final String sessionId); boolean removeConsumersForSessionHash(final String sessionHash); void pauseConsumer(final long viewId, final SessionIdentifier sessionIdentifier); void resumeConsumer(final long viewId, final SessionIdentifier sessionIdentifier); Collection<StreamConsumerDetails> getConsumers(); int countActiveConsumers(); @Override void run();  }
@Test public void testRemoveConsumersForSessionHash() { final long userId = 1212L; final String sessionId1 = "abc123"; final String sessionId2 = "123abc"; final long viewId = 5454L; Assert.assertTrue("Manager should start empty", manager.getConsumers().isEmpty()); final View view = new View(); view.setId(viewId); final Collection<FilterDefinition> filters = new ArrayList<>(); final StartingPosition startingPosition = StartingPosition.newHeadPosition(); final SessionIdentifier sessionIdentifier1 = SessionIdentifier.newStreamIdentifier(userId, sessionId1); final SessionIdentifier sessionIdentifier2 = SessionIdentifier.newStreamIdentifier(userId, sessionId2); final SocketKafkaConsumer mockSocketKafkaConsumer1 = mock(SocketKafkaConsumer.class); final SocketKafkaConsumer mockSocketKafkaConsumer2 = mock(SocketKafkaConsumer.class); when(mockConsumerFactory.createWebSocketClient(view, filters, startingPosition, sessionIdentifier1)) .thenReturn(mockSocketKafkaConsumer1); when(mockConsumerFactory.createWebSocketClient(view, filters, startingPosition, sessionIdentifier2)) .thenReturn(mockSocketKafkaConsumer2); manager.addNewConsumer(view, filters, startingPosition, sessionIdentifier1); manager.addNewConsumer(view, filters, startingPosition, sessionIdentifier2); final Collection<StreamConsumerDetails> consumers = manager.getConsumers(); Assert.assertFalse("Should not be empty", consumers.isEmpty()); Assert.assertEquals(2, consumers.size()); boolean result = manager.removeConsumersForSessionHash(Sha1Tools.sha1(sessionId1)); Assert.assertTrue(result); verify(mockSocketKafkaConsumer1, times(1)).requestStop(); verify(mockSocketKafkaConsumer2, times(0)).requestStop(); result = manager.removeConsumersForSessionHash("made up"); Assert.assertFalse(result); }