应用事件相关进一步封装及测试

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Channel.java

@@ -0,0 +1,53 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base;
+
+/**
+ * @author: aoshiguchen
+ * @date: 2022/9/29
+ */
+public interface Channel<D,R extends Receiver,Dis extends Dispatcher> {
+
+    /**
+     * 注册接收者
+     * @param receiver 接收者
+     */
+    void registerReceiver(R receiver);
+
+    /**
+     * 注销接收者
+     * @param receiver 接收者
+     */
+    void unRegisterReceiver(R receiver);
+
+    /**
+     * 设置调度器
+     * @param dispatcher 调度器
+     */
+    void setDispatcher(Dis dispatcher);
+
+    /**
+     * 发布消息
+     * @param msg 消息
+     */
+    void publish(D msg);
+}

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java

@@ -0,0 +1,46 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base;
+
+/**
+ * @author: aoshiguchen
+ * @date: 2022/9/29
+ */
+public interface Publisher<D,Ch extends Channel> {
+    /**
+     * 发布消息
+     * @param msg 消息
+     */
+    void publish(D msg);
+
+    /**
+     * 注册渠道
+     * @param channel 渠道
+     */
+    void registerChannel(Ch channel);
+
+    /**
+     * 注销渠道
+     * @param channel 渠道
+     */
+    void unRegisterChannel(Ch channel);
+}

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Receiver.java

@@ -19,12 +19,17 @@
  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  * SOFTWARE.
  */
-package fun.asgc.neutrino.core.base.event;
+package fun.asgc.neutrino.core.base;
 
 /**
  * @author: aoshiguchen
- * @date: 2022/9/28
+ * @date: 2022/9/29
  */
-public interface EventMarket {
+public interface Receiver<D> {
 
+    /**
+     * 接收消息
+     * @param msg
+     */
+    void receive(D msg);
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEvent.java

@@ -21,48 +21,30 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
-import java.util.Map;
-import java.util.Set;
-
 /**
  * @author: aoshiguchen
  * @date: 2022/9/28
  */
-public class ApplicationEvent<D> implements Event<ApplicationEventContext, ApplicationEventSource, D> {
+public class ApplicationEvent<D> implements Event<ApplicationEventContext,D> {
 
-    @Override
-    public ApplicationEventSource source() {
-        return null;
+    private D data;
+    private ApplicationEventContext context;
+
+    public ApplicationEvent() {
+        this.context = new ApplicationEventContext();
     }
 
     @Override
     public D data() {
-        return null;
+        return this.data;
     }
 
     @Override
     public ApplicationEventContext context() {
-        return null;
+        return this.context;
     }
 
-    @Override
-    public String topic() {
-        return null;
+    public void setData(D data) {
+        this.data = data;
     }
-
-    @Override
-    public Set<String> tags() {
-        return null;
-    }
-
-    @Override
-    public Map<String, Object> attachData() {
-        return null;
-    }
-
-    @Override
-    public String id() {
-        return null;
-    }
-
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventChannel.java

@@ -0,0 +1,79 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base.event;
+
+import fun.asgc.neutrino.core.base.CustomThreadFactory;
+import fun.asgc.neutrino.core.base.Dispatcher;
+import fun.asgc.neutrino.core.util.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author: aoshiguchen
+ * @date: 2022/9/29
+ */
+public class ApplicationEventChannel<D> implements EventChannel<D,ApplicationEventContext,ApplicationEvent<D>,ApplicationEventReceiver<D>,Dispatcher<ApplicationEventContext,ApplicationEvent<D>>> {
+    private List<ApplicationEventReceiver<D>> receiverList;
+    private Dispatcher<ApplicationEventContext,ApplicationEvent<D>> dispatcher;
+    private ThreadPoolExecutor threadPoolExecutor;
+
+    public ApplicationEventChannel() {
+        this.receiverList = new ArrayList<>();
+        this.threadPoolExecutor = new ThreadPoolExecutor(5, 20, 10L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(), new CustomThreadFactory("ApplicationEventChannel"));
+    }
+
+    @Override
+    public void registerReceiver(ApplicationEventReceiver<D> receiver) {
+        Assert.notNull(receiver, "receiver不能为空!");
+        if (this.receiverList.contains(receiver)) {
+            return;
+        }
+        this.receiverList.add(receiver);
+    }
+
+    @Override
+    public void unRegisterReceiver(ApplicationEventReceiver<D> receiver) {
+        Assert.notNull(receiver, "receiver不能为空!");
+        if (!this.receiverList.contains(receiver)) {
+            return;
+        }
+        this.receiverList.remove(receiver);
+    }
+
+    @Override
+    public void setDispatcher(Dispatcher<ApplicationEventContext, ApplicationEvent<D>> dispatcher) {
+        Assert.notNull(dispatcher, "dispatcher不能为空!");
+        this.dispatcher = dispatcher;
+    }
+
+    @Override
+    public void publish(ApplicationEvent<D> msg) {
+        this.receiverList.forEach(receiver -> {
+            threadPoolExecutor.submit(() -> receiver.receive(msg));
+        });
+    }
+}

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventContext.java

@@ -21,9 +21,71 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
+import fun.asgc.neutrino.core.util.StringUtil;
+
+import java.util.*;
+
 /**
  * @author: aoshiguchen
  * @date: 2022/9/28
  */
 public class ApplicationEventContext implements EventContext {
+
+    private String id;
+    private String topic;
+    private Set<String> tags;
+    private Object source;
+    private Date happenTime;
+    private Map<String, Object> attachData = new HashMap<>();
+
+    public ApplicationEventContext() {
+        this.id = StringUtil.genUUID();
+        this.happenTime = new Date();
+    }
+
+    @Override
+    public Map<String, Object> attachData() {
+        return attachData;
+    }
+
+    @Override
+    public String id() {
+        return id;
+    }
+
+    @Override
+    public Date happenTime() {
+        return happenTime;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    @Override
+    public <S> S source() {
+        return (S)source;
+    }
+
+    @Override
+    public String topic() {
+        return topic;
+    }
+
+    @Override
+    public Set<String> tags() {
+        return tags;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public void setTags(Set<String> tags) {
+        this.tags = tags;
+    }
+
+    public <S> void setSource(S source) {
+        this.source = source;
+    }
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base.event;
+
+
+import fun.asgc.neutrino.core.util.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author: aoshiguchen
+ * @date: 2022/9/29
+ */
+public class ApplicationEventPublisher<D> implements EventPublisher<D,ApplicationEventContext,ApplicationEvent<D>,ApplicationEventChannel<D>> {
+    private List<ApplicationEventChannel<D>> channelList;
+
+    public ApplicationEventPublisher() {
+        this.channelList = new ArrayList<>();
+    }
+
+    @Override
+    public void publish(ApplicationEvent<D> msg) {
+        this.channelList.forEach(channel -> channel.publish(msg));
+    }
+
+    @Override
+    public void registerChannel(ApplicationEventChannel<D> channel) {
+        Assert.notNull(channel, "channel不能为空!");
+        if (!this.channelList.contains(channel)) {
+            this.channelList.add(channel);
+        }
+    }
+
+    @Override
+    public void unRegisterChannel(ApplicationEventChannel<D> channel) {
+        Assert.notNull(channel, "channel不能为空!");
+        this.channelList.remove(channel);
+    }
+}

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java

@@ -0,0 +1,39 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base.event;
+
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author: aoshiguchen
+ * @date: 2022/9/29
+ */
+@Slf4j
+public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>> {
+
+    @Override
+    public void receive(ApplicationEvent<D> msg) {
+        log.debug("ApplicationEventReceiver receive {}", JSONObject.toJSONString(msg));
+    }
+
+}

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/Event.java

@@ -21,20 +21,11 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
-import java.util.Map;
-import java.util.Set;
-
 /**
  * @author: aoshiguchen
  * @date: 2022/9/28
  */
-public interface Event<C extends EventContext, S extends EventSource, D extends Object> {
-    /**
-     * 获取事件源
-     * @return 事件源
-     */
-    S source();
-
+public interface Event<C extends EventContext, D extends Object> {
     /**
      * 获取数据
      * @return 数据
@@ -46,30 +37,4 @@ public interface Event<C extends EventContext, S extends EventSource, D extends
      * @return 上下文
      */
     C context();
-
-    /**
-     * 事件主题
-     * 用于订阅一级过滤
-     * @return 主题
-     */
-    String topic();
-
-    /**
-     * 事件标签
-     * 用于订阅二级过滤
-     * @return 标签
-     */
-    Set<String> tags();
-
-    /**
-     * 附加数据
-     * @return 附加数据
-     */
-    Map<String, Object> attachData();
-
-    /**
-     * 事件ID
-     * @return 事件ID
-     */
-    String id();
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventChannel.java

@@ -21,10 +21,13 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
+import fun.asgc.neutrino.core.base.Channel;
+import fun.asgc.neutrino.core.base.Dispatcher;
+
 /**
  * @author: aoshiguchen
- * @date: 2022/9/28
+ * @date: 2022/9/29
  */
-public interface EventSource {
+public interface EventChannel<D,EC extends EventContext, E extends Event<EC,D>, R extends EventReceiver, Dis extends Dispatcher> extends Channel<E,R,Dis> {
 
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventContext.java

@@ -21,10 +21,48 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
+import java.util.Date;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * @author: wen.y
  * @date: 2022/9/28
  */
 public interface EventContext {
+    /**
+     * 获取事件源
+     * @return 事件源
+     */
+    <T> T source();
+    /**
+     * 事件主题
+     * 用于订阅一级过滤
+     * @return 主题
+     */
+    String topic();
 
+    /**
+     * 事件标签
+     * 用于订阅二级过滤
+     * @return 标签
+     */
+    Set<String> tags();
+    /**
+     * 附加数据
+     * @return 附加数据
+     */
+    Map<String, Object> attachData();
+
+    /**
+     * 事件ID
+     * @return 事件ID
+     */
+    String id();
+
+    /**
+     * 事件发生的时间
+     * @return 事件发生的时间
+     */
+    Date happenTime();
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventPublisher.java

@@ -21,10 +21,12 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
+import fun.asgc.neutrino.core.base.Publisher;
+
 /**
  * @author: aoshiguchen
  * @date: 2022/9/28
  */
-public interface EventPublisher {
+public interface EventPublisher<D,EC extends EventContext,E extends Event<EC,D>,Ch extends EventChannel> extends Publisher<E,Ch> {
 
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/EventReceiver.java

@@ -21,10 +21,13 @@
  */
 package fun.asgc.neutrino.core.base.event;
 
+
+import fun.asgc.neutrino.core.base.Receiver;
+
 /**
  * @author: aoshiguchen
- * @date: 2022/9/28
+ * @date: 2022/9/29
  */
-public class ApplicationEventSource implements EventSource {
+public interface EventReceiver<EC extends EventContext,D,E extends Event<EC,D>> extends Receiver<E> {
 
 }

neutrino-core/src/main/java/fun/asgc/neutrino/core/util/StringUtil.java

@@ -1107,4 +1107,11 @@ public class StringUtil {
 		return arrayToDelimitedString(arr, ",");
 	}
 
+	/**
+	 * 生成UUID
+	 * @return
+	 */
+	public static String genUUID() {
+		return UUID.randomUUID().toString().replace("-", "").toUpperCase();
+	}
 }

neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright (c) 2022 aoshiguchen
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+package fun.asgc.neutrino.core.base.event;
+
+import com.google.common.collect.Sets;
+import lombok.Data;
+import lombok.experimental.Accessors;
+import org.junit.Test;
+
+/**
+ * 应用事件测试
+ * 1、异步执行
+ * 2、业务解耦
+ * 3、topic订阅
+ * 4、支持多种模式无缝切换（本地模式、redis模式、rocketMQ模式、MQTT模式等）
+ * 5、不支持事务消息
+ * @author: aoshiguchen
+ * @date: 2022/10/3
+ */
+public class Test1 {
+    private ApplicationEventChannel<Student> channel = new ApplicationEventChannel<>();
+    private ApplicationEventPublisher<Student> publisher = new ApplicationEventPublisher<>();
+
+    {
+        publisher.registerChannel(channel);
+    }
+
+    @Test
+    public void test1() {
+        ApplicationEventReceiver<Student> receiver = new ApplicationEventReceiver<Student>() {
+            @Override
+            public void receive(ApplicationEvent<Student> msg) {
+                System.out.println("msg:" + msg);
+            }
+        };
+        channel.registerReceiver(receiver);
+
+        ApplicationEvent<Student> event = new ApplicationEvent<>();
+        event.context().setId("123");
+        event.context().setTopic("student");
+        event.context().setTags(Sets.newHashSet("create"));
+        event.setData(new Student().setId("1").setName("张三").setAge(28).setSex("男"));
+        publisher.publish(event);
+    }
+
+    @Accessors(chain = true)
+    @Data
+    public static class Student {
+        private String id;
+        private String name;
+        private Integer age;
+        private String sex;
+    }
+}

neutrino-core/src/test/java/fun/asgc/neutrino/core/util/ClassUtilTest.java

@@ -80,5 +80,4 @@ public class ClassUtilTest {
 		Set<Class<?>> c = ClassUtil.scan("fun.asgc.neutrino.proxy.server", url);
 		System.out.println(c);
 	}
-
 }