|
|
@@ -1,27 +1,19 @@
|
|
|
package org.jetlinks.community.network.mqtt.client;
|
|
|
|
|
|
-import io.netty.buffer.Unpooled;
|
|
|
-import io.netty.handler.codec.mqtt.MqttQoS;
|
|
|
import io.vertx.core.Vertx;
|
|
|
-import io.vertx.core.buffer.Buffer;
|
|
|
import io.vertx.mqtt.MqttClientOptions;
|
|
|
-import io.vertx.mqtt.MqttServer;
|
|
|
-import org.jetlinks.community.network.mqtt.server.vertx.VertxMqttServer;
|
|
|
-import org.jetlinks.core.message.Message;
|
|
|
+import org.jetlinks.core.event.EventBus;
|
|
|
+import org.jetlinks.core.event.Subscription;
|
|
|
import org.jetlinks.core.message.codec.MqttMessage;
|
|
|
-import org.jetlinks.core.message.codec.SimpleMqttMessage;
|
|
|
-import org.junit.jupiter.api.Test;
|
|
|
+import org.jetlinks.supports.event.BrokerEventBus;
|
|
|
import org.springframework.mock.env.MockEnvironment;
|
|
|
import reactor.core.publisher.Flux;
|
|
|
import reactor.core.publisher.Mono;
|
|
|
-import reactor.test.StepVerifier;
|
|
|
-
|
|
|
import java.nio.charset.StandardCharsets;
|
|
|
import java.util.Arrays;
|
|
|
-import java.util.Collections;
|
|
|
import java.util.UUID;
|
|
|
import java.util.concurrent.CompletableFuture;
|
|
|
-import java.util.concurrent.TimeUnit;
|
|
|
+import java.util.function.Function;
|
|
|
|
|
|
|
|
|
class MqttClientProviderTest {
|
|
|
@@ -30,10 +22,10 @@ class MqttClientProviderTest {
|
|
|
|
|
|
|
|
|
public static void main(String[] args) throws InterruptedException {
|
|
|
- vertx.setTimer(100,event -> {});
|
|
|
- MqttServer server = MqttServer.create(vertx);
|
|
|
- VertxMqttServer vertxMqttServer = new VertxMqttServer("123");
|
|
|
- vertxMqttServer.setMqttServer(Collections.singleton(server));
|
|
|
+// vertx.setTimer(100,event -> {});
|
|
|
+// MqttServer server = MqttServer.create(vertx);
|
|
|
+// VertxMqttServer vertxMqttServer = new VertxMqttServer("123");
|
|
|
+// vertxMqttServer.setMqttServer(Collections.singleton(server));
|
|
|
//
|
|
|
// server.endpointHandler(endpoint -> {
|
|
|
// endpoint
|
|
|
@@ -42,25 +34,43 @@ class MqttClientProviderTest {
|
|
|
// }).listen(11223);
|
|
|
|
|
|
MqttClientProvider sub = new MqttClientProvider(id -> Mono.empty(), vertx,new MockEnvironment());
|
|
|
+ EventBus eventBus =new BrokerEventBus();
|
|
|
|
|
|
MqttClientProperties properties = new MqttClientProperties();
|
|
|
properties.setHost("123.56.154.53");
|
|
|
properties.setPort(1883);
|
|
|
properties.setOptions(new MqttClientOptions());
|
|
|
properties.setClientId(UUID.randomUUID().toString());
|
|
|
+ eventBus.subscribe(Subscription.of("ceshi","/test",Subscription.DEFAULT_FEATURES))
|
|
|
+ .doOnNext(s->{
|
|
|
+ String topic = s.getTopic();
|
|
|
+// System.out.println(topic);
|
|
|
+ })
|
|
|
+ .map(s->{
|
|
|
+ String topic = s.getTopic();
|
|
|
+// System.out.println(topic);
|
|
|
+ return topic;
|
|
|
+ }).subscribe();
|
|
|
+
|
|
|
CompletableFuture.runAsync(()->{
|
|
|
|
|
|
VertxMqttClient client = sub.createNetwork(properties);
|
|
|
Flux<MqttMessage> subscribe = client.subscribe(Arrays.asList("/test"));
|
|
|
subscribe
|
|
|
+ .map(mqttMessage -> {
|
|
|
+
|
|
|
+ Long block = eventBus.publish(mqttMessage.getTopic(), mqttMessage).block();
|
|
|
+ return mqttMessage;
|
|
|
+ })
|
|
|
.map(MqttMessage::getPayload)
|
|
|
.map(payload -> payload.toString(StandardCharsets.UTF_8))
|
|
|
.doOnNext(string->{
|
|
|
+
|
|
|
System.out.println("处理缓存>>>>>>>>>>"+string);
|
|
|
})
|
|
|
- .subscribe(string->{
|
|
|
- System.out.println("接收到数据>>>>>>>>>>"+string);
|
|
|
- });
|
|
|
+ .subscribe(string->{
|
|
|
+ System.out.println("接收到数据>>>>>>>>>>"+string);
|
|
|
+ });
|
|
|
// .as(StepVerifier::create)
|
|
|
// .expectNext("test")
|
|
|
// .verifyComplete();
|
|
|
@@ -74,26 +84,26 @@ class MqttClientProviderTest {
|
|
|
} catch (InterruptedException e) {
|
|
|
e.printStackTrace();
|
|
|
}
|
|
|
- MqttClientProvider pub = new MqttClientProvider(id -> Mono.empty(), vertx, new MockEnvironment());
|
|
|
- MqttClientProperties pubProperties = new MqttClientProperties();
|
|
|
- pubProperties.setHost("123.56.154.53");
|
|
|
- pubProperties.setPort(1883);
|
|
|
- pubProperties.setOptions(new MqttClientOptions());
|
|
|
- pubProperties.setClientId(UUID.randomUUID().toString());
|
|
|
- VertxMqttClient pubClient = pub.createNetwork(pubProperties);
|
|
|
- SimpleMqttMessage simpleMqttMessage = new SimpleMqttMessage();
|
|
|
- simpleMqttMessage.setPayload(Unpooled.wrappedBuffer("123test".getBytes()));
|
|
|
- simpleMqttMessage.setTopic("/test");
|
|
|
- simpleMqttMessage.setMessageId(123);
|
|
|
- simpleMqttMessage.setQosLevel(1);
|
|
|
- pubClient.publish(simpleMqttMessage).subscribe();
|
|
|
- simpleMqttMessage.setPayload(Unpooled.wrappedBuffer("123".getBytes()));
|
|
|
- simpleMqttMessage.setTopic("/test");
|
|
|
- simpleMqttMessage.setMessageId(456);
|
|
|
- simpleMqttMessage.setQosLevel(2);
|
|
|
- simpleMqttMessage.setMessageId(456);
|
|
|
- pubClient.publish(simpleMqttMessage).subscribe();
|
|
|
-// });
|
|
|
+// MqttClientProvider pub = new MqttClientProvider(id -> Mono.empty(), vertx, new MockEnvironment());
|
|
|
+// MqttClientProperties pubProperties = new MqttClientProperties();
|
|
|
+// pubProperties.setHost("123.56.154.53");
|
|
|
+// pubProperties.setPort(1883);
|
|
|
+// pubProperties.setOptions(new MqttClientOptions());
|
|
|
+// pubProperties.setClientId(UUID.randomUUID().toString());
|
|
|
+// VertxMqttClient pubClient = pub.createNetwork(pubProperties);
|
|
|
+// SimpleMqttMessage simpleMqttMessage = new SimpleMqttMessage();
|
|
|
+// simpleMqttMessage.setPayload(Unpooled.wrappedBuffer("123test".getBytes()));
|
|
|
+// simpleMqttMessage.setTopic("/test");
|
|
|
+// simpleMqttMessage.setMessageId(123);
|
|
|
+// simpleMqttMessage.setQosLevel(1);
|
|
|
+// pubClient.publish(simpleMqttMessage).subscribe();
|
|
|
+// simpleMqttMessage.setPayload(Unpooled.wrappedBuffer("123".getBytes()));
|
|
|
+// simpleMqttMessage.setTopic("/test");
|
|
|
+// simpleMqttMessage.setMessageId(456);
|
|
|
+// simpleMqttMessage.setQosLevel(2);
|
|
|
+// simpleMqttMessage.setMessageId(456);
|
|
|
+// pubClient.publish(simpleMqttMessage).subscribe();
|
|
|
+//// });
|
|
|
while (true){
|
|
|
|
|
|
}
|
