协议网关修改

pom.xml

@@ -11,7 +11,7 @@
         <module>manager</module>
         <module>dao</module>
         <module>tppa-server</module>
-        <module>protocol</module>
+        <module>protocol-gateway</module>
     </modules>
     <parent>
         <groupId>org.springframework.boot</groupId>
@@ -149,6 +149,13 @@
                 <version>1.2.2</version>
             </dependency>
 
+            <!-- https://mvnrepository.com/artifact/org.apache.pulsar/pulsar-client -->
+            <dependency>
+                <groupId>org.apache.pulsar</groupId>
+                <artifactId>pulsar-client</artifactId>
+                <version>2.9.1</version>
+            </dependency>
+
             <dependency>
                 <groupId>cc.iotkit</groupId>
                 <artifactId>model</artifactId>
@@ -179,6 +186,12 @@
                 <version>${project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>cc.iotkit</groupId>
+                <artifactId>gateway-api</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>
 

protocol/pom.xml → protocol-gateway/gateway-api/pom.xml

@@ -3,13 +3,13 @@
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
-        <artifactId>iotkit-parent</artifactId>
+        <artifactId>protocol-gateway</artifactId>
         <groupId>cc.iotkit</groupId>
         <version>0.0.1-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>protocol</artifactId>
+    <artifactId>gateway-api</artifactId>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
@@ -19,23 +19,23 @@
     <dependencies>
 
         <dependency>
-            <groupId>org.springframework</groupId>
-            <artifactId>spring-context-support</artifactId>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot</artifactId>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-openfeign</artifactId>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>common</artifactId>
         </dependency>
 
     </dependencies>

protocol-gateway/gateway-api/src/main/java/cc/iotkit/protocol/DeviceBehaviour.java

@@ -0,0 +1,40 @@
+package cc.iotkit.protocol;
+
+
+/**
+ * 设备行为接口
+ */
+public interface DeviceBehaviour {
+
+    /**
+     * 设备注册
+     */
+    Result register(RegisterInfo info);
+
+    /**
+     * 设备注销
+     */
+    Result deregister(DeregisterInfo info);
+
+    /**
+     * 设备上线
+     */
+    Result online(String productKey, String deviceName);
+
+    /**
+     * 设备离线
+     */
+    Result offline(String productKey, String deviceName);
+
+    /**
+     * 设备消息上报
+     */
+    void messageReport(DeviceMessage msg);
+
+    /**
+     * OTA消息上报
+     */
+    void otaProgressReport(OtaMessage msg);
+
+
+}

protocol-gateway/gateway-api/src/main/java/cc/iotkit/protocol/Result.java

@@ -0,0 +1,16 @@
+package cc.iotkit.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Result {
+
+    private boolean success;
+
+    private String content;
+
+}

protocol-gateway/gateway-api/src/main/java/cc/iotkit/protocol/client/DeviceBehaviourClient.java

@@ -0,0 +1,98 @@
+package cc.iotkit.protocol.client;
+
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.protocol.*;
+import okhttp3.*;
+import org.apache.pulsar.client.api.PulsarClient;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class DeviceBehaviourClient implements DeviceBehaviour {
+
+    private final String server;
+
+    private final OkHttpClient httpClient;
+
+    private PulsarClient client;
+
+    public DeviceBehaviourClient(String server) {
+        this.server = server.replaceAll("/$", "");
+        httpClient = new OkHttpClient.Builder()
+                .connectTimeout(10, TimeUnit.SECONDS)
+                .readTimeout(30, TimeUnit.SECONDS)
+                .build();
+    }
+
+    public Request buildRequest(String url, String method, String type, Map<String, Object> data) {
+        Request.Builder requestBuilder = new Request.Builder()
+                .url(url);
+
+        RequestBody requestBody;
+        if ("json".equals(type)) {
+            requestBody = RequestBody.create(MediaType.get("application/json; charset=utf-8"),
+                    JsonUtil.toJsonString(data));
+        } else {
+            FormBody.Builder builder = new FormBody.Builder();
+            data.forEach((key, val) -> builder.add(key, val.toString()));
+            requestBody = builder.build();
+        }
+        requestBuilder.method(method.toUpperCase(), requestBody);
+        return requestBuilder.build();
+    }
+
+    private <T> Result invoke(String path, T data) {
+        Request request = buildRequest(server + path, "post", "form",
+                (data instanceof Map) ? (Map<String, Object>) data :
+                        JsonUtil.parse(JsonUtil.toJsonString(data), Map.class));
+
+        Call call = httpClient.newCall(request);
+        try {
+            Response response = call.execute();
+            if (!response.isSuccessful() || response.body() == null) {
+                return new Result(false, "接口调用失败");
+            }
+            String content = response.body().string();
+            return JsonUtil.parse(content, Result.class);
+        } catch (Throwable e) {
+            return new Result(false, "接口调用失败:" + e.getMessage());
+        }
+    }
+
+
+    @Override
+    public Result register(RegisterInfo info) {
+        return invoke("/register", info);
+    }
+
+    @Override
+    public Result deregister(DeregisterInfo info) {
+        return invoke("/deregister", info);
+    }
+
+    @Override
+    public Result online(String productKey, String deviceName) {
+        Map<String, Object> data = new HashMap<>();
+        data.put("productKey", productKey);
+        data.put("deviceName", deviceName);
+        return invoke("/online", data);
+    }
+
+    @Override
+    public Result offline(String productKey, String deviceName) {
+        Map<String, Object> data = new HashMap<>();
+        data.put("productKey", productKey);
+        data.put("deviceName", deviceName);
+        return invoke("/offline", data);
+    }
+
+    @Override
+    public void messageReport(DeviceMessage msg) {
+
+    }
+
+    @Override
+    public void otaProgressReport(OtaMessage msg) {
+    }
+}

protocol-gateway/gateway-server/pom.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>protocol-gateway</artifactId>
+        <groupId>cc.iotkit</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>gateway-server</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>gateway-api</artifactId>
+        </dependency>
+
+    </dependencies>
+
+</project>

protocol-gateway/gateway-server/src/main/java/cc/iotkit/protocol/server/Test1.java

@@ -0,0 +1,66 @@
+package cc.iotkit.protocol.server;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.pulsar.client.api.*;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.functions.FunctionConfig;
+
+import java.util.regex.Pattern;
+
+public class Test1 {
+
+    public static void main(String[] args) throws PulsarClientException, InterruptedException {
+//        FunctionConfig functionConfig = new FunctionConfig();
+//        functionConfig.setTenant("tenant");
+//        functionConfig.setNamespace("namespace");
+//        functionConfig.setName("functionName");
+//        functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+//        functionConfig.setParallelism(1);
+//        functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction");
+//        functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
+//        functionConfig.setTopicsPattern(sourceTopicPattern);
+//        functionConfig.setSubName(subscriptionName);
+//        functionConfig.setAutoAck(true);
+//        functionConfig.setOutput(sinkTopic);
+//        admin.functions().createFunction(functionConfig, fileName);
+
+
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl("pulsar://localhost:6650")
+                .build();
+
+        client.newConsumer(JSONSchema.of(Msg.class))
+                .topicsPattern(Pattern.compile("persistent://public/default/test.*"))
+                .subscriptionName("test1")
+                .consumerName("test1")
+                .messageListener((MessageListener<Msg>) (consumer, msg) -> {
+                    Msg m = msg.getValue();
+                    System.out.printf("=====received:%s,%s%n", m.getIdentifier(), m.getDeviceId());
+                }).subscribe();
+
+        Producer<Msg> producer = client.newProducer(JSONSchema.of(Msg.class))
+                .topic("test1234")
+                .create();
+
+        for (int i = 0; i < 1000; i++) {
+            producer.send(new Msg("test", "xxxx11222333" + i));
+            Thread.sleep(500);
+        }
+
+
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class Msg {
+
+        private String identifier;
+
+        private String deviceId;
+
+    }
+
+}

protocol/src/main/resources/spring.factories → protocol-gateway/gateway-server/src/main/resources/spring.factories

@@ -1 +1 @@
-org.springframework.boot.autoconfigure.EnableAutoConfiguration= cc.iotkit.protocol.config.ProtocolConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration= cc.iotkit.protocol.config.GatewayConfig

protocol-gateway/pom.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>iotkit-parent</artifactId>
+        <groupId>cc.iotkit</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>protocol-gateway</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>gateway-api</module>
+        <module>gateway-server</module>
+    </modules>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+</project>

protocol/src/main/java/cc/iotkit/protocol/DeviceBehaviour.java

@@ -1,46 +0,0 @@
-package cc.iotkit.protocol;
-
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-
-/**
- * 设备行为接口
- */
-public interface DeviceBehaviour {
-
-    /**
-     * 设备注册
-     */
-    @PostMapping("/register")
-    Result register(@RequestBody RegisterInfo info);
-
-    /**
-     * 设备注销
-     */
-    @PostMapping("/deregister")
-    Result deregister(@RequestBody DeregisterInfo info);
-
-    /**
-     * 设备上线
-     */
-    @PostMapping("/online")
-    void online(String productKey, String deviceName);
-
-    /**
-     * 设备离线
-     */
-    @PostMapping("/offline")
-    void offline(String productKey, String deviceName);
-
-    /**
-     * 设备消息上报
-     */
-    void messageReport(DeviceMessage msg);
-
-    /**
-     * OTA消息上报
-     */
-    void otaProgressReport(OtaMessage msg);
-
-
-}

protocol/src/main/java/cc/iotkit/protocol/Result.java

@@ -1,9 +0,0 @@
-package cc.iotkit.protocol;
-
-import lombok.Data;
-
-@Data
-public class Result {
-
-
-}

protocol/src/main/java/cc/iotkit/protocol/config/ProtocolConfiguration.java

@@ -1,24 +0,0 @@
-package cc.iotkit.protocol.config;
-
-import cc.iotkit.protocol.DeviceBehaviour;
-import cc.iotkit.protocol.impl.DeviceBehaviourImpl;
-import feign.Client;
-import feign.Contract;
-import feign.codec.Decoder;
-import feign.codec.Encoder;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-@Configuration
-public class ProtocolConfiguration {
-
-    @Value("${protocol.server}")
-    private String protocolServer;
-
-    @Bean
-    public DeviceBehaviour getDeviceBehaviour(Decoder decoder, Encoder encoder, Client client, Contract contract) {
-        return new DeviceBehaviourImpl(protocolServer, decoder, encoder, client, contract);
-    }
-
-}

protocol/src/main/java/cc/iotkit/protocol/impl/DeviceBehaviourImpl.java

@@ -1,66 +0,0 @@
-package cc.iotkit.protocol.impl;
-
-import cc.iotkit.protocol.*;
-import feign.Client;
-import feign.Contract;
-import feign.Feign;
-import feign.codec.Decoder;
-import feign.codec.Encoder;
-import org.springframework.cloud.openfeign.FeignClientsConfiguration;
-import org.springframework.context.annotation.Import;
-
-@Import(FeignClientsConfiguration.class)
-public class DeviceBehaviourImpl implements DeviceBehaviour {
-
-    private final Feign.Builder builder;
-
-    private final String protocolServer;
-
-    private DeviceBehaviour target;
-
-    public DeviceBehaviourImpl(String protocolServer, Decoder decoder,
-                               Encoder encoder, Client client, Contract contract) {
-        this.protocolServer = protocolServer;
-        this.builder = Feign.builder()
-                .client(client)
-                .encoder(encoder)
-                .decoder(decoder)
-                .contract(contract);
-    }
-
-    private DeviceBehaviour behaviour() {
-        if (target == null) {
-            target = this.builder.target(DeviceBehaviour.class, protocolServer);
-        }
-        return target;
-    }
-
-    @Override
-    public Result register(RegisterInfo info) {
-        return behaviour().register(info);
-    }
-
-    @Override
-    public Result deregister(DeregisterInfo info) {
-        return behaviour().deregister(info);
-    }
-
-    @Override
-    public void online(String productKey, String deviceName) {
-        behaviour().online(productKey, deviceName);
-    }
-
-    @Override
-    public void offline(String productKey, String deviceName) {
-        behaviour().offline(productKey, deviceName);
-    }
-
-    @Override
-    public void messageReport(DeviceMessage msg) {
-
-    }
-
-    @Override
-    public void otaProgressReport(OtaMessage msg) {
-    }
-}