Merge with release 2.1.2

application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java

@@ -85,11 +85,11 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         testDevice = doPost("/api/device", device, Device.class);
 
         telemetry = new TelemetryEntityView(
-                Arrays.asList("109", "209", "309"),
+                Arrays.asList("tsKey1", "tsKey2", "tsKey3"),
                 new AttributesEntityView(
-                        Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"),
+                        Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"),
-                        Arrays.asList("saValue1", "saValue2", "saValue3", "saValue4"),
+                        Arrays.asList("saKey1", "saKey2", "saKey3", "saKey4"),
-                        Arrays.asList("shValue1", "shValue2", "shValue3", "shValue4")));
+                        Arrays.asList("shKey1", "shKey2", "shKey3", "shKey4")));
     }
 
     @After
@@ -324,10 +324,10 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
     @Test
     public void testTheCopyOfAttrsIntoTSForTheView() throws Exception {
         Set<String> actualAttributesSet =
-                getAttributesByKeys("{\"caValue1\":\"value1\", \"caValue2\":true, \"caValue3\":42.0, \"caValue4\":73}");
+                getAttributesByKeys("{\"caKey1\":\"value1\", \"caKey2\":true, \"caKey3\":42.0, \"caKey4\":73}");
 
         Set<String> expectedActualAttributesSet =
-                new HashSet<>(Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"));
+                new HashSet<>(Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"));
         assertTrue(actualAttributesSet.containsAll(expectedActualAttributesSet));
         Thread.sleep(1000);
 
@@ -335,18 +335,18 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         List<Map<String, Object>> values = doGetAsync("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
                 "/values/attributes?keys=" + String.join(",", actualAttributesSet), List.class);
 
-        assertEquals("value1", getValue(values, "caValue1"));
+        assertEquals("value1", getValue(values, "caKey1"));
-        assertEquals(true, getValue(values, "caValue2"));
+        assertEquals(true, getValue(values, "caKey2"));
-        assertEquals(42.0, getValue(values, "caValue3"));
+        assertEquals(42.0, getValue(values, "caKey3"));
-        assertEquals(73, getValue(values, "caValue4"));
+        assertEquals(73, getValue(values, "caKey4"));
     }
 
     @Test
     public void testTheCopyOfAttrsOutOfTSForTheView() throws Exception {
         Set<String> actualAttributesSet =
-                getAttributesByKeys("{\"caValue1\":\"value1\", \"caValue2\":true, \"caValue3\":42.0, \"caValue4\":73}");
+                getAttributesByKeys("{\"caKey1\":\"value1\", \"caKey2\":true, \"caKey3\":42.0, \"caKey4\":73}");
 
-        Set<String> expectedActualAttributesSet = new HashSet<>(Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"));
+        Set<String> expectedActualAttributesSet = new HashSet<>(Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"));
         assertTrue(actualAttributesSet.containsAll(expectedActualAttributesSet));
         Thread.sleep(1000);
 
@@ -368,6 +368,69 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         assertEquals(0, values.size());
     }
 
+
+    @Test
+    public void testGetTelemetryWhenEntityViewTimeRangeInsideTimestampRange() throws Exception {
+        uploadTelemetry("{\"tsKey1\":\"value1\", \"tsKey2\":true, \"tsKey3\":40.0}");
+        Thread.sleep(1000);
+        long startTimeMs = System.currentTimeMillis();
+        uploadTelemetry("{\"tsKey1\":\"value2\", \"tsKey2\":false, \"tsKey3\":80.0}");
+        Thread.sleep(1000);
+        uploadTelemetry("{\"tsKey1\":\"value3\", \"tsKey2\":false, \"tsKey3\":120.0}");
+        long endTimeMs = System.currentTimeMillis();
+        uploadTelemetry("{\"tsKey1\":\"value4\", \"tsKey2\":true, \"tsKey3\":160.0}");
+
+        String deviceId = testDevice.getId().getId().toString();
+        Set<String> keys = getTelemetryKeys("DEVICE", deviceId);
+        Thread.sleep(1000);
+
+        EntityView view = createEntityView("Test entity view", startTimeMs, endTimeMs);
+        EntityView savedView = doPost("/api/entityView", view, EntityView.class);
+        String entityViewId = savedView.getId().getId().toString();
+
+        Map<String, List<Map<String, String>>> expectedValues = getTelemetryValues("DEVICE", deviceId, keys, 0L, (startTimeMs + endTimeMs) / 2);
+        Assert.assertEquals(2, expectedValues.get("tsKey1").size());
+        Assert.assertEquals(2, expectedValues.get("tsKey2").size());
+        Assert.assertEquals(2, expectedValues.get("tsKey3").size());
+
+        Map<String, List<Map<String, String>>> actualValues = getTelemetryValues("ENTITY_VIEW", entityViewId, keys, 0L, (startTimeMs + endTimeMs) / 2);
+        Assert.assertEquals(1, actualValues.get("tsKey1").size());
+        Assert.assertEquals(1, actualValues.get("tsKey2").size());
+        Assert.assertEquals(1, actualValues.get("tsKey3").size());
+    }
+
+    private void uploadTelemetry(String strKvs) throws Exception {
+        String viewDeviceId = testDevice.getId().getId().toString();
+        DeviceCredentials deviceCredentials =
+                doGet("/api/device/" + viewDeviceId + "/credentials", DeviceCredentials.class);
+        assertEquals(testDevice.getId(), deviceCredentials.getDeviceId());
+
+        String accessToken = deviceCredentials.getCredentialsId();
+        assertNotNull(accessToken);
+
+        String clientId = MqttAsyncClient.generateClientId();
+        MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:1883", clientId);
+
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(accessToken);
+        client.connect(options);
+        Thread.sleep(3000);
+
+        MqttMessage message = new MqttMessage();
+        message.setPayload(strKvs.getBytes());
+        client.publish("v1/devices/me/telemetry", message);
+        Thread.sleep(1000);
+    }
+
+    private Set<String> getTelemetryKeys(String type, String id) throws Exception {
+        return new HashSet<>(doGetAsync("/api/plugins/telemetry/" + type + "/" + id + "/keys/timeseries", List.class));
+    }
+
+    private Map<String, List<Map<String, String>>> getTelemetryValues(String type, String id, Set<String> keys, Long startTs, Long endTs) throws Exception {
+        return doGetAsync("/api/plugins/telemetry/" + type + "/" + id +
+                "/values/timeseries?keys=" + String.join(",", keys) + "&startTs=" + startTs + "&endTs=" + endTs, Map.class);
+    }
+
     private Set<String> getAttributesByKeys(String stringKV) throws Exception {
         String viewDeviceId = testDevice.getId().getId().toString();
         DeviceCredentials deviceCredentials =
@@ -390,7 +453,7 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
         client.publish("v1/devices/me/attributes", message);
         Thread.sleep(1000);
 
-        return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId +  "/keys/attributes", List.class));
+        return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", List.class));
     }
 
     private Object getValue(List<Map<String, Object>> values, String stringValue) {
@@ -401,13 +464,20 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
     }
 
     private EntityView getNewSavedEntityView(String name) throws Exception {
+        EntityView view = createEntityView(name, 0, 0);
+        return doPost("/api/entityView", view, EntityView.class);
+    }
+
+    private EntityView createEntityView(String name, long startTimeMs, long endTimeMs) {
         EntityView view = new EntityView();
         view.setEntityId(testDevice.getId());
         view.setTenantId(savedTenant.getId());
         view.setName(name);
         view.setType("default");
         view.setKeys(telemetry);
-        return doPost("/api/entityView", view, EntityView.class);
+        view.setStartTimeMs(startTimeMs);
+        view.setEndTimeMs(endTimeMs);
+        return view;
     }
 
     private Customer getNewCustomer(String title) {

application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.mqtt.rpc;
 
 import com.datastax.driver.core.utils.UUIDs;
+import io.netty.handler.codec.mqtt.MqttQoS;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
@@ -23,19 +24,19 @@ import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
-import org.junit.After;
+import org.junit.*;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
 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.security.Authority;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.controller.AbstractControllerTest;
+import org.thingsboard.server.mqtt.telemetry.AbstractMqttTelemetryIntegrationTest;
 import org.thingsboard.server.service.security.AccessValidator;
 
 import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -101,13 +102,19 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
         MqttConnectOptions options = new MqttConnectOptions();
         options.setUserName(accessToken);
         client.connect(options).waitForCompletion();
-        client.subscribe("v1/devices/me/rpc/request/+", 1);
+
-        client.setCallback(new TestMqttCallback(client));
+        CountDownLatch latch = new CountDownLatch(1);
+        TestMqttCallback callback = new TestMqttCallback(client, latch);
+        client.setCallback(callback);
+
+        client.subscribe("v1/devices/me/rpc/request/+", MqttQoS.AT_MOST_ONCE.value());
 
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
         String result = doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
         Assert.assertTrue(StringUtils.isEmpty(result));
+        latch.await(3, TimeUnit.SECONDS);
+        assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
     }
 
     @Test
@@ -156,7 +163,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
         options.setUserName(accessToken);
         client.connect(options).waitForCompletion();
         client.subscribe("v1/devices/me/rpc/request/+", 1);
-        client.setCallback(new TestMqttCallback(client));
+        client.setCallback(new TestMqttCallback(client, new CountDownLatch(1)));
 
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
@@ -204,9 +211,16 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
     private static class TestMqttCallback implements MqttCallback {
 
         private final MqttAsyncClient client;
+        private final CountDownLatch latch;
+        private Integer qoS;
 
-        TestMqttCallback(MqttAsyncClient client) {
+        TestMqttCallback(MqttAsyncClient client, CountDownLatch latch) {
             this.client = client;
+            this.latch = latch;
+        }
+
+        int getQoS() {
+            return qoS;
         }
 
         @Override
@@ -219,7 +233,9 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
             MqttMessage message = new MqttMessage();
             String responseTopic = requestTopic.replace("request", "response");
             message.setPayload("{\"value1\":\"A\", \"value2\":\"B\"}".getBytes("UTF-8"));
+            qoS = mqttMessage.getQos();
             client.publish(responseTopic, message);
+            latch.countDown();
         }
 
         @Override

application/src/test/java/org/thingsboard/server/mqtt/telemetry/AbstractMqttTelemetryIntegrationTest.java

@@ -15,10 +15,9 @@
  */
 package org.thingsboard.server.mqtt.telemetry;
 
+import io.netty.handler.codec.mqtt.MqttQoS;
 import lombok.extern.slf4j.Slf4j;
-import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
-import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -30,9 +29,12 @@ import org.thingsboard.server.dao.service.DaoNoSqlTest;
 
 import java.net.URI;
 import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 /**
  * @author Valerii Sosliuk
@@ -94,4 +96,62 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
         assertEquals("3.0", values.get("key3").get(0).get("value"));
         assertEquals("4", values.get("key4").get(0).get("value"));
     }
+
+    @Test
+    public void testMqttQoSLevel() throws Exception {
+        String clientId = MqttAsyncClient.generateClientId();
+        MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, clientId);
+
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(accessToken);
+        client.connect(options).waitForCompletion(3000);
+        CountDownLatch latch = new CountDownLatch(1);
+        TestMqttCallback callback = new TestMqttCallback(client, latch);
+        client.setCallback(callback);
+        client.subscribe("v1/devices/me/attributes", MqttQoS.AT_MOST_ONCE.value());
+        String payload = "{\"key\":\"value\"}";
+        String result = doPostAsync("/api/plugins/telemetry/" + savedDevice.getId() + "/SHARED_SCOPE", payload, String.class, status().isOk());
+        latch.await(3, TimeUnit.SECONDS);
+        assertEquals(payload, callback.getPayload());
+        assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
+    }
+
+    private static class TestMqttCallback implements MqttCallback {
+
+        private final MqttAsyncClient client;
+        private final CountDownLatch latch;
+        private Integer qoS;
+        private String payload;
+
+        String getPayload() {
+            return payload;
+        }
+
+        TestMqttCallback(MqttAsyncClient client, CountDownLatch latch) {
+            this.client = client;
+            this.latch = latch;
+        }
+
+        int getQoS() {
+            return qoS;
+        }
+
+        @Override
+        public void connectionLost(Throwable throwable) {
+        }
+
+        @Override
+        public void messageArrived(String requestTopic, MqttMessage mqttMessage) {
+            payload = new String(mqttMessage.getPayload());
+            qoS = mqttMessage.getQos();
+            latch.countDown();
+        }
+
+        @Override
+        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+
+        }
+    }
+
+
 }

dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java

@@ -87,7 +87,11 @@ public class BaseTimeseriesService implements TimeseriesService {
                             .map(key -> new BaseReadTsKvQuery(key, entityView.getStartTimeMs(), entityView.getEndTimeMs(), 1, "ASC"))
                             .collect(Collectors.toList());
 
-            return timeseriesDao.findAllAsync(entityView.getEntityId(), updateQueriesForEntityView(entityView, queries));
+            if (queries.size() > 0) {
+                return timeseriesDao.findAllAsync(entityView.getEntityId(), queries);
+            } else {
+                return Futures.immediateFuture(new ArrayList<>());
+            }
         }
         keys.forEach(key -> futures.add(timeseriesDao.findLatest(entityId, key)));
         return Futures.allAsList(futures);
@@ -133,11 +137,20 @@ public class BaseTimeseriesService implements TimeseriesService {
 
     private List<ReadTsKvQuery> updateQueriesForEntityView(EntityView entityView, List<ReadTsKvQuery> queries) {
         return queries.stream().map(query -> {
-            long startTs = entityView.getStartTimeMs() == 0 ? query.getStartTs() : entityView.getStartTimeMs();
+            long startTs;
-            long endTs = entityView.getEndTimeMs() == 0 ? query.getEndTs() : entityView.getEndTimeMs();
+            if (entityView.getStartTimeMs() != 0 && entityView.getStartTimeMs() > query.getStartTs()) {
+                startTs = entityView.getStartTimeMs();
+            } else {
+                startTs = query.getStartTs();
+            }
 
-            return startTs <= query.getStartTs() && endTs >= query.getEndTs() ? query :
+            long endTs;
-                    new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation());
+            if (entityView.getEndTimeMs() != 0 && entityView.getEndTimeMs() < query.getEndTs()) {
+                endTs = entityView.getEndTimeMs();
+            } else {
+                endTs = query.getEndTs();
+            }
+            return new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation());
         }).collect(Collectors.toList());
     }
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java

@@ -48,7 +48,7 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
 @Slf4j
 @RuleNode(
         type = ComponentType.ACTION,
-        name = "copy attributes",
+        name = "copy to view",
         configClazz = EmptyNodeConfiguration.class,
         nodeDescription = "Copy attributes from asset/device to entity view and changes message originator to related entity view",
         nodeDetails = "Copy attributes from asset/device to related entity view according to entity view configuration. \n " +

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTopicMatcher.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright © 2016-2018 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.transport.mqtt;
+
+import java.util.regex.Pattern;
+
+public class MqttTopicMatcher {
+
+    private final String topic;
+    private final Pattern topicRegex;
+
+    MqttTopicMatcher(String topic) {
+        if(topic == null){
+            throw new NullPointerException("topic");
+        }
+        this.topic = topic;
+        this.topicRegex = Pattern.compile(topic.replace("+", "[^/]+").replace("#", ".+") + "$");
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public boolean matches(String topic){
+        return this.topicRegex.matcher(topic).matches();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        MqttTopicMatcher that = (MqttTopicMatcher) o;
+
+        return topic.equals(that.topic);
+    }
+
+    @Override
+    public int hashCode() {
+        return topic.hashCode();
+    }
+}

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@@ -54,6 +54,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.regex.Pattern;
 
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
 import static io.netty.handler.codec.mqtt.MqttMessageType.*;
@@ -78,7 +79,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     private final RelationService relationService;
     private final QuotaService quotaService;
     private final SslHandler sslHandler;
-    private final ConcurrentMap<String, Integer> mqttQoSMap;
+    private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
 
     private volatile boolean connected;
     private volatile InetSocketAddress address;
@@ -278,7 +279,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
 
     private void registerSubQoS(String topic, List<Integer> grantedQoSList, MqttQoS reqQoS) {
         grantedQoSList.add(getMinSupportedQos(reqQoS));
-        mqttQoSMap.put(topic, getMinSupportedQos(reqQoS));
+        mqttQoSMap.put(new MqttTopicMatcher(topic), getMinSupportedQos(reqQoS));
     }
 
     private void processUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage mqttMsg) {
@@ -287,7 +288,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         }
         log.trace("[{}] Processing subscription [{}]!", sessionId, mqttMsg.variableHeader().messageId());
         for (String topicName : mqttMsg.payload().topics()) {
-            mqttQoSMap.remove(topicName);
+            mqttQoSMap.remove(new MqttTopicMatcher(topicName));
             try {
                 switch (topicName) {
                     case DEVICE_ATTRIBUTES_TOPIC: {

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java

@@ -16,7 +16,10 @@
 package org.thingsboard.server.transport.mqtt.session;
 
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.mqtt.*;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttQoS;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.id.SessionId;
 import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
@@ -27,10 +30,9 @@ import org.thingsboard.server.common.msg.session.ex.SessionException;
 import org.thingsboard.server.common.transport.SessionMsgProcessor;
 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
 import org.thingsboard.server.common.transport.auth.DeviceAuthService;
-import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
+import org.thingsboard.server.transport.mqtt.MqttTopicMatcher;
 import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 
-import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -46,7 +48,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
     private volatile boolean allowAttributeResponses;
     private AtomicInteger msgIdSeq = new AtomicInteger(0);
 
-    public DeviceSessionCtx(SessionMsgProcessor processor, DeviceAuthService authService, MqttTransportAdaptor adaptor, ConcurrentMap<String, Integer> mqttQoSMap) {
+    public DeviceSessionCtx(SessionMsgProcessor processor, DeviceAuthService authService, MqttTransportAdaptor adaptor, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap) {
         super(processor, authService, mqttQoSMap);
         this.adaptor = adaptor;
         this.sessionId = new MqttSessionId();

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java

@@ -33,6 +33,7 @@ import org.thingsboard.server.common.msg.session.*;
 import org.thingsboard.server.common.msg.session.ex.SessionException;
 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
+import org.thingsboard.server.transport.mqtt.MqttTopicMatcher;
 import org.thingsboard.server.transport.mqtt.MqttTopics;
 import org.thingsboard.server.transport.mqtt.MqttTransportHandler;
 
@@ -58,7 +59,7 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext {
     private volatile boolean closed;
     private AtomicInteger msgIdSeq = new AtomicInteger(0);
 
-    public GatewayDeviceSessionCtx(GatewaySessionCtx parent, Device device, ConcurrentMap<String, Integer> mqttQoSMap) {
+    public GatewayDeviceSessionCtx(GatewaySessionCtx parent, Device device, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap) {
         super(parent.getProcessor(), parent.getAuthService(), device, mqttQoSMap);
         this.parent = parent;
         this.sessionId = new MqttSessionId();

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java

@@ -39,6 +39,7 @@ import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.common.transport.auth.DeviceAuthService;
 import org.thingsboard.server.dao.device.DeviceService;
 import org.thingsboard.server.dao.relation.RelationService;
+import org.thingsboard.server.transport.mqtt.MqttTopicMatcher;
 import org.thingsboard.server.transport.mqtt.MqttTransportHandler;
 import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor;
 
@@ -64,7 +65,7 @@ public class GatewaySessionCtx {
     private final DeviceAuthService authService;
     private final RelationService relationService;
     private final Map<String, GatewayDeviceSessionCtx> devices;
-    private final ConcurrentMap<String, Integer> mqttQoSMap;
+    private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
     private ChannelHandlerContext channel;
 
     public GatewaySessionCtx(SessionMsgProcessor processor, DeviceService deviceService, DeviceAuthService authService, RelationService relationService, DeviceSessionCtx gatewaySessionCtx) {

transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/MqttDeviceAwareSessionContext.java

@@ -20,35 +20,42 @@ import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.transport.SessionMsgProcessor;
 import org.thingsboard.server.common.transport.auth.DeviceAuthService;
 import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
+import org.thingsboard.server.transport.mqtt.MqttTopicMatcher;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
 
 /**
  * Created by ashvayka on 30.08.18.
  */
 public abstract class MqttDeviceAwareSessionContext extends DeviceAwareSessionContext {
 
-    private final ConcurrentMap<String, Integer> mqttQoSMap;
+    private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
 
-    public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, ConcurrentMap<String, Integer> mqttQoSMap) {
+    public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap) {
         super(processor, authService);
         this.mqttQoSMap = mqttQoSMap;
     }
 
-    public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, Device device, ConcurrentMap<String, Integer> mqttQoSMap) {
+    public MqttDeviceAwareSessionContext(SessionMsgProcessor processor, DeviceAuthService authService, Device device, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap) {
         super(processor, authService, device);
         this.mqttQoSMap = mqttQoSMap;
     }
 
-    public ConcurrentMap<String, Integer> getMqttQoSMap() {
+    public ConcurrentMap<MqttTopicMatcher, Integer> getMqttQoSMap() {
         return mqttQoSMap;
     }
 
     public MqttQoS getQoSForTopic(String topic) {
-        Integer qos = mqttQoSMap.get(topic);
+        List<Integer> qosList = mqttQoSMap.entrySet()
-        if (qos != null) {
+                .stream()
-            return MqttQoS.valueOf(qos);
+                .filter(entry -> entry.getKey().matches(topic))
+                .map(Map.Entry::getValue)
+                .collect(Collectors.toList());
+        if (!qosList.isEmpty()) {
+            return MqttQoS.valueOf(qosList.get(0));
         } else {
             return MqttQoS.AT_LEAST_ONCE;
         }

ui/src/app/entity-view/entity-view.routes.js

@@ -42,7 +42,7 @@ export default function EntityViewRoutes($stateProvider, types) {
                 pageTitle: 'entity-view.entity-views'
             },
             ncyBreadcrumb: {
-                label: '{"icon": "view_stream", "label": "entity-view.entity-views"}'
+                label: '{"icon": "view_quilt", "label": "entity-view.entity-views"}'
             }
         })
         .state('home.customers.entityViews', {
@@ -65,7 +65,7 @@ export default function EntityViewRoutes($stateProvider, types) {
                 pageTitle: 'customer.entity-views'
             },
             ncyBreadcrumb: {
-                label: '{"icon": "view_stream", "label": "{{ vm.customerEntityViewsTitle }}", "translate": "false"}'
+                label: '{"icon": "view_quilt", "label": "{{ vm.customerEntityViewsTitle }}", "translate": "false"}'
             }
         });
 

ui/src/app/entity/entity-subtype-list.directive.js

@@ -47,6 +47,12 @@ export default function EntitySubtypeListDirective($compile, $templateCache, $q,
             scope.secondaryPlaceholder = '+' + $translate.instant('device.device-type');
             scope.noSubtypesMathingText = 'device.no-device-types-matching';
             scope.subtypeListEmptyText = 'device.device-type-list-empty';
+        } else if (scope.entityType == types.entityType.entityView) {
+            scope.placeholder = scope.tbRequired ? $translate.instant('entity-view.enter-entity-view-type')
+                : $translate.instant('entity-view.any-entity-view');
+            scope.secondaryPlaceholder = '+' + $translate.instant('entity-view.entity-view-type');
+            scope.noSubtypesMathingText = 'entity-view.no-entity-view-types-matching';
+            scope.subtypeListEmptyText = 'entity-view.entity-view-type-list-empty';
         }
 
         scope.$watch('tbRequired', function () {

ui/src/app/entity/entity-subtype-select.directive.js

@@ -102,6 +102,9 @@ export default function EntitySubtypeSelect($compile, $templateCache, $translate
             } else if (scope.entityType == types.entityType.device) {
                 scope.entitySubtypeTitle = 'device.device-type';
                 scope.entitySubtypeRequiredText = 'device.device-type-required';
+            } else if (scope.entityType == types.entityType.entityView) {
+                scope.entitySubtypeTitle = 'entity-view.entity-view-type';
+                scope.entitySubtypeRequiredText = 'entity-view.entity-view-type-required';
             }
             scope.entitySubtypes.length = 0;
             if (scope.entitySubtypesList && scope.entitySubtypesList.length) {
@@ -118,6 +121,10 @@ export default function EntitySubtypeSelect($compile, $templateCache, $translate
                     scope.$on('deviceSaved', function() {
                         loadSubTypes();
                     });
+                } else if (scope.entityType == types.entityType.entityView) {
+                    scope.$on('entityViewSaved', function() {
+                        loadSubTypes();
+                    });
                 }
             }
         }

ui/src/app/services/menu.service.js

@@ -171,7 +171,7 @@ function Menu(userService, $state, $rootScope) {
                             name: 'entity-view.entity-views',
                             type: 'link',
                             state: 'home.entityViews',
-                            icon: 'view_stream'
+                            icon: 'view_quilt'
                         },
                         {
                             name: 'widget.widget-library',
@@ -238,7 +238,7 @@ function Menu(userService, $state, $rootScope) {
                                 places: [
                                     {
                                         name: 'entity-view.entity-views',
-                                        icon: 'view_stream',
+                                        icon: 'view_quilt',
                                         state: 'home.entityViews'
                                     }
                                 ]
@@ -293,7 +293,7 @@ function Menu(userService, $state, $rootScope) {
                             name: 'entity-view.entity-views',
                             type: 'link',
                             state: 'home.entityViews',
-                            icon: 'view_stream'
+                            icon: 'view_quilt'
                         },
                         {
                             name: 'dashboard.dashboards',
@@ -328,7 +328,7 @@ function Menu(userService, $state, $rootScope) {
                             places: [
                                 {
                                     name: 'entity-view.entity-views',
-                                    icon: 'view_stream',
+                                    icon: 'view_quilt',
                                     state: 'home.entityViews'
                                 }
                             ]