优化协议转换

jetlinks · Mar 13, 2019 · bfb066e · bfb066e
1 parent c51cf62
commit bfb066e
Show file tree

Hide file tree

Showing 6 changed files with 19 additions and 15 deletions.
diff --git a/src/main/java/org/jetlinks/gateway/session/DefaultDeviceSessionManager.java b/src/main/java/org/jetlinks/gateway/session/DefaultDeviceSessionManager.java
@@ -27,9 +27,4 @@ public DeviceClient unregister(String clientId) {
         return repository.remove(clientId);
     }
 
-    @Override
-    public DeviceSession getSession(String sessionId) {
-
-        return null;
-    }
 }
diff --git a/src/main/java/org/jetlinks/gateway/session/DeviceClient.java b/src/main/java/org/jetlinks/gateway/session/DeviceClient.java
@@ -1,6 +1,7 @@
 package org.jetlinks.gateway.session;
 
 import org.jetlinks.protocol.message.codec.EncodedMessage;
+import org.jetlinks.protocol.message.codec.Transport;
 
 /**
  * @author zhouhao
@@ -17,6 +18,8 @@ public interface DeviceClient {
 
     void send(EncodedMessage encodedMessage);
 
+    Transport getTransport();
+
     void close();
 
     void ping();

diff --git a/src/main/java/org/jetlinks/gateway/session/DeviceSessionManager.java b/src/main/java/org/jetlinks/gateway/session/DeviceSessionManager.java
@@ -12,6 +12,4 @@ public interface DeviceSessionManager {
 
     DeviceClient unregister(String clientId);
 
-    DeviceSession getSession(String sessionId);
-
 }
diff --git a/src/main/java/org/jetlinks/gateway/vertx/mqtt/MqttClient.java b/src/main/java/org/jetlinks/gateway/vertx/mqtt/MqttClient.java
@@ -7,6 +7,7 @@
 import org.jetlinks.gateway.session.DeviceClient;
 import org.jetlinks.protocol.message.codec.EncodedMessage;
 import org.jetlinks.protocol.message.codec.MqttMessage;
+import org.jetlinks.protocol.message.codec.Transport;
 
 import java.nio.charset.StandardCharsets;
 
@@ -57,6 +58,11 @@ public void close() {
         }
     }
 
+    @Override
+    public Transport getTransport() {
+        return Transport.MQTT;
+    }
+
     @Override
     public void send(EncodedMessage encodedMessage) {
         if (encodedMessage instanceof MqttMessage) {
@@ -68,7 +74,7 @@ public void send(EncodedMessage encodedMessage) {
             }
             endpoint.publish(message.getTopic(), buffer, MqttQoS.AT_MOST_ONCE, false, false);
         } else {
-            log.error("不支持发送消息{}到mqtt:", encodedMessage);
+            log.error("不支持发送消息{}到MQTT:", encodedMessage);
         }
     }
 

diff --git a/src/main/java/org/jetlinks/gateway/vertx/mqtt/MqttServer.java b/src/main/java/org/jetlinks/gateway/vertx/mqtt/MqttServer.java
@@ -16,6 +16,7 @@
 import org.jetlinks.protocol.ProtocolSupports;
 import org.jetlinks.protocol.message.codec.EncodedMessage;
 import org.jetlinks.protocol.message.DeviceMessage;
+import org.jetlinks.protocol.message.codec.Transport;
 import org.jetlinks.registry.api.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -174,8 +175,8 @@ protected void accept(MqttEndpoint endpoint) {
                             ProtocolSupport protocolSupport = getProtocol(protocol);
                             //转换消息
                             DeviceMessage deviceMessage = protocolSupport
-                                    .getMessageConverter()
-                                    .convert(EncodedMessage.mqtt(clientId, topicName, buffer.getByteBuf()));
+                                    .getMessageCodec()
+                                    .decode(Transport.MQTT, EncodedMessage.mqtt(clientId, topicName, buffer.getByteBuf()));
                             if (messageConsumer != null) {
                                 messageConsumer.accept(deviceMessage);
                             }

diff --git a/src/test/java/org/jetlinks/gateway/vertx/mqtt/MockProtocolSupports.java b/src/test/java/org/jetlinks/gateway/vertx/mqtt/MockProtocolSupports.java
@@ -9,8 +9,9 @@
 import org.jetlinks.protocol.message.DeviceMessage;
 import org.jetlinks.protocol.message.codec.DeviceMessageCodec;
 import org.jetlinks.protocol.message.codec.EncodedMessage;
+import org.jetlinks.protocol.message.codec.Transport;
 import org.jetlinks.protocol.message.property.ReadPropertyMessageReply;
-import org.jetlinks.protocol.metadata.DeviceMetadataParser;
+import org.jetlinks.protocol.metadata.DeviceMetadataCodec;
 
 import java.nio.charset.StandardCharsets;
 
@@ -23,17 +24,17 @@ public class MockProtocolSupports implements ProtocolSupports {
     public ProtocolSupport getProtocol(String protocol) {
         return new ProtocolSupport() {
             @Override
-            public DeviceMessageCodec getMessageConverter() {
+            public DeviceMessageCodec getMessageCodec() {
 
                 return new DeviceMessageCodec() {
                     @Override
-                    public EncodedMessage convert(DeviceMessage deviceMessage) {
+                    public EncodedMessage encode(Transport transport, DeviceMessage deviceMessage) {
                         return EncodedMessage.mqtt(deviceMessage.getDeviceId(), "command",
                                 Unpooled.copiedBuffer(deviceMessage.toJson().toJSONString().getBytes()));
                     }
 
                     @Override
-                    public DeviceMessage convert(EncodedMessage message) {
+                    public DeviceMessage decode(Transport transport, EncodedMessage message) {
                         JSONObject jsonObject = JSON.parseObject(message.getByteBuf().toString(StandardCharsets.UTF_8));
                         if ("read-property".equals(jsonObject.get("type"))) {
                             return jsonObject.toJavaObject(ReadPropertyMessageReply.class);
@@ -44,7 +45,7 @@ public DeviceMessage convert(EncodedMessage message) {
             }
 
             @Override
-            public DeviceMetadataParser getMetadataParser() {
+            public DeviceMetadataCodec getMetadataCodec() {
                 throw new UnsupportedOperationException();
             }
         };