StreamController extends BaseController { @MessageMapping("/consume/{viewId}") @Transactional(readOnly = true) public String newConsumer( @DestinationVariable final Long viewId, final ConsumeRequest consumeRequest, final SimpMessageHeaderAccessor headerAccessor) { final Optional<View> viewOptional = viewRepository.findById(viewId); if (!viewOptional.isPresent()) { return "{success: false}"; } final View view = viewOptional.get(); final long userId = getLoggedInUserId(headerAccessor); final String sessionId = headerAccessor.getSessionId(); final SessionIdentifier sessionIdentifier = SessionIdentifier.newStreamIdentifier(userId, sessionId); final ViewCustomizer viewCustomizer = new ViewCustomizer(view, consumeRequest); viewCustomizer.overrideViewSettings(); final List<FilterDefinition> configuredFilters = viewCustomizer.getFilterDefinitions(); final StartingPosition startingPosition = viewCustomizer.getStartingPosition(); webSocketConsumersManager.addNewConsumer(view, configuredFilters, startingPosition, sessionIdentifier); return "{success: true}"; }  @RequestMapping(path = "", method = RequestMethod.GET) String index(final Model model); @RequestMapping(path = "/{id}", method = RequestMethod.GET) String stream(
        @PathVariable final Long id,
        final Model model,
        final RedirectAttributes redirectAttributes); @MessageMapping("/consume/{viewId}") @Transactional(readOnly = true) String newConsumer(
        @DestinationVariable final Long viewId,
        final ConsumeRequest consumeRequest,
        final SimpMessageHeaderAccessor headerAccessor); @MessageMapping("/pause/{viewId}") @Transactional(readOnly = true) String pauseConsumer(
        @DestinationVariable final Long viewId,
        final SimpMessageHeaderAccessor headerAccessor); @MessageMapping("/resume/{viewId}") @Transactional(readOnly = true) String resumeConsumer(
        @DestinationVariable final Long viewId,
        final SimpMessageHeaderAccessor headerAccessor);  }
@Test public void shouldReceiveAMessageFromTheServer() throws Exception { final String expectedSessionId = "MYSESSIONID"; final String topicName = "TestTopic" + System.currentTimeMillis(); try { Assert.assertEquals("Should have no active consumers", 0, webSocketConsumersManager.countActiveConsumers()); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); final Cluster cluster = clusterTestTools .createCluster("TestCluster", sharedKafkaTestResource.getKafkaConnectString()); final View view = viewTestTools .createViewWithCluster("TestView-" + System.currentTimeMillis(), cluster); assertEquals("Partitions Property should be empty string", "", view.getPartitions()); final ConsumeRequest consumeRequest = new ConsumeRequest(); consumeRequest.setAction("head"); consumeRequest.setPartitions("0,1"); consumeRequest.setFilters(new ArrayList<>()); consumeRequest.setResultsPerPartition(10); final SimpMessageHeaderAccessor mockHeaderAccessor = mock(SimpMessageHeaderAccessor.class); final Authentication mockPrincipal = mock(Authentication.class); when(mockHeaderAccessor.getUser()) .thenReturn(mockPrincipal); when(mockPrincipal.getPrincipal()) .thenReturn(nonAdminUserDetails); when(mockHeaderAccessor.getSessionId()) .thenReturn(expectedSessionId); final String result = streamController.newConsumer( view.getId(), consumeRequest, mockHeaderAccessor ); assertEquals("Should return success response", "{success: true}", result); assertEquals("Should have one active consumer", 1, webSocketConsumersManager.countActiveConsumers()); final View updatedView = viewRepository.findById(view.getId()).get(); assertEquals("Partitions Property should be empty string", "", updatedView.getPartitions()); } finally { webSocketConsumersManager.removeConsumersForSessionId(expectedSessionId); viewTestTools.deleteAllViews(); clusterTestTools.deleteAllClusters(); } }