sample

sample/build.gradle

@@ -31,10 +31,14 @@ dependencies {
     //compile "com.github.linyuzai:concept-sync-waiting:1.0.0"
     implementation project(path: ':concept-sync-waiting', configuration: 'default')
 
-    //compile "com.github.linyuzai:concept-plugin-jar:1.0.0"
+    //compile "com.github.linyuzai:concept-plugin-jar:1.0.1"
     implementation project(path: ':concept-plugin:concept-plugin-core', configuration: 'default')
     implementation project(path: ':concept-plugin:concept-plugin-jar', configuration: 'default')
 
+    implementation project(path: ':concept-event:concept-event-core', configuration: 'default')
+    implementation project(path: ':concept-event:concept-event-kafka', configuration: 'default')
+    implementation project(path: ':concept-event:concept-event-spring-boot-starter', configuration: 'default')
+
     implementation project(path: ':concept-connection-loadbalance:concept-connection-loadbalance-core', configuration: 'default')
     implementation project(path: ':concept-connection-loadbalance:concept-connection-loadbalance-websocket', configuration: 'default')
     implementation project(path: ':concept-connection-loadbalance:concept-connection-loadbalance-autoconfigure', configuration: 'default')
@@ -57,6 +61,10 @@ dependencies {
     //implementation 'io.springfox:springfox-swagger2:2.9.2'
     //implementation 'io.springfox:springfox-swagger-ui:2.9.2'
 
+    implementation 'org.springframework.kafka:spring-kafka'
+
+    implementation 'io.socket:socket.io-server:4.0.1'
+
     implementation 'org.springframework.cloud:spring-cloud-starter'
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'

sample/src/main/java/com/github/linyuzai/concept/sample/ConceptSampleApplication.java

@@ -1,14 +1,83 @@
 package com.github.linyuzai.concept.sample;
 
+import com.github.linyuzai.concept.sample.sync.MapBlockingQueue;
+import lombok.SneakyThrows;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+import java.io.Serializable;
+import java.lang.reflect.*;
+import java.util.*;
 //import springfox.documentation.swagger2.annotations.EnableSwagger2;
 
 //@EnableSwagger2
-@SpringBootApplication
+@SpringBootApplication(scanBasePackages = "com.github.linyuzai.concept.sample.event")
 public class ConceptSampleApplication {
 
+    private final MapBlockingQueue<String> queue = new MapBlockingQueue<>();
+
     public static void main(String[] args) {
         SpringApplication.run(ConceptSampleApplication.class, args);
     }
+
+    @SneakyThrows
+    public static void test() {
+        long sleep = 100;
+        ConceptSampleApplication application = new ConceptSampleApplication();
+        application.startPut("A");
+        Thread.sleep(sleep);
+        application.startPut("B");
+        Thread.sleep(sleep);
+        application.startPut("C");
+        Thread.sleep(sleep);
+        application.startPut("D");
+        Thread.sleep(sleep);
+        application.startTake();
+    }
+
+    @SneakyThrows
+    public void startTake() {
+        while (true) {
+            String s = queue.take();
+            System.out.println(s);
+            System.out.println(queue.size() + ":" + queue.getMap());
+            Thread.sleep(1000);
+        }
+    }
+
+    public void startPut(String s) {
+        new Thread() {
+
+            int i;
+
+            @SneakyThrows
+            @Override
+            public void run() {
+                while (true) {
+                    queue.put(s, s + i++);
+                    Thread.sleep(100);
+                }
+            }
+        }.start();
+
+    }
+
+    public class StringArrayList extends ArrayList<String> {
+
+    }
+
+    public class ClassArrayList extends ArrayList<Class<? extends Serializable>> {
+
+    }
+
+    public class ClassesArrayList extends ArrayList<Class<? extends Serializable>[]> {
+
+    }
+
+    public class Custom<T extends String & Serializable & Cloneable> {
+    }
+
+    public interface CustomEx extends Serializable, Cloneable {
+
+    }
 }

sample/src/main/java/com/github/linyuzai/concept/sample/connection/loadbalance/websocket/socketio/SocketIOWebSocketServer.java

@@ -0,0 +1,34 @@
+package com.github.linyuzai.concept.sample.connection.loadbalance.websocket.socketio;
+
+import io.socket.engineio.server.Emitter;
+import io.socket.engineio.server.EngineIoServer;
+import io.socket.socketio.server.SocketIoServer;
+import org.springframework.web.socket.server.HandshakeInterceptor;
+
+import javax.servlet.annotation.WebServlet;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+//@WebServlet("/websocket/*")
+public class SocketIOWebSocketServer extends HttpServlet {
+
+    private final EngineIoServer mEngineIoServer = new EngineIoServer();
+    private final SocketIoServer mSocketIoServer = new SocketIoServer(mEngineIoServer);
+
+    public SocketIOWebSocketServer() {
+        mSocketIoServer.namespace("/websocket/").on("connection", new Emitter.Listener() {
+            @Override
+            public void call(Object... args) {
+
+            }
+        });
+    }
+
+    @Override
+    protected void service(HttpServletRequest request, HttpServletResponse response) throws IOException {
+        mEngineIoServer.handleRequest(request, response);
+        //mSocketIoServer.namespace("/websocket/*").broadcast();
+    }
+}

sample/src/main/java/com/github/linyuzai/concept/sample/event/EventConfig.java

@@ -0,0 +1,9 @@
+package com.github.linyuzai.concept.sample.event;
+
+import com.github.linyuzai.event.autoconfigure.EnableEventConcept;
+import org.springframework.context.annotation.Configuration;
+
+@EnableEventConcept
+@Configuration
+public class EventConfig {
+}

sample/src/main/java/com/github/linyuzai/concept/sample/event/kafka/KafkaEventConfig.java

@@ -0,0 +1,23 @@
+package com.github.linyuzai.concept.sample.event.kafka;
+
+import com.github.linyuzai.event.core.codec.EventDecoder;
+import com.github.linyuzai.event.core.codec.EventEncoder;
+import com.github.linyuzai.event.core.codec.JacksonEventDecoder;
+import com.github.linyuzai.event.core.codec.JacksonEventEncoder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+//@EnableKafka
+@Configuration
+public class KafkaEventConfig {
+
+    @Bean
+    public EventEncoder eventEncoder() {
+        return new JacksonEventEncoder();
+    }
+
+    @Bean
+    public EventDecoder eventDecoder() {
+        return new JacksonEventDecoder();
+    }
+}

sample/src/main/java/com/github/linyuzai/concept/sample/event/kafka/KafkaEventController.java

@@ -0,0 +1,20 @@
+package com.github.linyuzai.concept.sample.event.kafka;
+
+import com.github.linyuzai.event.core.concept.EventConcept;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/concept-event")
+public class KafkaEventController {
+
+    @Autowired
+    private EventConcept concept;
+
+    @GetMapping("/kafka")
+    public void send() {
+        concept.event("123").publish();
+    }
+}

sample/src/main/java/com/github/linyuzai/concept/sample/event/kafka/KafkaEventSubscriberRegister.java

@@ -0,0 +1,43 @@
+package com.github.linyuzai.concept.sample.event.kafka;
+
+import com.github.linyuzai.event.core.concept.EventConcept;
+import com.github.linyuzai.event.core.context.EventContext;
+import com.github.linyuzai.event.kafka.exchange.KafkaEngineExchange;
+import com.github.linyuzai.event.kafka.endpoint.KafkaEventEndpoint;
+import com.github.linyuzai.event.kafka.subscriber.TopicKafkaEventSubscriber;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.support.Acknowledgment;
+
+@Configuration
+public class KafkaEventSubscriberRegister implements ApplicationRunner {
+
+    @Autowired
+    public EventConcept concept;
+
+    //@KafkaListener(topics = "sample", containerFactory = "localKafkaListenerContainerFactory")
+    public void receiveLocal(String msg, Acknowledgment acknowledgment) {
+        System.out.println("local-" + msg);
+        acknowledgment.acknowledge();
+    }
+
+    //@KafkaListener(topics = "sample", containerFactory = "devKafkaListenerContainerFactory")
+    public void receiveDev(String msg, Acknowledgment acknowledgment) {
+        System.out.println("dev-" + msg);
+        acknowledgment.acknowledge();
+    }
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        concept.event(String.class)
+                .exchange(new KafkaEngineExchange())
+                .subscribe(new TopicKafkaEventSubscriber<String>("sample") {
+                    @Override
+                    public void onEvent(String event, KafkaEventEndpoint endpoint, EventContext context) {
+                        System.out.println("subscribe-" + endpoint.getName() + ":" + event);
+                    }
+                });
+    }
+}

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/ConceptPluginController.java

@@ -649,30 +649,30 @@ public void onExtract(Map<Object, ? extends CustomPlugin> plugin) {
                     System.out.println(append("InstanceExtractor<Map<Object, ? extends CustomPlugin>>: ") + plugin);
                 }
             })*/
-            .addExtractor(new InstanceExtractor<CustomPlugin>() {
+            /*.addExtractor(new InstanceExtractor<CustomPlugin>() {
                 @Override
                 public void onExtract(CustomPlugin plugin) {
                     ConceptPluginController.this.plugin = plugin;
                 }
-            })
+            })*/
             /*.addExtractor(new ClassExtractor<Class<? extends CustomPlugin>>() {
 
                 @Override
                 public void onExtract(Class<? extends CustomPlugin> plugin) {
                 }
             })*/
-            //.extractTo(this)//自动匹配回调添加了@OnPluginExtract注解的方法参数
+            .extractTo(this)//自动匹配回调添加了@OnPluginExtract注解的方法参数
             .addEventListener(new PluginLoadLogger(log::info))
             .build();
 
     private final PluginAutoLoader loader = new WatchServicePluginAutoLoader.Builder()
             .locations(
                     new PluginLocation.Builder()
-                            .path("/Users/concept/plugin")
+                            .path("/Users/tanghanzheng/concept/plugin")
                             .filter(it -> it.endsWith(".jar"))
                             .build(),
                     new PluginLocation.Builder()
-                            .path("/Users/concept/plugin2")
+                            .path("/Users/tanghanzheng/concept/plugin2")
                             .filter(it -> it.endsWith(".jar"))
                             .build())
             .executor(Executors.newSingleThreadExecutor())

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/ConceptPluginSample.java

@@ -4,6 +4,7 @@
 import com.github.linyuzai.plugin.core.autoload.PluginLocation;
 import com.github.linyuzai.plugin.core.autoload.WatchServicePluginAutoLoader;
 import com.github.linyuzai.plugin.core.extract.OnPluginExtract;
+import com.github.linyuzai.plugin.core.match.PluginProperties;
 import com.github.linyuzai.plugin.core.util.PluginLoadLogger;
 import com.github.linyuzai.plugin.jar.autoload.JarNotifier;
 import com.github.linyuzai.plugin.jar.concept.JarPluginConcept;
@@ -13,7 +14,9 @@
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 
 @Slf4j
@@ -23,25 +26,52 @@ public class ConceptPluginSample {
      * 插件提取配置
      */
     private final JarPluginConcept concept = new JarPluginConcept.Builder()
-            //添加类提取器
+            //回调到标注了@OnPluginExtract的方法
             .extractTo(this)
-            .addEventListener(new PluginLoadLogger(log::info))
             .build();
 
+    /**
+     * 缓存设备类型和对应的操作对象
+     */
+    private final Map<String, DeviceOperation> operationMap = new ConcurrentHashMap<>();
+
+    /**
+     * 插件匹配回调
+     *
+     * @param operation  匹配到的 DeviceOperation 实例
+     * @param deviceType 配置文件中定义的设备类型
+     */
     @OnPluginExtract
-    public void onPluginExtract(Class<? extends CustomPlugin> pluginClass, Properties properties) {
-        //任意一个参数匹配上都会触发回调
+    public void onPluginExtract(DeviceOperation operation, @PluginProperties("device.type") String deviceType) {
+        operationMap.put(deviceType, operation);
     }
 
     /**
-     * 加载一个 jar 插件
+     * 加载 jar 插件
      *
      * @param filePath jar 文件路径
      */
     public void load(String filePath) {
         concept.load(filePath);
     }
 
+    /**
+     * 操作一个设备
+     *
+     * @param device  设备对象
+     * @param opType  操作类型
+     * @param opValue 操作值
+     * @return 操作结果
+     */
+    public OperationResult operate(Device device, String opType, Object opValue) {
+        String deviceType = device.getDeviceType();
+        DeviceOperation operation = operationMap.get(deviceType);
+        if (operation == null) {
+            throw new DeviceOperationNotFoundException(deviceType + " not found");
+        }
+        return operation.operate(device, opType, opValue);
+    }
+
     private final PluginAutoLoader loader = new WatchServicePluginAutoLoader.Builder()
             .locations(new PluginLocation.Builder()
                     .path("/Users/concept/plugin/")

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/Device.java

@@ -0,0 +1,6 @@
+package com.github.linyuzai.concept.sample.plugin;
+
+public interface Device {
+
+    String getDeviceType();
+}

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/DeviceOperation.java

@@ -0,0 +1,15 @@
+package com.github.linyuzai.concept.sample.plugin;
+
+public interface DeviceOperation {
+
+    /**
+     * 设备操作
+     *
+     * @param device  设备
+     * @param opType  操作类型
+     * @param opValue 操作值
+     * @return 操作结果
+     */
+    OperationResult operate(Device device, String opType, Object opValue);
+
+}

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/DeviceOperationNotFoundException.java

@@ -0,0 +1,7 @@
+package com.github.linyuzai.concept.sample.plugin;
+
+public class DeviceOperationNotFoundException extends RuntimeException {
+    public DeviceOperationNotFoundException(String message) {
+        super(message);
+    }
+}

sample/src/main/java/com/github/linyuzai/concept/sample/plugin/OperationResult.java

@@ -0,0 +1,5 @@
+package com.github.linyuzai.concept.sample.plugin;
+
+public class OperationResult {
+
+}