class StreamController extends BaseController {

    @MessageMapping("/consume/{viewId}")
    @Transactional(readOnly = true)
    public String newConsumer(
        @DestinationVariable final Long viewId,
        final ConsumeRequest consumeRequest,
        final SimpMessageHeaderAccessor headerAccessor) {

        // Retrieve view
        final Optional<View> viewOptional = viewRepository.findById(viewId);
        if (!viewOptional.isPresent()) {
            return "{success: false}";
        }
        final View view = viewOptional.get();

        // Build a session identifier
        final long userId = getLoggedInUserId(headerAccessor);
        final String sessionId = headerAccessor.getSessionId();
        final SessionIdentifier sessionIdentifier = SessionIdentifier.newStreamIdentifier(userId, sessionId);

        // Override settings
        // TODO View gets flushed and changes are persisted.
        final ViewCustomizer viewCustomizer = new ViewCustomizer(view, consumeRequest);
        viewCustomizer.overrideViewSettings();
        final List<FilterDefinition> configuredFilters = viewCustomizer.getFilterDefinitions();

        // Configure where to start from
        final StartingPosition startingPosition = viewCustomizer.getStartingPosition();

        webSocketConsumersManager.addNewConsumer(view, configuredFilters, startingPosition, sessionIdentifier);
        return "{success: true}";
    }

}