Websocket refactoring

@@ -304,6 +304,11 @@
             <groupId>com.github.ua-parser</groupId>
             <artifactId>uap-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.java-websocket</groupId>
+            <artifactId>Java-WebSocket</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

+/**
+ * Copyright © 2016-2020 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.subscription;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.EntityView;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.EntityViewId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.query.EntityData;
+import org.thingsboard.server.common.data.query.EntityDataQuery;
+import org.thingsboard.server.common.data.query.TsValue;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
+import org.thingsboard.server.dao.entity.EntityService;
+import org.thingsboard.server.dao.entityview.EntityViewService;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.queue.TbClusterService;
+import org.thingsboard.server.service.telemetry.TelemetryWebSocketService;
+import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.LatestValueCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.TimeSeriesCmd;
+import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Slf4j
+@TbCoreComponent
+@Service
+public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubscriptionService {
+
+    private static final int DEFAULT_LIMIT = 100;
+    private final Set<TopicPartitionInfo> currentPartitions = ConcurrentHashMap.newKeySet();
+    private final Map<String, Map<Integer, TbSubscription>> subscriptionsBySessionId = new ConcurrentHashMap<>();
+
+    @Autowired
+    private TelemetryWebSocketService wsService;
+
+    @Autowired
+    private EntityViewService entityViewService;
+
+    @Autowired
+    private EntityService entityService;
+
+    @Autowired
+    private PartitionService partitionService;
+
+    @Autowired
+    private TbClusterService clusterService;
+
+    @Autowired
+    @Lazy
+    private SubscriptionManagerService subscriptionManagerService;
+
+    @Autowired
+    private TimeseriesService tsService;
+
+    private ExecutorService wsCallBackExecutor;
+
+    @PostConstruct
+    public void initExecutor() {
+        wsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ws-entity-sub-callback"));
+    }
+
+    @PreDestroy
+    public void shutdownExecutor() {
+        if (wsCallBackExecutor != null) {
+            wsCallBackExecutor.shutdownNow();
+        }
+    }
+
+    @Override
+    @EventListener(PartitionChangeEvent.class)
+    public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
+        if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) {
+            currentPartitions.clear();
+            currentPartitions.addAll(partitionChangeEvent.getPartitions());
+        }
+    }
+
+    @Override
+    @EventListener(ClusterTopologyChangeEvent.class)
+    public void onApplicationEvent(ClusterTopologyChangeEvent event) {
+        if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) {
+            /*
+             * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again.
+             * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart.
+             * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically
+             * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService.
+             * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache
+             * Since number of subscriptions is usually much less then number of devices that are pushing data.
+//             */
+//            subscriptionsBySessionId.values().forEach(map -> map.values()
+//                    .forEach(sub -> pushSubscriptionToManagerService(sub, false)));
+        }
+    }
+
+    @Override
+    public void handleCmd(TelemetryWebSocketSessionRef session, EntityDataCmd cmd) {
+        if (cmd.getHistoryCmd() != null) {
+            handleHistoryCmd(session, cmd.getCmdId(), cmd.getQuery(), cmd.getHistoryCmd());
+        } else if (cmd.getLatestCmd() != null) {
+            handleLatestCmd(session, cmd.getCmdId(), cmd.getQuery(), cmd.getLatestCmd());
+        } else {
+            handleTimeseriesCmd(session, cmd.getCmdId(), cmd.getQuery(), cmd.getTsCmd());
+        }
+    }
+
+    private void handleTimeseriesCmd(TelemetryWebSocketSessionRef session, int cmdId, EntityDataQuery query, TimeSeriesCmd tsCmd) {
+    }
+
+    private void handleLatestCmd(TelemetryWebSocketSessionRef session, int cmdId, EntityDataQuery query, LatestValueCmd latestCmd) {
+
+    }
+
+    private void handleHistoryCmd(TelemetryWebSocketSessionRef session, int cmdId, EntityDataQuery query, EntityHistoryCmd historyCmd) {
+        TenantId tenantId = session.getSecurityCtx().getTenantId();
+        CustomerId customerId = session.getSecurityCtx().getCustomerId();
+        PageData<EntityData> data = entityService.findEntityDataByQuery(tenantId, customerId, query);
+        List<ReadTsKvQuery> tsKvQueryList = historyCmd.getKeys().stream().map(key -> new BaseReadTsKvQuery(
+                key, historyCmd.getStartTs(), historyCmd.getEndTs(), historyCmd.getInterval(), getLimit(historyCmd.getLimit()), historyCmd.getAgg()
+        )).collect(Collectors.toList());
+        Map<EntityData, ListenableFuture<List<TsKvEntry>>> fetchResultMap = new HashMap<>();
+        data.getData().forEach(entityData -> fetchResultMap.put(entityData,
+                tsService.findAll(tenantId, entityData.getEntityId(), tsKvQueryList)));
+        Futures.allAsList(fetchResultMap.values()).addListener(() -> {
+            fetchResultMap.forEach((entityData, future) -> {
+                Map<String, List<TsValue>> keyData = new LinkedHashMap<>();
+                historyCmd.getKeys().forEach(key -> keyData.put(key, new ArrayList<>()));
+                try {
+                    List<TsKvEntry> entityTsData = future.get();
+                    if (entityTsData != null) {
+                        entityTsData.forEach(entry -> keyData.get(entry.getKey()).add(new TsValue(entry.getTs(), entry.getValueAsString())));
+                    }
+                    keyData.forEach((k, v) -> entityData.getTimeseries().put(k, v.toArray(new TsValue[v.size()])));
+                } catch (InterruptedException | ExecutionException e) {
+                    log.warn("[{}][{}][{}] Failed to fetch historical data", session.getSessionId(), cmdId, entityData.getEntityId(), e);
+                }
+            });
+            EntityDataUpdate update = new EntityDataUpdate(cmdId, data, null);
+            wsService.sendWsMsg(session.getSessionId(), update);
+        }, wsCallBackExecutor);
+    }
+
+
+    @Override
+    public void cancelSubscription(String sessionId, EntityDataUnsubscribeCmd subscriptionId) {
+
+    }
+
+//    //TODO 3.1: replace null callbacks with callbacks from websocket service.
+//    @Override
+//    public void addSubscription(TbSubscription subscription) {
+//        EntityId entityId = subscription.getEntityId();
+//        // Telemetry subscription on Entity Views are handled differently, because we need to allow only certain keys and time ranges;
+//        if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW) && TbSubscriptionType.TIMESERIES.equals(subscription.getType())) {
+//            subscription = resolveEntityViewSubscription((TbTimeseriesSubscription) subscription);
+//        }
+//        pushSubscriptionToManagerService(subscription, true);
+//        registerSubscription(subscription);
+//    }
+
+//    private void pushSubscriptionToManagerService(TbSubscription subscription, boolean pushToLocalService) {
+//        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId());
+//        if (currentPartitions.contains(tpi)) {
+//            // Subscription is managed on the same server;
+//            if (pushToLocalService) {
+//                subscriptionManagerService.addSubscription(subscription, TbCallback.EMPTY);
+//            }
+//        } else {
+//            // Push to the queue;
+//            TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toNewSubscriptionProto(subscription);
+//            clusterService.pushMsgToCore(tpi, subscription.getEntityId().getId(), toCoreMsg, null);
+//        }
+//    }
+
+    @Override
+    public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback) {
+//        TbSubscription subscription = subscriptionsBySessionId
+//                .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId());
+//        if (subscription != null) {
+//            switch (subscription.getType()) {
+//                case TIMESERIES:
+//                    TbTimeseriesSubscription tsSub = (TbTimeseriesSubscription) subscription;
+//                    update.getLatestValues().forEach((key, value) -> tsSub.getKeyStates().put(key, value));
+//                    break;
+//                case ATTRIBUTES:
+//                    TbAttributeSubscription attrSub = (TbAttributeSubscription) subscription;
+//                    update.getLatestValues().forEach((key, value) -> attrSub.getKeyStates().put(key, value));
+//                    break;
+//            }
+//            wsService.sendWsMsg(sessionId, update);
+//        }
+//        callback.onSuccess();
+    }
+
+//    @Override
+//    public void cancelSubscription(String sessionId, int subscriptionId) {
+//        log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId);
+//        Map<Integer, TbSubscription> sessionSubscriptions = subscriptionsBySessionId.get(sessionId);
+//        if (sessionSubscriptions != null) {
+//            TbSubscription subscription = sessionSubscriptions.remove(subscriptionId);
+//            if (subscription != null) {
+//                if (sessionSubscriptions.isEmpty()) {
+//                    subscriptionsBySessionId.remove(sessionId);
+//                }
+//                TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId());
+//                if (currentPartitions.contains(tpi)) {
+//                    // Subscription is managed on the same server;
+//                    subscriptionManagerService.cancelSubscription(sessionId, subscriptionId, TbCallback.EMPTY);
+//                } else {
+//                    // Push to the queue;
+//                    TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toCloseSubscriptionProto(subscription);
+//                    clusterService.pushMsgToCore(tpi, subscription.getEntityId().getId(), toCoreMsg, null);
+//                }
+//            } else {
+//                log.debug("[{}][{}] Subscription not found!", sessionId, subscriptionId);
+//            }
+//        } else {
+//            log.debug("[{}] No session subscriptions found!", sessionId);
+//        }
+//    }
+
+    @Override
+    public void cancelAllSessionSubscriptions(String sessionId) {
+//        Map<Integer, TbSubscription> subscriptions = subscriptionsBySessionId.get(sessionId);
+//        if (subscriptions != null) {
+//            Set<Integer> toRemove = new HashSet<>(subscriptions.keySet());
+//            toRemove.forEach(id -> cancelSubscription(sessionId, id));
+//        }
+    }
+
+    private TbSubscription resolveEntityViewSubscription(TbTimeseriesSubscription subscription) {
+        EntityView entityView = entityViewService.findEntityViewById(TenantId.SYS_TENANT_ID, new EntityViewId(subscription.getEntityId().getId()));
+
+        Map<String, Long> keyStates;
+        if (subscription.isAllKeys()) {
+            keyStates = entityView.getKeys().getTimeseries().stream().collect(Collectors.toMap(k -> k, k -> 0L));
+        } else {
+            keyStates = subscription.getKeyStates().entrySet()
+                    .stream().filter(entry -> entityView.getKeys().getTimeseries().contains(entry.getKey()))
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
+
+        return TbTimeseriesSubscription.builder()
+                .serviceId(subscription.getServiceId())
+                .sessionId(subscription.getSessionId())
+                .subscriptionId(subscription.getSubscriptionId())
+                .tenantId(subscription.getTenantId())
+                .entityId(entityView.getEntityId())
+                .startTime(entityView.getStartTimeMs())
+                .endTime(entityView.getEndTimeMs())
+                .allKeys(false)
+                .keyStates(keyStates).build();
+    }
+
+    private void registerSubscription(TbSubscription subscription) {
+        Map<Integer, TbSubscription> sessionSubscriptions = subscriptionsBySessionId.computeIfAbsent(subscription.getSessionId(), k -> new ConcurrentHashMap<>());
+        sessionSubscriptions.put(subscription.getSubscriptionId(), subscription);
+    }
+
+    private int getLimit(int limit) {
+        return limit == 0 ? DEFAULT_LIMIT : limit;
+    }
+
+}

+/**
+ * Copyright © 2016-2020 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.subscription;
+
+import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent;
+import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
+import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
+import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
+
+public interface TbEntityDataSubscriptionService {
+
+    void handleCmd(TelemetryWebSocketSessionRef sessionId, EntityDataCmd cmd);
+
+    void cancelSubscription(String sessionId, EntityDataUnsubscribeCmd subscriptionId);
+
+    void cancelAllSessionSubscriptions(String sessionId);
+
+    void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback);
+
+    void onApplicationEvent(PartitionChangeEvent event);
+
+    void onApplicationEvent(ClusterTopologyChangeEvent event);
+}

@@ -49,8 +49,10 @@ import org.thingsboard.server.service.security.AccessValidator;
 import org.thingsboard.server.service.security.ValidationCallback;
 import org.thingsboard.server.service.security.ValidationResult;
 import org.thingsboard.server.service.security.ValidationResultCode;
+import org.thingsboard.server.service.security.model.SecurityUser;
 import org.thingsboard.server.service.security.model.UserPrincipal;
 import org.thingsboard.server.service.security.permission.Operation;
+import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionService;
 import org.thingsboard.server.service.subscription.TbLocalSubscriptionService;
 import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
 import org.thingsboard.server.service.subscription.TbAttributeSubscription;
@@ -61,6 +63,9 @@ import org.thingsboard.server.service.telemetry.cmd.v1.SubscriptionCmd;
 import org.thingsboard.server.service.telemetry.cmd.v1.TelemetryPluginCmd;
 import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
 import org.thingsboard.server.service.telemetry.cmd.v1.TimeseriesSubscriptionCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUnsubscribeCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
 import org.thingsboard.server.service.telemetry.exception.UnauthorizedException;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
@@ -104,7 +109,10 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
     private final ConcurrentMap<String, WsSessionMetaData> wsSessionsMap = new ConcurrentHashMap<>();
 
     @Autowired
-    private TbLocalSubscriptionService subService;
+    private TbLocalSubscriptionService oldSubService;
+
+    @Autowired
+    private TbEntityDataSubscriptionService entityDataSubService;
 
     @Autowired
     private TelemetryWebSocketMsgEndpoint msgEndpoint;
@@ -164,7 +172,8 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                 break;
             case CLOSED:
                 wsSessionsMap.remove(sessionId);
-                subService.cancelAllSessionSubscriptions(sessionId);
+                oldSubService.cancelAllSessionSubscriptions(sessionId);
+                entityDataSubService.cancelAllSessionSubscriptions(sessionId);
                 processSessionClose(sessionRef);
                 break;
         }
@@ -196,6 +205,12 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                 if (cmdsWrapper.getHistoryCmds() != null) {
                     cmdsWrapper.getHistoryCmds().forEach(cmd -> handleWsHistoryCmd(sessionRef, cmd));
                 }
+                if (cmdsWrapper.getEntityDataCmds() != null) {
+                    cmdsWrapper.getEntityDataCmds().forEach(cmd -> handleWsEntityDataCmd(sessionRef, cmd));
+                }
+                if (cmdsWrapper.getEntityDataUnsubscribeCmds() != null) {
+                    cmdsWrapper.getEntityDataUnsubscribeCmds().forEach(cmd -> handleWsEntityDataUnsubscribeCmd(sessionRef, cmd));
+                }
             }
         } catch (IOException e) {
             log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e);
@@ -204,11 +219,39 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         }
     }
 
+    private void handleWsEntityDataCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
+        String sessionId = sessionRef.getSessionId();
+        log.debug("[{}] Processing: {}", sessionId, cmd);
+
+        if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)
+                && validateSubscriptionCmd(sessionRef, cmd)) {
+            entityDataSubService.handleCmd(sessionRef, cmd);
+        }
+    }
+
+    private void handleWsEntityDataUnsubscribeCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataUnsubscribeCmd cmd) {
+        String sessionId = sessionRef.getSessionId();
+        log.debug("[{}] Processing: {}", sessionId, cmd);
+
+        if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) {
+            entityDataSubService.cancelSubscription(sessionRef.getSessionId(), cmd);
+        }
+    }
+
     @Override
     public void sendWsMsg(String sessionId, SubscriptionUpdate update) {
+        sendWsMsg(sessionId, update.getSubscriptionId(), update);
+    }
+
+    @Override
+    public void sendWsMsg(String sessionId, EntityDataUpdate update) {
+        sendWsMsg(sessionId, update.getCmdId(), update);
+    }
+
+    private <T> void sendWsMsg(String sessionId, int cmdId, T update) {
         WsSessionMetaData md = wsSessionsMap.get(sessionId);
         if (md != null) {
-            sendWsMsg(md.getSessionRef(), update);
+            sendWsMsg(md.getSessionRef(), cmdId, update);
         }
     }
 
@@ -356,7 +399,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                         .allKeys(false)
                         .keyStates(subState)
                         .scope(scope).build();
-                subService.addSubscription(sub);
+                oldSubService.addSubscription(sub);
             }
 
             @Override
@@ -453,7 +496,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                         .allKeys(true)
                         .keyStates(subState)
                         .scope(scope).build();
-                subService.addSubscription(sub);
+                oldSubService.addSubscription(sub);
             }
 
             @Override
@@ -534,7 +577,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                         .entityId(entityId)
                         .allKeys(true)
                         .keyStates(subState).build();
-                subService.addSubscription(sub);
+                oldSubService.addSubscription(sub);
             }
 
             @Override
@@ -571,7 +614,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
                         .entityId(entityId)
                         .allKeys(false)
                         .keyStates(subState).build();
-                subService.addSubscription(sub);
+                oldSubService.addSubscription(sub);
             }
 
             @Override
@@ -590,12 +633,32 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
 
     private void unsubscribe(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) {
         if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) {
-            subService.cancelAllSessionSubscriptions(sessionId);
+            oldSubService.cancelAllSessionSubscriptions(sessionId);
         } else {
-            subService.cancelSubscription(sessionId, cmd.getCmdId());
+            oldSubService.cancelSubscription(sessionId, cmd.getCmdId());
         }
     }
 
+    private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) {
+        if (cmd.getCmdId() < 0) {
+            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+                    "Cmd id is negative value!");
+            sendWsMsg(sessionRef, update);
+            return false;
+        } else if (cmd.getQuery() == null) {
+            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+                    "Query is empty!");
+            sendWsMsg(sessionRef, update);
+            return false;
+        } else if (cmd.getHistoryCmd() == null && cmd.getLatestCmd() == null && cmd.getTsCmd() == null) {
+            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
+                    "No history, latest or timeseries command present!");
+            sendWsMsg(sessionRef, update);
+            return false;
+        }
+        return true;
+    }
+
     private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) {
         if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) {
             SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
@@ -607,10 +670,14 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
     }
 
     private boolean validateSessionMetadata(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) {
+        return validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId);
+    }
+
+    private boolean validateSessionMetadata(TelemetryWebSocketSessionRef sessionRef, int cmdId, String sessionId) {
         WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId);
         if (sessionMD == null) {
             log.warn("[{}] Session meta data not found. ", sessionId);
-            SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
+            SubscriptionUpdate update = new SubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR,
                     SESSION_META_DATA_NOT_FOUND);
             sendWsMsg(sessionRef, update);
             return false;
@@ -619,10 +686,18 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         }
     }
 
+    private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, EntityDataUpdate update) {
+        sendWsMsg(sessionRef, update.getCmdId(), update);
+    }
+
     private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, SubscriptionUpdate update) {
+        sendWsMsg(sessionRef, update.getSubscriptionId(), update);
+    }
+
+    private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, int cmdId, Object update) {
         executor.submit(() -> {
             try {
-                msgEndpoint.send(sessionRef, update.getSubscriptionId(), jsonMapper.writeValueAsString(update));
+                msgEndpoint.send(sessionRef, cmdId, jsonMapper.writeValueAsString(update));
             } catch (JsonProcessingException e) {
                 log.warn("[{}] Failed to encode reply: {}", sessionRef.getSessionId(), update, e);
             } catch (IOException e) {
@@ -631,6 +706,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
         });
     }
 
+
     private static Optional<Set<String>> getKeys(TelemetryPluginCmd cmd) {
         if (!StringUtils.isEmpty(cmd.getKeys())) {
             Set<String> keys = new HashSet<>();

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.service.telemetry;
 
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate;
 
 /**
@@ -27,4 +28,7 @@ public interface TelemetryWebSocketService {
     void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg);
 
     void sendWsMsg(String sessionId, SubscriptionUpdate update);
+
+    void sendWsMsg(String sessionId, EntityDataUpdate update);
+
 }

@@ -15,10 +15,12 @@
  */
 package org.thingsboard.server.service.telemetry.cmd.v2;
 
+import lombok.Data;
 import org.thingsboard.server.common.data.kv.Aggregation;
 
 import java.util.List;
 
+@Data
 public class EntityHistoryCmd {
 
     private List<String> keys;

@@ -91,409 +91,8 @@ import static org.springframework.test.web.servlet.setup.MockMvcBuilders.webAppC
 @Configuration
 @ComponentScan({"org.thingsboard.server"})
 @WebAppConfiguration
-@SpringBootTest
+@SpringBootTest()
 @Slf4j
-public abstract class AbstractControllerTest {
-
-    protected ObjectMapper mapper = new ObjectMapper();
-
-    protected static final String TEST_TENANT_NAME = "TEST TENANT";
-
-    protected static final String SYS_ADMIN_EMAIL = "sysadmin@thingsboard.org";
-    private static final String SYS_ADMIN_PASSWORD = "sysadmin";
-
-    protected static final String TENANT_ADMIN_EMAIL = "testtenant@thingsboard.org";
-    private static final String TENANT_ADMIN_PASSWORD = "tenant";
-
-    protected static final String CUSTOMER_USER_EMAIL = "testcustomer@thingsboard.org";
-    private static final String CUSTOMER_USER_PASSWORD = "customer";
-
-    /** See {@link org.springframework.test.web.servlet.DefaultMvcResult#getAsyncResult(long)}
-     *  and {@link org.springframework.mock.web.MockAsyncContext#getTimeout()}
-     */
-    private static final long DEFAULT_TIMEOUT = -1L;
-
-    protected MediaType contentType = MediaType.APPLICATION_JSON;
-
-    protected MockMvc mockMvc;
-
-    protected String token;
-    protected String refreshToken;
-    protected String username;
-
-    private TenantId tenantId;
-
-    @SuppressWarnings("rawtypes")
-    private HttpMessageConverter mappingJackson2HttpMessageConverter;
-
-    @SuppressWarnings("rawtypes")
-    private HttpMessageConverter stringHttpMessageConverter;
-
-    @Autowired
-    private WebApplicationContext webApplicationContext;
-
-    @Rule
-    public TestRule watcher = new TestWatcher() {
-        protected void starting(Description description) {
-            log.info("Starting test: {}", description.getMethodName());
-        }
-
-        protected void finished(Description description) {
-            log.info("Finished test: {}", description.getMethodName());
-        }
-    };
-
-    @Autowired
-    void setConverters(HttpMessageConverter<?>[] converters) {
-
-        this.mappingJackson2HttpMessageConverter = Arrays.stream(converters)
-                .filter(hmc -> hmc instanceof MappingJackson2HttpMessageConverter)
-                .findAny()
-                .get();
-
-        this.stringHttpMessageConverter = Arrays.stream(converters)
-                .filter(hmc -> hmc instanceof StringHttpMessageConverter)
-                .findAny()
-                .get();
-
-        Assert.assertNotNull("the JSON message converter must not be null",
-                this.mappingJackson2HttpMessageConverter);
-    }
-
-    @Before
-    public void setup() throws Exception {
-        log.info("Executing setup");
-        if (this.mockMvc == null) {
-            this.mockMvc = webAppContextSetup(webApplicationContext)
-                    .apply(springSecurity()).build();
-        }
-        loginSysAdmin();
-
-        Tenant tenant = new Tenant();
-        tenant.setTitle(TEST_TENANT_NAME);
-        Tenant savedTenant = doPost("/api/tenant", tenant, Tenant.class);
-        Assert.assertNotNull(savedTenant);
-        tenantId = savedTenant.getId();
-
-        User tenantAdmin = new User();
-        tenantAdmin.setAuthority(Authority.TENANT_ADMIN);
-        tenantAdmin.setTenantId(tenantId);
-        tenantAdmin.setEmail(TENANT_ADMIN_EMAIL);
-
-        createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD);
-
-        Customer customer = new Customer();
-        customer.setTitle("Customer");
-        customer.setTenantId(tenantId);
-        Customer savedCustomer = doPost("/api/customer", customer, Customer.class);
-
-        User customerUser = new User();
-        customerUser.setAuthority(Authority.CUSTOMER_USER);
-        customerUser.setTenantId(tenantId);
-        customerUser.setCustomerId(savedCustomer.getId());
-        customerUser.setEmail(CUSTOMER_USER_EMAIL);
-
-        createUserAndLogin(customerUser, CUSTOMER_USER_PASSWORD);
-
-        logout();
-
-        log.info("Executed setup");
-    }
-
-    @After
-    public void teardown() throws Exception {
-        log.info("Executing teardown");
-        loginSysAdmin();
-        doDelete("/api/tenant/" + tenantId.getId().toString())
-                .andExpect(status().isOk());
-        log.info("Executed teardown");
-    }
-
-    protected void loginSysAdmin() throws Exception {
-        login(SYS_ADMIN_EMAIL, SYS_ADMIN_PASSWORD);
-    }
-
-    protected void loginTenantAdmin() throws Exception {
-        login(TENANT_ADMIN_EMAIL, TENANT_ADMIN_PASSWORD);
-    }
-
-    protected void loginCustomerUser() throws Exception {
-        login(CUSTOMER_USER_EMAIL, CUSTOMER_USER_PASSWORD);
-    }
-
-    private Tenant savedDifferentTenant;
-    protected void loginDifferentTenant() throws Exception {
-        loginSysAdmin();
-        Tenant tenant = new Tenant();
-        tenant.setTitle("Different tenant");
-        savedDifferentTenant = doPost("/api/tenant", tenant, Tenant.class);
-        Assert.assertNotNull(savedDifferentTenant);
-        User differentTenantAdmin = new User();
-        differentTenantAdmin.setAuthority(Authority.TENANT_ADMIN);
-        differentTenantAdmin.setTenantId(savedDifferentTenant.getId());
-        differentTenantAdmin.setEmail("different_tenant@thingsboard.org");
-
-        createUserAndLogin(differentTenantAdmin, "testPassword");
-    }
-
-    protected void deleteDifferentTenant() throws Exception {
-        loginSysAdmin();
-        doDelete("/api/tenant/" + savedDifferentTenant.getId().getId().toString())
-                .andExpect(status().isOk());
-    }
-
-    protected User createUserAndLogin(User user, String password) throws Exception {
-        User savedUser = doPost("/api/user", user, User.class);
-        logout();
-        doGet("/api/noauth/activate?activateToken={activateToken}", TestMailService.currentActivateToken)
-                .andExpect(status().isSeeOther())
-                .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + TestMailService.currentActivateToken));
-        JsonNode activateRequest = new ObjectMapper().createObjectNode()
-                .put("activateToken", TestMailService.currentActivateToken)
-                .put("password", password);
-        JsonNode tokenInfo = readResponse(doPost("/api/noauth/activate", activateRequest).andExpect(status().isOk()), JsonNode.class);
-        validateAndSetJwtToken(tokenInfo, user.getEmail());
-        return savedUser;
-    }
-
-    protected void login(String username, String password) throws Exception {
-        this.token = null;
-        this.refreshToken = null;
-        this.username = null;
-        JsonNode tokenInfo = readResponse(doPost("/api/auth/login", new LoginRequest(username, password)).andExpect(status().isOk()), JsonNode.class);
-        validateAndSetJwtToken(tokenInfo, username);
-    }
-
-    protected void refreshToken() throws Exception {
-        this.token = null;
-        JsonNode tokenInfo = readResponse(doPost("/api/auth/token", new RefreshTokenRequest(this.refreshToken)).andExpect(status().isOk()), JsonNode.class);
-        validateAndSetJwtToken(tokenInfo, this.username);
-    }
-
-    protected void validateAndSetJwtToken(JsonNode tokenInfo, String username) {
-        Assert.assertNotNull(tokenInfo);
-        Assert.assertTrue(tokenInfo.has("token"));
-        Assert.assertTrue(tokenInfo.has("refreshToken"));
-        String token = tokenInfo.get("token").asText();
-        String refreshToken = tokenInfo.get("refreshToken").asText();
-        validateJwtToken(token, username);
-        validateJwtToken(refreshToken, username);
-        this.token = token;
-        this.refreshToken = refreshToken;
-        this.username = username;
-    }
-
-    protected void validateJwtToken(String token, String username) {
-        Assert.assertNotNull(token);
-        Assert.assertFalse(token.isEmpty());
-        int i = token.lastIndexOf('.');
-        Assert.assertTrue(i > 0);
-        String withoutSignature = token.substring(0, i + 1);
-        Jwt<Header, Claims> jwsClaims = Jwts.parser().parseClaimsJwt(withoutSignature);
-        Claims claims = jwsClaims.getBody();
-        String subject = claims.getSubject();
-        Assert.assertEquals(username, subject);
-    }
-
-    protected void logout() throws Exception {
-        this.token = null;
-        this.refreshToken = null;
-        this.username = null;
-    }
-
-    protected void setJwtToken(MockHttpServletRequestBuilder request) {
-        if (this.token != null) {
-            request.header(ThingsboardSecurityConfiguration.JWT_TOKEN_HEADER_PARAM, "Bearer " + this.token);
-        }
-    }
-
-    protected ResultActions doGet(String urlTemplate, Object... urlVariables) throws Exception {
-        MockHttpServletRequestBuilder getRequest = get(urlTemplate, urlVariables);
-        setJwtToken(getRequest);
-        return mockMvc.perform(getRequest);
-    }
-
-    protected <T> T doGet(String urlTemplate, Class<T> responseClass, Object... urlVariables) throws Exception {
-        return readResponse(doGet(urlTemplate, urlVariables).andExpect(status().isOk()), responseClass);
-    }
-
-    protected <T> T doGetAsync(String urlTemplate, Class<T> responseClass, Object... urlVariables) throws Exception {
-        return readResponse(doGetAsync(urlTemplate, urlVariables).andExpect(status().isOk()), responseClass);
-    }
-
-    protected ResultActions doGetAsync(String urlTemplate, Object... urlVariables) throws Exception {
-        MockHttpServletRequestBuilder getRequest;
-        getRequest = get(urlTemplate, urlVariables);
-        setJwtToken(getRequest);
-        return mockMvc.perform(asyncDispatch(mockMvc.perform(getRequest).andExpect(request().asyncStarted()).andReturn()));
-    }
-
-    protected <T> T doGetTyped(String urlTemplate, TypeReference<T> responseType, Object... urlVariables) throws Exception {
-        return readResponse(doGet(urlTemplate, urlVariables).andExpect(status().isOk()), responseType);
-    }
-
-    protected <T> T doGetTypedWithPageLink(String urlTemplate, TypeReference<T> responseType,
-                                           PageLink pageLink,
-                                           Object... urlVariables) throws Exception {
-        List<Object> pageLinkVariables = new ArrayList<>();
-        urlTemplate += "pageSize={pageSize}&page={page}";
-        pageLinkVariables.add(pageLink.getPageSize());
-        pageLinkVariables.add(pageLink.getPage());
-        if (StringUtils.isNotEmpty(pageLink.getTextSearch())) {
-            urlTemplate += "&textSearch={textSearch}";
-            pageLinkVariables.add(pageLink.getTextSearch());
-        }
-        if (pageLink.getSortOrder() != null) {
-            urlTemplate += "&sortProperty={sortProperty}&sortOrder={sortOrder}";
-            pageLinkVariables.add(pageLink.getSortOrder().getProperty());
-            pageLinkVariables.add(pageLink.getSortOrder().getDirection().name());
-        }
-
-        Object[] vars = new Object[urlVariables.length + pageLinkVariables.size()];
-        System.arraycopy(urlVariables, 0, vars, 0, urlVariables.length);
-        System.arraycopy(pageLinkVariables.toArray(), 0, vars, urlVariables.length, pageLinkVariables.size());
-
-        return readResponse(doGet(urlTemplate, vars).andExpect(status().isOk()), responseType);
-    }
-
-    protected <T> T  doGetTypedWithTimePageLink(String urlTemplate, TypeReference<T> responseType,
-                                                TimePageLink pageLink,
-                                                Object... urlVariables) throws Exception {
-        List<Object> pageLinkVariables = new ArrayList<>();
-        urlTemplate += "pageSize={pageSize}&page={page}";
-        pageLinkVariables.add(pageLink.getPageSize());
-        pageLinkVariables.add(pageLink.getPage());
-        if (pageLink.getStartTime() != null) {
-            urlTemplate += "&startTime={startTime}";
-            pageLinkVariables.add(pageLink.getStartTime());
-        }
-        if (pageLink.getEndTime() != null) {
-            urlTemplate += "&endTime={endTime}";
-            pageLinkVariables.add(pageLink.getEndTime());
-        }
-        if (StringUtils.isNotEmpty(pageLink.getTextSearch())) {
-            urlTemplate += "&textSearch={textSearch}";
-            pageLinkVariables.add(pageLink.getTextSearch());
-        }
-        if (pageLink.getSortOrder() != null) {
-            urlTemplate += "&sortProperty={sortProperty}&sortOrder={sortOrder}";
-            pageLinkVariables.add(pageLink.getSortOrder().getProperty());
-            pageLinkVariables.add(pageLink.getSortOrder().getDirection().name());
-        }
-        Object[] vars = new Object[urlVariables.length + pageLinkVariables.size()];
-        System.arraycopy(urlVariables, 0, vars, 0, urlVariables.length);
-        System.arraycopy(pageLinkVariables.toArray(), 0, vars, urlVariables.length, pageLinkVariables.size());
-
-        return readResponse(doGet(urlTemplate, vars).andExpect(status().isOk()), responseType);
-    }
-
-    protected <T> T doPost(String urlTemplate, Class<T> responseClass, String... params) throws Exception {
-        return readResponse(doPost(urlTemplate, params).andExpect(status().isOk()), responseClass);
-    }
-
-    protected <T> T doPost(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, String... params) throws Exception {
-        return readResponse(doPost(urlTemplate, content, params).andExpect(resultMatcher), responseClass);
-    }
-
-    protected <T> T doPost(String urlTemplate, T content, Class<T> responseClass, String... params) throws Exception {
-        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseClass);
-    }
-
-    protected <T,R> R doPostWithResponse(String urlTemplate, T content, Class<R> responseClass, String... params) throws Exception {
-        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseClass);
-    }
-
-    protected <T,R> R doPostWithTypedResponse(String urlTemplate, T content, TypeReference<R> responseType, String... params) throws Exception {
-        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseType);
-    }
-
-    protected <T> T doPostAsync(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, String... params) throws Exception {
-        return readResponse(doPostAsync(urlTemplate, content, DEFAULT_TIMEOUT, params).andExpect(resultMatcher), responseClass);
-    }
-
-    protected <T> T doPostAsync(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, Long timeout, String... params) throws Exception {
-        return readResponse(doPostAsync(urlTemplate, content, timeout, params).andExpect(resultMatcher), responseClass);
-    }
-
-    protected <T> T doDelete(String urlTemplate, Class<T> responseClass, String... params) throws Exception {
-        return readResponse(doDelete(urlTemplate, params).andExpect(status().isOk()), responseClass);
-    }
-
-    protected ResultActions doPost(String urlTemplate, String... params) throws Exception {
-        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
-        setJwtToken(postRequest);
-        populateParams(postRequest, params);
-        return mockMvc.perform(postRequest);
-    }
-
-    protected <T> ResultActions doPost(String urlTemplate, T content, String... params) throws Exception {
-        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
-        setJwtToken(postRequest);
-        String json = json(content);
-        postRequest.contentType(contentType).content(json);
-        return mockMvc.perform(postRequest);
-    }
-
-    protected <T> ResultActions doPostAsync(String urlTemplate, T content, Long timeout, String... params)  throws Exception {
-        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
-        setJwtToken(postRequest);
-        String json = json(content);
-        postRequest.contentType(contentType).content(json);
-        MvcResult result = mockMvc.perform(postRequest).andReturn();
-        result.getAsyncResult(timeout);
-        return mockMvc.perform(asyncDispatch(result));
-    }
-
-    protected ResultActions doDelete(String urlTemplate, String... params) throws Exception {
-        MockHttpServletRequestBuilder deleteRequest = delete(urlTemplate);
-        setJwtToken(deleteRequest);
-        populateParams(deleteRequest, params);
-        return mockMvc.perform(deleteRequest);
-    }
-
-    protected void populateParams(MockHttpServletRequestBuilder request, String... params) {
-        if (params != null && params.length > 0) {
-            Assert.assertEquals(0, params.length % 2);
-            MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
-            for (int i = 0; i < params.length; i += 2) {
-                paramsMap.add(params[i], params[i + 1]);
-            }
-            request.params(paramsMap);
-        }
-    }
-
-    @SuppressWarnings("unchecked")
-    protected String json(Object o) throws IOException {
-        MockHttpOutputMessage mockHttpOutputMessage = new MockHttpOutputMessage();
-
-        HttpMessageConverter converter = o instanceof String ? stringHttpMessageConverter : mappingJackson2HttpMessageConverter;
-        converter.write(o, MediaType.APPLICATION_JSON, mockHttpOutputMessage);
-        return mockHttpOutputMessage.getBodyAsString();
-    }
-
-    @SuppressWarnings("unchecked")
-    protected <T> T readResponse(ResultActions result, Class<T> responseClass) throws Exception {
-        byte[] content = result.andReturn().getResponse().getContentAsByteArray();
-        MockHttpInputMessage mockHttpInputMessage = new MockHttpInputMessage(content);
-        HttpMessageConverter converter = responseClass.equals(String.class) ? stringHttpMessageConverter : mappingJackson2HttpMessageConverter;
-        return (T) converter.read(responseClass, mockHttpInputMessage);
-    }
-
-    protected <T> T readResponse(ResultActions result, TypeReference<T> type) throws Exception {
-        byte[] content = result.andReturn().getResponse().getContentAsByteArray();
-        ObjectMapper mapper = new ObjectMapper();
-        return mapper.readerFor(type).readValue(content);
-    }
-
-    public class IdComparator<D extends BaseData<? extends UUIDBased>> implements Comparator<D> {
-        @Override
-        public int compare(D o1, D o2) {
-            return o1.getId().getId().compareTo(o2.getId().getId());
-        }
-    }
-
-    protected static <T> ResultMatcher statusReason(Matcher<T> matcher) {
-        return jsonPath("$.message", matcher);
-    }
+public abstract class AbstractControllerTest extends AbstractWebTest  {
 
 }

+/**
+ * Copyright © 2016-2020 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.controller;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Header;
+import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.Jwts;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.hamcrest.Matcher;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootContextLoader;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.StringHttpMessageConverter;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.mock.http.MockHttpInputMessage;
+import org.springframework.mock.http.MockHttpOutputMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.test.web.servlet.ResultActions;
+import org.springframework.test.web.servlet.ResultMatcher;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.context.WebApplicationContext;
+import org.thingsboard.server.common.data.BaseData;
+import org.thingsboard.server.common.data.Customer;
+import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.id.UUIDBased;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.page.TimePageLink;
+import org.thingsboard.server.common.data.security.Authority;
+import org.thingsboard.server.config.ThingsboardSecurityConfiguration;
+import org.thingsboard.server.service.mail.TestMailService;
+import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest;
+import org.thingsboard.server.service.security.auth.rest.LoginRequest;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.setup.MockMvcBuilders.webAppContextSetup;
+
+@Slf4j
+public abstract class AbstractWebTest {
+
+    protected ObjectMapper mapper = new ObjectMapper();
+
+    protected static final String TEST_TENANT_NAME = "TEST TENANT";
+
+    protected static final String SYS_ADMIN_EMAIL = "sysadmin@thingsboard.org";
+    private static final String SYS_ADMIN_PASSWORD = "sysadmin";
+
+    protected static final String TENANT_ADMIN_EMAIL = "testtenant@thingsboard.org";
+    private static final String TENANT_ADMIN_PASSWORD = "tenant";
+
+    protected static final String CUSTOMER_USER_EMAIL = "testcustomer@thingsboard.org";
+    private static final String CUSTOMER_USER_PASSWORD = "customer";
+
+    /** See {@link org.springframework.test.web.servlet.DefaultMvcResult#getAsyncResult(long)}
+     *  and {@link org.springframework.mock.web.MockAsyncContext#getTimeout()}
+     */
+    private static final long DEFAULT_TIMEOUT = -1L;
+
+    protected MediaType contentType = MediaType.APPLICATION_JSON;
+
+    protected MockMvc mockMvc;
+
+    protected String token;
+    protected String refreshToken;
+    protected String username;
+
+    private TenantId tenantId;
+
+    @SuppressWarnings("rawtypes")
+    private HttpMessageConverter mappingJackson2HttpMessageConverter;
+
+    @SuppressWarnings("rawtypes")
+    private HttpMessageConverter stringHttpMessageConverter;
+
+    @Autowired
+    private WebApplicationContext webApplicationContext;
+
+    @Rule
+    public TestRule watcher = new TestWatcher() {
+        protected void starting(Description description) {
+            log.info("Starting test: {}", description.getMethodName());
+        }
+
+        protected void finished(Description description) {
+            log.info("Finished test: {}", description.getMethodName());
+        }
+    };
+
+    @Autowired
+    void setConverters(HttpMessageConverter<?>[] converters) {
+
+        this.mappingJackson2HttpMessageConverter = Arrays.stream(converters)
+                .filter(hmc -> hmc instanceof MappingJackson2HttpMessageConverter)
+                .findAny()
+                .get();
+
+        this.stringHttpMessageConverter = Arrays.stream(converters)
+                .filter(hmc -> hmc instanceof StringHttpMessageConverter)
+                .findAny()
+                .get();
+
+        Assert.assertNotNull("the JSON message converter must not be null",
+                this.mappingJackson2HttpMessageConverter);
+    }
+
+    @Before
+    public void setup() throws Exception {
+        log.info("Executing setup");
+        if (this.mockMvc == null) {
+            this.mockMvc = webAppContextSetup(webApplicationContext)
+                    .apply(springSecurity()).build();
+        }
+        loginSysAdmin();
+
+        Tenant tenant = new Tenant();
+        tenant.setTitle(TEST_TENANT_NAME);
+        Tenant savedTenant = doPost("/api/tenant", tenant, Tenant.class);
+        Assert.assertNotNull(savedTenant);
+        tenantId = savedTenant.getId();
+
+        User tenantAdmin = new User();
+        tenantAdmin.setAuthority(Authority.TENANT_ADMIN);
+        tenantAdmin.setTenantId(tenantId);
+        tenantAdmin.setEmail(TENANT_ADMIN_EMAIL);
+
+        createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD);
+
+        Customer customer = new Customer();
+        customer.setTitle("Customer");
+        customer.setTenantId(tenantId);
+        Customer savedCustomer = doPost("/api/customer", customer, Customer.class);
+
+        User customerUser = new User();
+        customerUser.setAuthority(Authority.CUSTOMER_USER);
+        customerUser.setTenantId(tenantId);
+        customerUser.setCustomerId(savedCustomer.getId());
+        customerUser.setEmail(CUSTOMER_USER_EMAIL);
+
+        createUserAndLogin(customerUser, CUSTOMER_USER_PASSWORD);
+
+        logout();
+
+        log.info("Executed setup");
+    }
+
+    @After
+    public void teardown() throws Exception {
+        log.info("Executing teardown");
+        loginSysAdmin();
+        doDelete("/api/tenant/" + tenantId.getId().toString())
+                .andExpect(status().isOk());
+        log.info("Executed teardown");
+    }
+
+    protected void loginSysAdmin() throws Exception {
+        login(SYS_ADMIN_EMAIL, SYS_ADMIN_PASSWORD);
+    }
+
+    protected void loginTenantAdmin() throws Exception {
+        login(TENANT_ADMIN_EMAIL, TENANT_ADMIN_PASSWORD);
+    }
+
+    protected void loginCustomerUser() throws Exception {
+        login(CUSTOMER_USER_EMAIL, CUSTOMER_USER_PASSWORD);
+    }
+
+    private Tenant savedDifferentTenant;
+    protected void loginDifferentTenant() throws Exception {
+        loginSysAdmin();
+        Tenant tenant = new Tenant();
+        tenant.setTitle("Different tenant");
+        savedDifferentTenant = doPost("/api/tenant", tenant, Tenant.class);
+        Assert.assertNotNull(savedDifferentTenant);
+        User differentTenantAdmin = new User();
+        differentTenantAdmin.setAuthority(Authority.TENANT_ADMIN);
+        differentTenantAdmin.setTenantId(savedDifferentTenant.getId());
+        differentTenantAdmin.setEmail("different_tenant@thingsboard.org");
+
+        createUserAndLogin(differentTenantAdmin, "testPassword");
+    }
+
+    protected void deleteDifferentTenant() throws Exception {
+        loginSysAdmin();
+        doDelete("/api/tenant/" + savedDifferentTenant.getId().getId().toString())
+                .andExpect(status().isOk());
+    }
+
+    protected User createUserAndLogin(User user, String password) throws Exception {
+        User savedUser = doPost("/api/user", user, User.class);
+        logout();
+        doGet("/api/noauth/activate?activateToken={activateToken}", TestMailService.currentActivateToken)
+                .andExpect(status().isSeeOther())
+                .andExpect(header().string(HttpHeaders.LOCATION, "/login/createPassword?activateToken=" + TestMailService.currentActivateToken));
+        JsonNode activateRequest = new ObjectMapper().createObjectNode()
+                .put("activateToken", TestMailService.currentActivateToken)
+                .put("password", password);
+        JsonNode tokenInfo = readResponse(doPost("/api/noauth/activate", activateRequest).andExpect(status().isOk()), JsonNode.class);
+        validateAndSetJwtToken(tokenInfo, user.getEmail());
+        return savedUser;
+    }
+
+    protected void login(String username, String password) throws Exception {
+        this.token = null;
+        this.refreshToken = null;
+        this.username = null;
+        JsonNode tokenInfo = readResponse(doPost("/api/auth/login", new LoginRequest(username, password)).andExpect(status().isOk()), JsonNode.class);
+        validateAndSetJwtToken(tokenInfo, username);
+    }
+
+    protected void refreshToken() throws Exception {
+        this.token = null;
+        JsonNode tokenInfo = readResponse(doPost("/api/auth/token", new RefreshTokenRequest(this.refreshToken)).andExpect(status().isOk()), JsonNode.class);
+        validateAndSetJwtToken(tokenInfo, this.username);
+    }
+
+    protected void validateAndSetJwtToken(JsonNode tokenInfo, String username) {
+        Assert.assertNotNull(tokenInfo);
+        Assert.assertTrue(tokenInfo.has("token"));
+        Assert.assertTrue(tokenInfo.has("refreshToken"));
+        String token = tokenInfo.get("token").asText();
+        String refreshToken = tokenInfo.get("refreshToken").asText();
+        validateJwtToken(token, username);
+        validateJwtToken(refreshToken, username);
+        this.token = token;
+        this.refreshToken = refreshToken;
+        this.username = username;
+    }
+
+    protected void validateJwtToken(String token, String username) {
+        Assert.assertNotNull(token);
+        Assert.assertFalse(token.isEmpty());
+        int i = token.lastIndexOf('.');
+        Assert.assertTrue(i > 0);
+        String withoutSignature = token.substring(0, i + 1);
+        Jwt<Header, Claims> jwsClaims = Jwts.parser().parseClaimsJwt(withoutSignature);
+        Claims claims = jwsClaims.getBody();
+        String subject = claims.getSubject();
+        Assert.assertEquals(username, subject);
+    }
+
+    protected void logout() throws Exception {
+        this.token = null;
+        this.refreshToken = null;
+        this.username = null;
+    }
+
+    protected void setJwtToken(MockHttpServletRequestBuilder request) {
+        if (this.token != null) {
+            request.header(ThingsboardSecurityConfiguration.JWT_TOKEN_HEADER_PARAM, "Bearer " + this.token);
+        }
+    }
+
+    protected ResultActions doGet(String urlTemplate, Object... urlVariables) throws Exception {
+        MockHttpServletRequestBuilder getRequest = get(urlTemplate, urlVariables);
+        setJwtToken(getRequest);
+        return mockMvc.perform(getRequest);
+    }
+
+    protected <T> T doGet(String urlTemplate, Class<T> responseClass, Object... urlVariables) throws Exception {
+        return readResponse(doGet(urlTemplate, urlVariables).andExpect(status().isOk()), responseClass);
+    }
+
+    protected <T> T doGetAsync(String urlTemplate, Class<T> responseClass, Object... urlVariables) throws Exception {
+        return readResponse(doGetAsync(urlTemplate, urlVariables).andExpect(status().isOk()), responseClass);
+    }
+
+    protected ResultActions doGetAsync(String urlTemplate, Object... urlVariables) throws Exception {
+        MockHttpServletRequestBuilder getRequest;
+        getRequest = get(urlTemplate, urlVariables);
+        setJwtToken(getRequest);
+        return mockMvc.perform(asyncDispatch(mockMvc.perform(getRequest).andExpect(request().asyncStarted()).andReturn()));
+    }
+
+    protected <T> T doGetTyped(String urlTemplate, TypeReference<T> responseType, Object... urlVariables) throws Exception {
+        return readResponse(doGet(urlTemplate, urlVariables).andExpect(status().isOk()), responseType);
+    }
+
+    protected <T> T doGetTypedWithPageLink(String urlTemplate, TypeReference<T> responseType,
+                                           PageLink pageLink,
+                                           Object... urlVariables) throws Exception {
+        List<Object> pageLinkVariables = new ArrayList<>();
+        urlTemplate += "pageSize={pageSize}&page={page}";
+        pageLinkVariables.add(pageLink.getPageSize());
+        pageLinkVariables.add(pageLink.getPage());
+        if (StringUtils.isNotEmpty(pageLink.getTextSearch())) {
+            urlTemplate += "&textSearch={textSearch}";
+            pageLinkVariables.add(pageLink.getTextSearch());
+        }
+        if (pageLink.getSortOrder() != null) {
+            urlTemplate += "&sortProperty={sortProperty}&sortOrder={sortOrder}";
+            pageLinkVariables.add(pageLink.getSortOrder().getProperty());
+            pageLinkVariables.add(pageLink.getSortOrder().getDirection().name());
+        }
+
+        Object[] vars = new Object[urlVariables.length + pageLinkVariables.size()];
+        System.arraycopy(urlVariables, 0, vars, 0, urlVariables.length);
+        System.arraycopy(pageLinkVariables.toArray(), 0, vars, urlVariables.length, pageLinkVariables.size());
+
+        return readResponse(doGet(urlTemplate, vars).andExpect(status().isOk()), responseType);
+    }
+
+    protected <T> T  doGetTypedWithTimePageLink(String urlTemplate, TypeReference<T> responseType,
+                                                TimePageLink pageLink,
+                                                Object... urlVariables) throws Exception {
+        List<Object> pageLinkVariables = new ArrayList<>();
+        urlTemplate += "pageSize={pageSize}&page={page}";
+        pageLinkVariables.add(pageLink.getPageSize());
+        pageLinkVariables.add(pageLink.getPage());
+        if (pageLink.getStartTime() != null) {
+            urlTemplate += "&startTime={startTime}";
+            pageLinkVariables.add(pageLink.getStartTime());
+        }
+        if (pageLink.getEndTime() != null) {
+            urlTemplate += "&endTime={endTime}";
+            pageLinkVariables.add(pageLink.getEndTime());
+        }
+        if (StringUtils.isNotEmpty(pageLink.getTextSearch())) {
+            urlTemplate += "&textSearch={textSearch}";
+            pageLinkVariables.add(pageLink.getTextSearch());
+        }
+        if (pageLink.getSortOrder() != null) {
+            urlTemplate += "&sortProperty={sortProperty}&sortOrder={sortOrder}";
+            pageLinkVariables.add(pageLink.getSortOrder().getProperty());
+            pageLinkVariables.add(pageLink.getSortOrder().getDirection().name());
+        }
+        Object[] vars = new Object[urlVariables.length + pageLinkVariables.size()];
+        System.arraycopy(urlVariables, 0, vars, 0, urlVariables.length);
+        System.arraycopy(pageLinkVariables.toArray(), 0, vars, urlVariables.length, pageLinkVariables.size());
+
+        return readResponse(doGet(urlTemplate, vars).andExpect(status().isOk()), responseType);
+    }
+
+    protected <T> T doPost(String urlTemplate, Class<T> responseClass, String... params) throws Exception {
+        return readResponse(doPost(urlTemplate, params).andExpect(status().isOk()), responseClass);
+    }
+
+    protected <T> T doPost(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, String... params) throws Exception {
+        return readResponse(doPost(urlTemplate, content, params).andExpect(resultMatcher), responseClass);
+    }
+
+    protected <T> T doPost(String urlTemplate, T content, Class<T> responseClass, String... params) throws Exception {
+        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseClass);
+    }
+
+    protected <T,R> R doPostWithResponse(String urlTemplate, T content, Class<R> responseClass, String... params) throws Exception {
+        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseClass);
+    }
+
+    protected <T,R> R doPostWithTypedResponse(String urlTemplate, T content, TypeReference<R> responseType, String... params) throws Exception {
+        return readResponse(doPost(urlTemplate, content, params).andExpect(status().isOk()), responseType);
+    }
+
+    protected <T> T doPostAsync(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, String... params) throws Exception {
+        return readResponse(doPostAsync(urlTemplate, content, DEFAULT_TIMEOUT, params).andExpect(resultMatcher), responseClass);
+    }
+
+    protected <T> T doPostAsync(String urlTemplate, T content, Class<T> responseClass, ResultMatcher resultMatcher, Long timeout, String... params) throws Exception {
+        return readResponse(doPostAsync(urlTemplate, content, timeout, params).andExpect(resultMatcher), responseClass);
+    }
+
+    protected <T> T doDelete(String urlTemplate, Class<T> responseClass, String... params) throws Exception {
+        return readResponse(doDelete(urlTemplate, params).andExpect(status().isOk()), responseClass);
+    }
+
+    protected ResultActions doPost(String urlTemplate, String... params) throws Exception {
+        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
+        setJwtToken(postRequest);
+        populateParams(postRequest, params);
+        return mockMvc.perform(postRequest);
+    }
+
+    protected <T> ResultActions doPost(String urlTemplate, T content, String... params) throws Exception {
+        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
+        setJwtToken(postRequest);
+        String json = json(content);
+        postRequest.contentType(contentType).content(json);
+        return mockMvc.perform(postRequest);
+    }
+
+    protected <T> ResultActions doPostAsync(String urlTemplate, T content, Long timeout, String... params)  throws Exception {
+        MockHttpServletRequestBuilder postRequest = post(urlTemplate);
+        setJwtToken(postRequest);
+        String json = json(content);
+        postRequest.contentType(contentType).content(json);
+        MvcResult result = mockMvc.perform(postRequest).andReturn();
+        result.getAsyncResult(timeout);
+        return mockMvc.perform(asyncDispatch(result));
+    }
+
+    protected ResultActions doDelete(String urlTemplate, String... params) throws Exception {
+        MockHttpServletRequestBuilder deleteRequest = delete(urlTemplate);
+        setJwtToken(deleteRequest);
+        populateParams(deleteRequest, params);
+        return mockMvc.perform(deleteRequest);
+    }
+
+    protected void populateParams(MockHttpServletRequestBuilder request, String... params) {
+        if (params != null && params.length > 0) {
+            Assert.assertEquals(0, params.length % 2);
+            MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
+            for (int i = 0; i < params.length; i += 2) {
+                paramsMap.add(params[i], params[i + 1]);
+            }
+            request.params(paramsMap);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    protected String json(Object o) throws IOException {
+        MockHttpOutputMessage mockHttpOutputMessage = new MockHttpOutputMessage();
+
+        HttpMessageConverter converter = o instanceof String ? stringHttpMessageConverter : mappingJackson2HttpMessageConverter;
+        converter.write(o, MediaType.APPLICATION_JSON, mockHttpOutputMessage);
+        return mockHttpOutputMessage.getBodyAsString();
+    }
+
+    @SuppressWarnings("unchecked")
+    protected <T> T readResponse(ResultActions result, Class<T> responseClass) throws Exception {
+        byte[] content = result.andReturn().getResponse().getContentAsByteArray();
+        MockHttpInputMessage mockHttpInputMessage = new MockHttpInputMessage(content);
+        HttpMessageConverter converter = responseClass.equals(String.class) ? stringHttpMessageConverter : mappingJackson2HttpMessageConverter;
+        return (T) converter.read(responseClass, mockHttpInputMessage);
+    }
+
+    protected <T> T readResponse(ResultActions result, TypeReference<T> type) throws Exception {
+        byte[] content = result.andReturn().getResponse().getContentAsByteArray();
+        ObjectMapper mapper = new ObjectMapper();
+        return mapper.readerFor(type).readValue(content);
+    }
+
+    public class IdComparator<D extends BaseData<? extends UUIDBased>> implements Comparator<D> {
+        @Override
+        public int compare(D o1, D o2) {
+            return o1.getId().getId().compareTo(o2.getId().getId());
+        }
+    }
+
+    protected static <T> ResultMatcher statusReason(Matcher<T> matcher) {
+        return jsonPath("$.message", matcher);
+    }
+
+}

+/**
+ * Copyright © 2016-2020 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.controller;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Header;
+import io.jsonwebtoken.Jwt;
+import io.jsonwebtoken.Jwts;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.hamcrest.Matcher;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootContextLoader;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.StringHttpMessageConverter;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.mock.http.MockHttpInputMessage;
+import org.springframework.mock.http.MockHttpOutputMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.test.web.servlet.ResultActions;
+import org.springframework.test.web.servlet.ResultMatcher;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.context.WebApplicationContext;
+import org.thingsboard.server.common.data.BaseData;
+import org.thingsboard.server.common.data.Customer;
+import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.id.UUIDBased;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.page.TimePageLink;
+import org.thingsboard.server.common.data.security.Authority;
+import org.thingsboard.server.config.ThingsboardSecurityConfiguration;
+import org.thingsboard.server.service.mail.TestMailService;
+import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest;
+import org.thingsboard.server.service.security.auth.rest.LoginRequest;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.springSecurity;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.springframework.test.web.servlet.setup.MockMvcBuilders.webAppContextSetup;
+
+@ActiveProfiles("test")
+@RunWith(SpringRunner.class)
+@ContextConfiguration(classes = AbstractControllerTest.class, loader = SpringBootContextLoader.class)
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+@Configuration
+@ComponentScan({"org.thingsboard.server"})
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@Slf4j
+public abstract class AbstractWebsocketTest extends AbstractWebTest {
+
+    protected static final String WS_URL = "ws://localhost:";
+
+    @LocalServerPort
+    protected int wsPort;
+
+    protected TbTestWebSocketClient buildAndConnectWebSocketClient() throws URISyntaxException, InterruptedException {
+        TbTestWebSocketClient wsClient = new TbTestWebSocketClient(new URI(WS_URL + wsPort + "/api/ws/plugins/telemetry?token=" + token));
+        Assert.assertTrue(wsClient.connectBlocking());
+        return wsClient;
+    }
+
+}

@@ -16,10 +16,16 @@
 package org.thingsboard.server.controller;
 
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import org.apache.http.conn.ssl.TrustStrategy;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.ssl.SSLContexts;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.springframework.boot.web.server.LocalServerPort;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.EntityType;
@@ -27,6 +33,7 @@ import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.User;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.kv.Aggregation;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.query.DeviceTypeFilter;
 import org.thingsboard.server.common.data.query.EntityCountQuery;
@@ -40,10 +47,16 @@ import org.thingsboard.server.common.data.query.EntityListFilter;
 import org.thingsboard.server.common.data.query.KeyFilter;
 import org.thingsboard.server.common.data.query.NumericFilterPredicate;
 import org.thingsboard.server.common.data.security.Authority;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Random;
 import java.util.stream.Collectors;
 
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -190,23 +203,23 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
         List<Device> devices = new ArrayList<>();
         List<Long> temperatures = new ArrayList<>();
         List<Long> highTemperatures = new ArrayList<>();
-        for (int i=0;i<67;i++) {
+        for (int i = 0; i < 67; i++) {
             Device device = new Device();
-            String name = "Device"+i;
+            String name = "Device" + i;
             device.setName(name);
             device.setType("default");
-            device.setLabel("testLabel"+(int)(Math.random()*1000));
-            devices.add(doPost("/api/device?accessToken="+name, device, Device.class));
-            long temperature = (long)(Math.random()*100);
+            device.setLabel("testLabel" + (int) (Math.random() * 1000));
+            devices.add(doPost("/api/device?accessToken=" + name, device, Device.class));
+            long temperature = (long) (Math.random() * 100);
             temperatures.add(temperature);
             if (temperature > 45) {
                 highTemperatures.add(temperature);
             }
         }
-        for (int i=0;i<devices.size();i++) {
+        for (int i = 0; i < devices.size(); i++) {
             Device device = devices.get(i);
-            String payload = "{\"temperature\":"+temperatures.get(i)+"}";
-            doPost("/api/plugins/telemetry/"+device.getId()+"/"+ DataConstants.SHARED_SCOPE, payload, String.class, status().isOk());
+            String payload = "{\"temperature\":" + temperatures.get(i) + "}";
+            doPost("/api/plugins/telemetry/" + device.getId() + "/" + DataConstants.SHARED_SCOPE, payload, String.class, status().isOk());
         }
         Thread.sleep(1000);
 

+/**
+ * Copyright © 2016-2020 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.controller;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.kv.Aggregation;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.query.DeviceTypeFilter;
+import org.thingsboard.server.common.data.query.EntityData;
+import org.thingsboard.server.common.data.query.EntityDataPageLink;
+import org.thingsboard.server.common.data.query.EntityDataQuery;
+import org.thingsboard.server.common.data.query.TsValue;
+import org.thingsboard.server.common.data.security.Authority;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+public class BaseWebsocketApiTest extends AbstractWebsocketTest {
+
+    private Tenant savedTenant;
+    private User tenantAdmin;
+    private TbTestWebSocketClient wsClient;
+
+    @Autowired
+    private TimeseriesService tsService;
+
+    @Before
+    public void beforeTest() throws Exception {
+        loginSysAdmin();
+
+        Tenant tenant = new Tenant();
+        tenant.setTitle("My tenant");
+        savedTenant = doPost("/api/tenant", tenant, Tenant.class);
+        Assert.assertNotNull(savedTenant);
+
+        tenantAdmin = new User();
+        tenantAdmin.setAuthority(Authority.TENANT_ADMIN);
+        tenantAdmin.setTenantId(savedTenant.getId());
+        tenantAdmin.setEmail("tenant2@thingsboard.org");
+        tenantAdmin.setFirstName("Joe");
+        tenantAdmin.setLastName("Downs");
+
+        tenantAdmin = createUserAndLogin(tenantAdmin, "testPassword1");
+
+        wsClient = buildAndConnectWebSocketClient();
+    }
+
+    @After
+    public void afterTest() throws Exception {
+        wsClient.close();
+
+        loginSysAdmin();
+
+        doDelete("/api/tenant/" + savedTenant.getId().getId().toString())
+                .andExpect(status().isOk());
+    }
+
+    @Test
+    public void testEntityDataHistoryWsCmd() throws Exception {
+        Device device = new Device();
+        device.setName("Device");
+        device.setType("default");
+        device.setLabel("testLabel" + (int) (Math.random() * 1000));
+        device = doPost("/api/device", device, Device.class);
+
+        long now = System.currentTimeMillis();
+
+        DeviceTypeFilter dtf = new DeviceTypeFilter();
+        dtf.setDeviceNameFilter("D");
+        dtf.setDeviceType("default");
+        EntityDataQuery edq = new EntityDataQuery(dtf, new EntityDataPageLink(1, 0, null, null), Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
+
+        EntityHistoryCmd historyCmd = new EntityHistoryCmd();
+        historyCmd.setKeys(Arrays.asList("temperature"));
+        historyCmd.setAgg(Aggregation.NONE);
+        historyCmd.setLimit(1000);
+        historyCmd.setStartTs(now - TimeUnit.HOURS.toMillis(1));
+        historyCmd.setEndTs(now);
+        EntityDataCmd cmd = new EntityDataCmd(1, edq, historyCmd, null, null);
+
+        TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper();
+        wrapper.setEntityDataCmds(Collections.singletonList(cmd));
+
+        wsClient.send(mapper.writeValueAsString(wrapper));
+        String msg = wsClient.waitForReply();
+        EntityDataUpdate update = mapper.readValue(msg, EntityDataUpdate.class);
+        Assert.assertEquals(1, update.getCmdId());
+        PageData<EntityData> pageData = update.getData();
+        Assert.assertNotNull(pageData);
+        Assert.assertEquals(1, pageData.getData().size());
+        Assert.assertEquals(device.getId(), pageData.getData().get(0).getEntityId());
+        Assert.assertEquals(0, pageData.getData().get(0).getTimeseries().get("temperature").length);
+
+        TsKvEntry dataPoint1 = new BasicTsKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("temperature", 42L));
+        TsKvEntry dataPoint2 = new BasicTsKvEntry(now - TimeUnit.MINUTES.toMillis(2), new LongDataEntry("temperature", 42L));
+        TsKvEntry dataPoint3 = new BasicTsKvEntry(now - TimeUnit.MINUTES.toMillis(3), new LongDataEntry("temperature", 42L));
+        tsService.save(device.getTenantId(), device.getId(), Arrays.asList(dataPoint1, dataPoint2, dataPoint3), 0).get();
+
+        wsClient.send(mapper.writeValueAsString(wrapper));
+        msg = wsClient.waitForReply();
+        update = mapper.readValue(msg, EntityDataUpdate.class);
+        Assert.assertEquals(1, update.getCmdId());
+        pageData = update.getData();
+        Assert.assertNotNull(pageData);
+        Assert.assertEquals(1, pageData.getData().size());
+        Assert.assertEquals(device.getId(), pageData.getData().get(0).getEntityId());
+        TsValue[] tsArray = pageData.getData().get(0).getTimeseries().get("temperature");
+        Assert.assertEquals(3, tsArray.length);
+        Assert.assertEquals(new TsValue(dataPoint1.getTs(), dataPoint1.getValueAsString()), tsArray[0]);
+        Assert.assertEquals(new TsValue(dataPoint2.getTs(), dataPoint2.getValueAsString()), tsArray[1]);
+        Assert.assertEquals(new TsValue(dataPoint3.getTs(), dataPoint3.getValueAsString()), tsArray[2]);
+    }
+
+}

@@ -26,6 +26,8 @@ import java.util.Arrays;
 
 @RunWith(ClasspathSuite.class)
 @ClasspathSuite.ClassnameFilters({
+//        "org.thingsboard.server.controller.sql.WebsocketApiSqlTest",
+//        "org.thingsboard.server.controller.sql.EntityQueryControllerSqlTest",
         "org.thingsboard.server.controller.sql.*Test",
         })
 public class ControllerSqlTestSuite {

+/**
+ * Copyright © 2016-2020 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.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.java_websocket.client.WebSocketClient;
+import org.java_websocket.handshake.ServerHandshake;
+
+import java.net.URI;
+import java.nio.channels.NotYetConnectedException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class TbTestWebSocketClient extends WebSocketClient {
+
+    private volatile String lastMsg;
+    private volatile boolean replyReceived;
+    private CountDownLatch reply;
+
+    public TbTestWebSocketClient(URI serverUri) {
+        super(serverUri);
+    }
+
+    @Override
+    public void onOpen(ServerHandshake serverHandshake) {
+
+    }
+
+    @Override
+    public void onMessage(String s) {
+        if (!replyReceived) {
+            replyReceived = true;
+            lastMsg = s;
+            if (reply != null) {
+                reply.countDown();
+            }
+        }
+    }
+
+    @Override
+    public void onClose(int i, String s, boolean b) {
+
+    }
+
+    @Override
+    public void onError(Exception e) {
+
+    }
+
+    @Override
+    public void send(String text) throws NotYetConnectedException {
+        reply = new CountDownLatch(1);
+        replyReceived = false;
+        super.send(text);
+    }
+
+    public String waitForReply() {
+        try {
+            reply.await(3, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            log.warn("Failed to await reply", e);
+        }
+        return lastMsg;
+    }
+}

+/**
+ * Copyright © 2016-2020 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.controller.sql;
+
+import org.thingsboard.server.controller.BaseEntityQueryControllerTest;
+import org.thingsboard.server.controller.BaseWebsocketApiTest;
+import org.thingsboard.server.dao.service.DaoSqlTest;
+
+@DaoSqlTest
+public class WebsocketApiSqlTest extends BaseWebsocketApiTest {
+}

@@ -37,7 +37,6 @@
         <blackBoxTests.skip>true</blackBoxTests.skip>
         <testcontainers.version>1.9.1</testcontainers.version>
         <zeroturnaround.version>1.10</zeroturnaround.version>
-        <java-websocket.version>1.3.9</java-websocket.version>
         <httpclient.version>4.5.6</httpclient.version>
     </properties>
 
@@ -55,7 +54,6 @@
         <dependency>
             <groupId>org.java-websocket</groupId>
             <artifactId>Java-WebSocket</artifactId>
-            <version>${java-websocket.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>

@@ -105,6 +105,7 @@
         <ua-parser.version>1.4.3</ua-parser.version>
         <commons-beanutils.version>1.9.4</commons-beanutils.version>
         <commons-collections.version>3.2.2</commons-collections.version>
+        <java-websocket.version>1.3.9</java-websocket.version>
     </properties>
 
     <modules>
@@ -1346,6 +1347,12 @@
                 <artifactId>commons-collections</artifactId>
                 <version>${commons-collections.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.java-websocket</groupId>
+                <artifactId>Java-WebSocket</artifactId>
+                <version>${java-websocket.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>