Websockets reverted

application/pom.xml

@@ -126,7 +126,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-webflux</artifactId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
         </dependency>
         <dependency>
             <groupId>io.jsonwebtoken</groupId>

application/src/main/java/org/thingsboard/server/config/ThingsboardWebFluxSecurityConfig.java

@@ -1,60 +0,0 @@
-/**
- * Copyright © 2016-2019 The Thingsboard Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.thingsboard.server.config;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.security.authentication.AuthenticationManager;
-import org.springframework.security.authentication.ReactiveAuthenticationManager;
-import org.springframework.security.config.annotation.method.configuration.EnableReactiveMethodSecurity;
-import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
-import org.springframework.security.config.web.server.SecurityWebFiltersOrder;
-import org.springframework.security.config.web.server.ServerHttpSecurity;
-import org.springframework.security.web.server.SecurityWebFilterChain;
-import org.springframework.security.web.server.context.ServerSecurityContextRepository;
-
-//@EnableWebFluxSecurity
-//@EnableReactiveMethodSecurity
-public class ThingsboardWebFluxSecurityConfig {
-
-    private static final String WS_TOKEN_BASED_AUTH_ENTRY_POINT = "/api/ws/**";
-
-    @Autowired
-    private ReactiveAuthenticationManager webfluxAuthenticationManager;
-
-    @Autowired
-    private ServerSecurityContextRepository jwtTokenSecurityContextRepository;
-
-    @Bean
-    public SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {
-        return http
-                .cors()
-                .and()
-                .csrf().disable()
-                .formLogin().disable()
-                .httpBasic().disable()
-                .exceptionHandling()
-                .and()
-                .authenticationManager(webfluxAuthenticationManager)
-                .securityContextRepository(jwtTokenSecurityContextRepository)
-                .authorizeExchange()
-                .pathMatchers(WS_TOKEN_BASED_AUTH_ENTRY_POINT)
-                .authenticated()
-                .and()
-                .build();
-    }
-
-}

application/src/main/java/org/thingsboard/server/config/WebSocketConfiguration.java

@@ -21,63 +21,44 @@ import org.springframework.http.HttpStatus;
 import org.springframework.http.server.ServerHttpRequest;
 import org.springframework.http.server.ServerHttpResponse;
 import org.springframework.security.core.Authentication;
-import org.springframework.security.core.context.SecurityContext;
 import org.springframework.security.core.context.SecurityContextHolder;
-import org.springframework.security.web.server.context.WebSessionServerSecurityContextRepository;
+import org.springframework.web.socket.WebSocketHandler;
-import org.springframework.web.reactive.HandlerMapping;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
-import org.springframework.web.reactive.HandlerResult;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
-import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
-import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.socket.server.HandshakeInterceptor;
-import org.springframework.web.reactive.socket.server.support.HandshakeWebSocketService;
+import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;
-import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
+import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;
-import org.springframework.web.server.ServerWebExchange;
 import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.controller.plugin.TbWebSocketHandler;
 import org.thingsboard.server.service.security.model.SecurityUser;
-import reactor.core.publisher.Mono;
 
-import java.util.HashMap;
 import java.util.Map;
 
 @Configuration
-public class WebSocketConfiguration {
+@EnableWebSocket
+public class WebSocketConfiguration implements WebSocketConfigurer {
 
     public static final String WS_PLUGIN_PREFIX = "/api/ws/plugins/";
     private static final String WS_PLUGIN_MAPPING = WS_PLUGIN_PREFIX + "**";
 
-/*    @Bean
+    @Bean
     public ServletServerContainerFactoryBean createWebSocketContainer() {
         ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
         container.setMaxTextMessageBufferSize(32768);
         container.setMaxBinaryMessageBufferSize(32768);
         return container;
-    }*/
-
-    @Bean
-    public HandlerMapping handlerMapping() {
-        Map<String, WebSocketHandler> map = new HashMap<>();
-        map.put(WS_PLUGIN_MAPPING, wsHandler());
-
-        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
-        mapping.setUrlMap(map);
-        mapping.setOrder(-1); // before annotated controllers
-        return mapping;
     }
 
-    @Bean
+    @Override
-    public WebSocketHandlerAdapter handlerAdapter() {
-        return new WebSocketHandlerAdapter();
-    }
-
-/*    @Override
     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
         registry.addHandler(wsHandler(), WS_PLUGIN_MAPPING).setAllowedOrigins("*")
                 .addInterceptors(new HttpSessionHandshakeInterceptor(), new HandshakeInterceptor() {
 
                     @Override
                     public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
-                            Map<String, Object> attributes) throws Exception {
+                                                   Map<String, Object> attributes) throws Exception {
                         SecurityUser user = null;
                         try {
                             user = getCurrentUser();
@@ -92,23 +73,23 @@ public class WebSocketConfiguration {
 
                     @Override
                     public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
-                            Exception exception) {
+                                               Exception exception) {
                         //Do nothing
                     }
                 });
-    }*/
+    }
 
     @Bean
     public WebSocketHandler wsHandler() {
         return new TbWebSocketHandler();
     }
 
-/*    protected SecurityUser getCurrentUser() throws ThingsboardException {
+    protected SecurityUser getCurrentUser() throws ThingsboardException {
         Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
         if (authentication != null && authentication.getPrincipal() instanceof SecurityUser) {
             return (SecurityUser) authentication.getPrincipal();
         } else {
             throw new ThingsboardException("You aren't authorized to perform this operation!", ThingsboardErrorCode.AUTHENTICATION);
         }
-    }*/
+    }
 }

application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java

@@ -16,51 +16,48 @@
 package org.thingsboard.server.controller.plugin;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.BeanCreationNotAllowedException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.security.core.Authentication;
 import org.springframework.stereotype.Service;
-import org.springframework.web.reactive.socket.CloseStatus;
+import org.springframework.util.StringUtils;
-import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.socket.CloseStatus;
-import org.springframework.web.reactive.socket.WebSocketSession;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.adapter.NativeWebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.id.UserId;
+import org.thingsboard.server.common.msg.tools.TbRateLimits;
+import org.thingsboard.server.config.WebSocketConfiguration;
 import org.thingsboard.server.service.security.model.SecurityUser;
+import org.thingsboard.server.service.security.model.UserPrincipal;
+import org.thingsboard.server.service.telemetry.SessionEvent;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketMsgEndpoint;
+import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
 import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
-import reactor.core.publisher.Mono;
 
+import javax.websocket.*;
 import java.io.IOException;
-import java.security.Principal;
+import java.net.URI;
+import java.security.InvalidParameterException;
+import java.util.Queue;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.LinkedBlockingQueue;
 
 @Service
 @Slf4j
-public class TbWebSocketHandler implements WebSocketHandler, TelemetryWebSocketMsgEndpoint {
+public class TbWebSocketHandler extends TextWebSocketHandler implements TelemetryWebSocketMsgEndpoint {
 
-    @Override
+    private static final ConcurrentMap<String, SessionMetaData> internalSessionMap = new ConcurrentHashMap<>();
-    public Mono<Void> handle(WebSocketSession session) {
+    private static final ConcurrentMap<String, String> externalSessionMap = new ConcurrentHashMap<>();
-        return session.receive()
-                .doOnNext(message -> {
-                    Principal principal = session.getHandshakeInfo().getPrincipal().block();
-                    if (principal instanceof SecurityUser) {
-                        SecurityUser currentUser = (SecurityUser) principal;
-                        log.info("[{}][{}] Processing {}", currentUser.getTenantId(), session.getId(), message.getPayloadAsText());
-                    } else {
-                        log.info("[{}] Principal {}", session.getId(), principal);
-                        log.info("[{}] Processing {}", session.getId(), message.getPayloadAsText());
-                    }
-                })
-                .then();
-    }
-
-    @Override
-    public void send(TelemetryWebSocketSessionRef sessionRef, int subscriptionId, String msg) throws IOException {
 
-    }
-
-    @Override
-    public void close(TelemetryWebSocketSessionRef sessionRef, CloseStatus withReason) throws IOException {
-
-    }
-
-//    private static final ConcurrentMap<String, SessionMetaData> internalSessionMap = new ConcurrentHashMap<>();
-//    private static final ConcurrentMap<String, String> externalSessionMap = new ConcurrentHashMap<>();
-/*
     @Autowired
     private TelemetryWebSocketService webSocketService;
 
@@ -105,22 +102,6 @@ public class TbWebSocketHandler implements WebSocketHandler, TelemetryWebSocketM
         }
     }
 
-    @Override
-    public Mono<Void> handle(WebSocketSession session) {
-        return session.receive()
-                .doOnNext(message -> {
-                    Principal principal = session.getHandshakeInfo().getPrincipal().block();
-                    if (principal instanceof SecurityUser) {
-                        SecurityUser currentUser = (SecurityUser) principal;
-                        log.info("[{}][{}] Processing {}", currentUser.getTenantId(), session.getId(), message.getPayloadAsText());
-                    } else {
-                        log.info("[{}] Principal {}", session.getId(), principal);
-                        log.info("[{}] Processing {}", session.getId(), message.getPayloadAsText());
-                    }
-                })
-                .then();
-    }
-
     @Override
     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
         super.afterConnectionEstablished(session);
@@ -410,5 +391,5 @@ public class TbWebSocketHandler implements WebSocketHandler, TelemetryWebSocketM
             }
         }
     }
-    */
+
-}
+}

application/src/main/java/org/thingsboard/server/service/security/auth/webflux/WebfluxAuthenticationManager.java

@@ -1,48 +0,0 @@
-/**
- * Copyright © 2016-2019 The Thingsboard Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.thingsboard.server.service.security.auth.webflux;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.security.authentication.ReactiveAuthenticationManager;
-import org.springframework.security.core.Authentication;
-import org.springframework.stereotype.Component;
-import org.thingsboard.server.service.security.auth.JwtAuthenticationToken;
-import org.thingsboard.server.service.security.model.SecurityUser;
-import org.thingsboard.server.service.security.model.token.JwtTokenFactory;
-import org.thingsboard.server.service.security.model.token.RawAccessJwtToken;
-import reactor.core.publisher.Mono;
-
-@Component
-public class WebfluxAuthenticationManager implements ReactiveAuthenticationManager {
-
-    @Autowired
-    private JwtTokenFactory tokenFactory;
-
-    @Override
-    public Mono<Authentication> authenticate(Authentication authentication) {
-        try {
-            if (authentication.getCredentials() != null && authentication.getCredentials() instanceof RawAccessJwtToken) {
-                RawAccessJwtToken rawAccessToken = (RawAccessJwtToken) authentication.getCredentials();
-                SecurityUser securityUser = tokenFactory.parseAccessJwtToken(rawAccessToken);
-                JwtAuthenticationToken auth = new JwtAuthenticationToken(securityUser);
-                return Mono.just(auth);
-            }
-            return Mono.empty();
-        } catch (Exception e) {
-            return Mono.error(e);
-        }
-    }
-}

application/src/main/java/org/thingsboard/server/service/security/auth/webflux/jwt/JwtTokenSecurityContextRepository.java

@@ -1,86 +0,0 @@
-/**
- * Copyright © 2016-2019 The Thingsboard Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.thingsboard.server.service.security.auth.webflux.jwt;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.http.server.reactive.ServerHttpRequest;
-import org.springframework.security.authentication.AuthenticationManager;
-import org.springframework.security.authentication.ReactiveAuthenticationManager;
-import org.springframework.security.core.Authentication;
-import org.springframework.security.core.context.SecurityContext;
-import org.springframework.security.core.context.SecurityContextImpl;
-import org.springframework.security.web.server.context.ServerSecurityContextRepository;
-import org.springframework.security.web.server.context.WebSessionServerSecurityContextRepository;
-import org.springframework.stereotype.Component;
-import org.springframework.util.StringUtils;
-import org.springframework.web.server.ServerWebExchange;
-import org.thingsboard.server.config.ThingsboardSecurityConfiguration;
-import org.thingsboard.server.service.security.auth.JwtAuthenticationToken;
-import org.thingsboard.server.service.security.auth.jwt.extractor.TokenExtractor;
-import org.thingsboard.server.service.security.model.token.RawAccessJwtToken;
-import reactor.core.publisher.Mono;
-
-import java.util.List;
-
-@Component
-public class JwtTokenSecurityContextRepository implements ServerSecurityContextRepository {
-
-    public static final String DEFAULT_SPRING_SECURITY_CONTEXT_ATTR_NAME = "SPRING_SECURITY_CONTEXT";
-
-    @Autowired
-    private ReactiveAuthenticationManager webfluxAuthenticationManager;
-
-    @Override
-    public Mono<Void> save(ServerWebExchange exchange, SecurityContext context) {
-        return exchange.getSession()
-                .doOnNext(session -> {
-                    if (context == null) {
-                        session.getAttributes().remove(WebSessionServerSecurityContextRepository.DEFAULT_SPRING_SECURITY_CONTEXT_ATTR_NAME);
-                    } else {
-                        session.getAttributes().put(WebSessionServerSecurityContextRepository.DEFAULT_SPRING_SECURITY_CONTEXT_ATTR_NAME, context);
-                    }
-                })
-                .flatMap(session -> session.changeSessionId());
-    }
-
-    @Override
-    public Mono<SecurityContext> load(ServerWebExchange exchange) {
-        ServerHttpRequest request = exchange.getRequest();
-        String token = extractTokenFromQuery(request);
-        if (!StringUtils.isEmpty(token)) {
-            RawAccessJwtToken rawToken = new RawAccessJwtToken(token);
-            Authentication auth = new JwtAuthenticationToken(rawToken);
-            return this.webfluxAuthenticationManager.authenticate(auth).map((authentication) -> {
-                return new SecurityContextImpl(authentication);
-            });
-        } else {
-            return Mono.empty();
-        }
-    }
-
-    private String extractTokenFromQuery(ServerHttpRequest request) {
-        String token = null;
-        if (request.getQueryParams() != null) {
-            List<String> tokenParamValue = request.getQueryParams().get(ThingsboardSecurityConfiguration.JWT_TOKEN_QUERY_PARAM);
-            if (tokenParamValue != null && !tokenParamValue.isEmpty()) {
-                token = tokenParamValue.get(0);
-            }
-        }
-        return token;
-    }
-
-}

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java

@@ -26,7 +26,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
-import org.springframework.web.reactive.socket.CloseStatus;
+import org.springframework.web.socket.CloseStatus;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;

application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java

@@ -15,7 +15,7 @@
  */
 package org.thingsboard.server.service.telemetry;
 
-import org.springframework.web.reactive.socket.CloseStatus;
+import org.springframework.web.socket.CloseStatus;
 
 import java.io.IOException;
 

pom.xml

@@ -50,7 +50,7 @@
         <commons-validator.version>1.5.0</commons-validator.version>
         <commons-io.version>2.5</commons-io.version>
         <commons-csv.version>1.4</commons-csv.version>
-        <jackson.version>2.9.0</jackson.version>
+        <jackson.version>2.9.7</jackson.version>
         <json-schema-validator.version>2.2.6</json-schema-validator.version>
         <scala.version>2.11</scala.version>
         <akka.version>2.4.2</akka.version>
@@ -424,7 +424,7 @@
             </dependency>
             <dependency>
                 <groupId>org.springframework.boot</groupId>
-                <artifactId>spring-boot-starter-webflux</artifactId>
+                <artifactId>spring-boot-starter-websocket</artifactId>
                 <version>${spring-boot.version}</version>
             </dependency>
             <dependency>