MQTT RPC Client test

application/pom.xml

@@ -85,6 +85,12 @@
             <artifactId>takari-cpsuite</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.1.0</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.cassandraunit</groupId>
             <artifactId>cassandra-unit</artifactId>
@@ -212,6 +218,12 @@
             <groupId>io.grpc</groupId>
             <artifactId>grpc-stub</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.thingsboard.server</groupId>
+            <artifactId>tools</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

application/src/test/java/org/thingsboard/server/rpc/AbstractRpcIntegrationTest.java

@@ -0,0 +1,71 @@
+/**
+ * Copyright © 2016 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.rpc;
+
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.IntegrationTest;
+import org.springframework.boot.test.SpringApplicationContextLoader;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.converter.HttpMessageConverter;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.web.context.WebApplicationContext;
+import org.thingsboard.server.rpc.mqtt.MqttRpcIntergrationTest;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * @author Valerii Sosliuk
+ */
+@ActiveProfiles("default")
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes=MqttRpcIntergrationTest.class, loader=SpringApplicationContextLoader.class)
+@TestPropertySource("classpath:cassandra-test.properties")
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+@Configuration
+@EnableAutoConfiguration
+@ComponentScan({"org.thingsboard.server"})
+@WebAppConfiguration
+@IntegrationTest("server.port:8080")
+public class AbstractRpcIntegrationTest {
+
+    @SuppressWarnings("rawtypes")
+    private HttpMessageConverter mappingJackson2HttpMessageConverter;
+
+    @Autowired
+    private WebApplicationContext webApplicationContext;
+
+    @Autowired
+    void setConverters(HttpMessageConverter<?>[] converters) {
+
+        this.mappingJackson2HttpMessageConverter = Arrays.asList(converters).stream().filter(
+                hmc -> hmc instanceof MappingJackson2HttpMessageConverter).findAny().get();
+
+        assertNotNull("the JSON message converter must not be null",
+                this.mappingJackson2HttpMessageConverter);
+    }
+
+}

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

@@ -0,0 +1,99 @@
+/**
+ * Copyright © 2016 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.rpc.mqtt;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.web.util.UriComponentsBuilder;
+import org.thingsboard.client.tools.RestClient;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.security.DeviceCredentials;
+import org.thingsboard.server.rpc.AbstractRpcIntegrationTest;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * @author Valerii Sosliuk
+ */
+@Slf4j
+public class MqttRpcIntergrationTest extends AbstractRpcIntegrationTest {
+
+    private static final String MQTT_URL = "tcp://localhost:1883";
+    private static final String BASE_URL = "http://localhost:8080";
+
+    private static final String USERNAME = "tenant@thingsboard.org";
+    private static final String PASSWORD = "tenant";
+
+    private Device savedDevice;
+
+    private String accessToken;
+    private RestClient restClient;
+
+    @Before
+    public void beforeTest() throws Exception {
+        restClient = new RestClient(BASE_URL);
+        restClient.login(USERNAME, PASSWORD);
+
+        Device device = new Device();
+        device.setName("Test device");
+        savedDevice = restClient.getRestTemplate().postForEntity(BASE_URL + "/api/device", device, Device.class).getBody();
+        DeviceCredentials deviceCredentials =
+                restClient.getRestTemplate().getForEntity(BASE_URL + "/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class).getBody();
+        assertEquals(savedDevice.getId(), deviceCredentials.getDeviceId());
+        accessToken = deviceCredentials.getCredentialsId();
+        assertNotNull(accessToken);
+    }
+
+    @Test
+    public void testPushMqttRpcData() throws Exception {
+        String clientId = MqttAsyncClient.generateClientId();
+        MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, clientId);
+
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(accessToken);
+        client.connect(options);
+        Thread.sleep(3000);
+        MqttMessage message = new MqttMessage();
+        message.setPayload("{\"key1\":\"value1\", \"key2\":true, \"key3\": 3.0, \"key4\": 4}".getBytes());
+        client.publish("v1/devices/me/telemetry", message);
+
+        String deviceId = savedDevice.getId().getId().toString();
+
+        Thread.sleep(1000);
+        List keys = restClient.getRestTemplate().getForEntity(BASE_URL + "/api/plugins/telemetry/" + deviceId +  "/keys/timeseries", List.class).getBody();
+        assertEquals(Arrays.asList("key1", "key2", "key3", "key4"), keys);
+
+        UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(BASE_URL + "/api/plugins/telemetry/" + deviceId +  "/values/timeseries")
+                .queryParam("keys", String.join(",", keys));
+        URI uri = builder.build().encode().toUri();
+        Map<String, List<Map<String, String>>> values = restClient.getRestTemplate().getForEntity(uri, Map.class).getBody();
+
+        assertEquals("value1", values.get("key1").get(0).get("value"));
+        assertEquals("true", values.get("key2").get(0).get("value"));
+        assertEquals("3.0", values.get("key3").get(0).get("value"));
+        assertEquals("4", values.get("key4").get(0).get("value"));
+    }
+}

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

@@ -0,0 +1,41 @@
+/**
+ * Copyright © 2016 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.rpc.mqtt;
+
+import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
+import org.junit.ClassRule;
+import org.junit.extensions.cpsuite.ClasspathSuite;
+import org.junit.runner.RunWith;
+import org.thingsboard.server.dao.CustomCassandraCQLUnit;
+
+import java.util.Arrays;
+
+/**
+ * @author Valerii Sosliuk
+ */
+@RunWith(ClasspathSuite.class)
+@ClasspathSuite.ClassnameFilters({"org.thingsboard.server.rpc.mqtt.*Test"})
+public class MqttRpcSuite {
+
+    @ClassRule
+    public static CustomCassandraCQLUnit cassandraUnit =
+            new CustomCassandraCQLUnit(
+                    Arrays.asList(new ClassPathCQLDataSet("schema.cql", false, false),
+                            new ClassPathCQLDataSet("system-data.cql", false, false),
+                            new ClassPathCQLDataSet("demo-data.cql", false, false)),
+                    "cassandra-test.yaml", 30000l);
+
+}

tools/src/main/java/org/thingsboard/client/tools/MqttStressTestClient.java

@@ -1,4 +1,4 @@
-/**
+package org.thingsboard.client.tools; /**
  * Copyright © 2016 The Thingsboard Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");

tools/src/main/java/org/thingsboard/client/tools/MqttStressTestTool.java

@@ -1,4 +1,4 @@
-/**
+package org.thingsboard.client.tools; /**
  * Copyright © 2016 The Thingsboard Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");

tools/src/main/java/org/thingsboard/client/tools/RestClient.java

@@ -1,4 +1,4 @@
-/**
+package org.thingsboard.client.tools; /**
  * Copyright © 2016 The Thingsboard Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -62,6 +62,10 @@ public class RestClient implements ClientHttpRequestInterceptor {
         return restTemplate.getForEntity(baseURL + "/api/device/" + id.getId().toString() + "/credentials", DeviceCredentials.class).getBody();
     }
 
+    public RestTemplate getRestTemplate() {
+        return restTemplate;
+    }
+
     @Override
     public ClientHttpResponse intercept(HttpRequest request, byte[] bytes, ClientHttpRequestExecution execution) throws IOException {
         HttpRequest wrapper = new HttpRequestWrapper(request);

tools/src/main/java/org/thingsboard/client/tools/ResultAccumulator.java

@@ -1,4 +1,4 @@
-/**
+package org.thingsboard.client.tools; /**
  * Copyright © 2016 The Thingsboard Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -73,7 +73,7 @@ public class ResultAccumulator {
 
     @Override
     public String toString() {
-        return "ResultAccumulator{" +
+        return "org.thingsboard.client.tools.ResultAccumulator{" +
                 "successCount=" + getSuccessCount() +
                 ", errorCount=" + getErrorCount() +
                 ", totalTime=" + getTimeSpent() +