@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {

    LOGGER.info("WEBSOCKETCHANNELINTERCEPTOR -> "+message.toString());

    StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

    if (accessor != null && StompCommand.CONNECT.equals(accessor.getCommand())) {
        List<String> headers = accessor.getNativeHeader(AUTHORIZATION);
        accessor.setUser(websocketAuthenticationManager.authenticate(new JWTTokenAuthentication(tokenUtils.resolveToken(headers != null ? headers.get(0) : null))));

    }

    return message;
}

	/**
	 * @param session
	 *            - session object
	 * @param closeReason
	 *            - The reason of close of session
	 */
	@OnClose
	public void onClose(Session session, CloseReason closeReason) {
		logger.info("Server: Session " + session.getId() + " closed because of " + closeReason.toString()); 
		
    	// Remove session from the connected sessions set
    	pullConnections.remove(session);
	}

	/**
	 * @param session
	 *            - current session object
	 * @param t
	 *            - Throwable instance containing error info
	 */
	@OnError
	public void onError(Session session, Throwable t) {
		logger.error("Server: Session " + session.getId() + " error " + t.getMessage());
		
    	// Remove session from the connected sessions set
    	pullConnections.remove(session);
	}