@Test public void testRemoveConsumersForSessionId() { 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()); manager.removeConsumersForSessionId(sessionId1); verify(mockSocketKafkaConsumer1, times(1)).requestStop(); verify(mockSocketKafkaConsumer2, times(0)).requestStop(); }
public void removeConsumersForSessionId(final String sessionId) { synchronized (consumers) { consumers .entrySet() .stream() .filter((entry) -> entry.getKey().getSessionId().equals(sessionId)) .forEach((entry) -> entry.getValue().requestStop()); } }
WebSocketConsumersManager implements Runnable { public void removeConsumersForSessionId(final String sessionId) { synchronized (consumers) { consumers .entrySet() .stream() .filter((entry) -> entry.getKey().getSessionId().equals(sessionId)) .forEach((entry) -> entry.getValue().requestStop()); } } }
WebSocketConsumersManager implements Runnable { public void removeConsumersForSessionId(final String sessionId) { synchronized (consumers) { consumers .entrySet() .stream() .filter((entry) -> entry.getKey().getSessionId().equals(sessionId)) .forEach((entry) -> entry.getValue().requestStop()); } } WebSocketConsumersManager(
        final WebKafkaConsumerFactory webKafkaConsumerFactory,
        final SimpMessagingTemplate messagingTemplate,
        final int maxConcurrentConsumers
    );  WebSocketConsumersManager(
        final WebKafkaConsumerFactory webKafkaConsumerFactory,
        final SimpMessagingTemplate messagingTemplate,
        final ThreadPoolExecutor threadPoolExecutor
    ); }
WebSocketConsumersManager implements Runnable { public void removeConsumersForSessionId(final String sessionId) { synchronized (consumers) { consumers .entrySet() .stream() .filter((entry) -> entry.getKey().getSessionId().equals(sessionId)) .forEach((entry) -> entry.getValue().requestStop()); } } 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(); }
WebSocketConsumersManager implements Runnable { public void removeConsumersForSessionId(final String sessionId) { synchronized (consumers) { consumers .entrySet() .stream() .filter((entry) -> entry.getKey().getSessionId().equals(sessionId)) .forEach((entry) -> entry.getValue().requestStop()); } } 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();  }