diff --git a/mqtt-broker/pom.xml b/mqtt-broker/pom.xml index 6e93eb65..3d32402a 100644 --- a/mqtt-broker/pom.xml +++ b/mqtt-broker/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 4.2.0-SNAPSHOT + 4.2.1.9 4.0.0 pulsar-protocol-handler-mqtt diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml index d1ef745e..145605af 100644 --- a/mqtt-common/pom.xml +++ b/mqtt-common/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 4.2.0-SNAPSHOT + 4.2.1.9 4.0.0 pulsar-protocol-handler-mqtt-common diff --git a/mqtt-proxy/pom.xml b/mqtt-proxy/pom.xml index 80706090..6d404158 100644 --- a/mqtt-proxy/pom.xml +++ b/mqtt-proxy/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 4.2.0-SNAPSHOT + 4.2.1.9 4.0.0 pulsar-protocol-handler-mqtt-proxy diff --git a/pom.xml b/pom.xml index e33cd345..79d870fc 100644 --- a/pom.xml +++ b/pom.xml @@ -19,7 +19,7 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> io.streamnative.pulsar.handlers pulsar-protocol-handler-mqtt-parent - 4.2.0-SNAPSHOT + 4.2.1.9 StreamNative :: Pulsar Protocol Handler :: MoP Parent Parent for MQTT on Pulsar implemented using Pulsar Protocol Handler. @@ -51,8 +51,8 @@ 2.22.0 6.14.3 4.0.2 - 4.2.0-SNAPSHOT - 4.2.0-SNAPSHOT + 4.2.1.9 + 4.2.1.9 2.18.0 1.16 1.2.2 diff --git a/tests/pom.xml b/tests/pom.xml index 62b6231f..b37cf562 100644 --- a/tests/pom.xml +++ b/tests/pom.xml @@ -20,7 +20,7 @@ pulsar-protocol-handler-mqtt-parent io.streamnative.pulsar.handlers - 4.2.0-SNAPSHOT + 4.2.1.9 4.0.0 pulsar-protocol-handler-mqtt-tests diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyHttpTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyHttpTest.java index ebb9db1f..0128c165 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyHttpTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/proxy/ProxyHttpTest.java @@ -14,25 +14,26 @@ package io.streamnative.pulsar.handlers.mqtt.mqtt3.fusesource.proxy; +import static org.awaitility.Awaitility.await; import com.google.gson.Gson; import io.streamnative.pulsar.handlers.mqtt.base.MQTTTestBase; import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonConfiguration; -import java.io.BufferedReader; -import java.io.InputStream; -import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpResponse; -import org.apache.http.client.HttpClient; import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.util.EntityUtils; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.QoS; import org.testng.Assert; import org.testng.annotations.Test; + /** * Integration tests for MQTT protocol handler with proxy. */ @@ -48,33 +49,36 @@ protected MQTTCommonConfiguration initConfig() throws Exception { @Test public void testGetDeviceList() throws Exception { - int index = random.nextInt(mqttProxyPortList.size()); List mqttProxyPortList = getMqttProxyPortList(); List mqttProxyHttpPortList = getMqttProxyHttpPortList(); + int index = random.nextInt(mqttProxyPortList.size()); MQTT mqttProducer = new MQTT(); int port = mqttProxyPortList.get(index); String clientId = "device-list-client"; mqttProducer.setHost("127.0.0.1", port); mqttProducer.setClientId(clientId); BlockingConnection producer = mqttProducer.blockingConnection(); - producer.connect(); - producer.publish("testHttp", "Hello MQTT".getBytes(StandardCharsets.UTF_8), QoS.AT_MOST_ONCE, false); - Thread.sleep(4000); - HttpClient httpClient = HttpClientBuilder.create().build(); - final String mopEndPoint = "http://localhost:" + mqttProxyHttpPortList.get(index) + "/admin/devices/list"; - HttpResponse response = httpClient.execute(new HttpGet(mopEndPoint)); - InputStream inputStream = response.getEntity().getContent(); - InputStreamReader isReader = new InputStreamReader(inputStream); - BufferedReader reader = new BufferedReader(isReader); - StringBuffer buffer = new StringBuffer(); - String str; - while ((str = reader.readLine()) != null){ - buffer.append(str); + try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) { + producer.connect(); + producer.publish("testHttp", "Hello MQTT".getBytes(StandardCharsets.UTF_8), QoS.AT_MOST_ONCE, false); + final String mopEndPoint = "http://localhost:" + mqttProxyHttpPortList.get(index) + "/admin/devices/list"; + await().atMost(30, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { + String ret = getDeviceList(httpClient, mopEndPoint); + ArrayList deviceList = new Gson().fromJson(ret, ArrayList.class); + Assert.assertNotNull(deviceList, "Invalid device list response: " + ret); + Assert.assertEquals(deviceList.size(), 1, "Unexpected device list response: " + ret); + Assert.assertTrue(deviceList.contains(clientId), "Unexpected device list response: " + ret); + }); + } finally { + producer.disconnect(); } - String ret = buffer.toString(); - ArrayList deviceList = new Gson().fromJson(ret, ArrayList.class); - Assert.assertEquals(deviceList.size(), 1); - Assert.assertTrue(deviceList.contains(clientId)); } + private String getDeviceList(CloseableHttpClient httpClient, String mopEndPoint) throws Exception { + HttpResponse response = httpClient.execute(new HttpGet(mopEndPoint)); + String body = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); + Assert.assertEquals(response.getStatusLine().getStatusCode(), 200, + "Unexpected HTTP response from " + mopEndPoint + ": " + body); + return body; + } }