ApplicationEvent支持根据topic、tag进行订阅

2022-10-04 20:09:03 +08:00
parent e02ed1db8d
commit bd17313db8
6 changed files with 132 additions and 6 deletions
@@ -0,0 +1,35 @@
+/**
+ * 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/10/4
+ */
+public interface TagMatcher {
+    /**
+     * tag匹配
+     * @param tag 标签
+     * @return 是否匹配
+     */
+    boolean tagMatch(String tag);
+}
@@ -0,0 +1,35 @@
+/**
+ * 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/10/4
+ */
+public interface TopicMatcher {
+    /**
+     * topic匹配
+     * @param topic topic
+     * @return 是否匹配
+     */
+    boolean topicMatch(String topic);
+}
@@ -21,12 +21,15 @@
 */
 package fun.asgc.neutrino.core.base.event;

+import com.google.common.collect.Sets;
 import fun.asgc.neutrino.core.base.CustomThreadFactory;
 import fun.asgc.neutrino.core.base.Dispatcher;
 import fun.asgc.neutrino.core.util.Assert;
+import fun.asgc.neutrino.core.util.CollectionUtil;

 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -73,7 +76,21 @@ public class ApplicationEventChannel<D> implements EventChannel<D,ApplicationEve
    @Override
    public void publish(ApplicationEvent<D> msg) {
        this.receiverList.forEach(receiver -> {
-            threadPoolExecutor.submit(() -> receiver.receive(msg));
+            threadPoolExecutor.submit(() -> {
+                if (!receiver.topicMatch(msg.context().topic())) {
+                    return;
+                }
+                Set<String> tags = msg.context().tags();
+                if (CollectionUtil.isEmpty(tags)) {
+                    tags = Sets.newHashSet("");
+                }
+                for (String tag : tags) {
+                    if (receiver.tagMatch(tag)) {
+                        receiver.receive(msg);
+                        break;
+                    }
+                }
+            });
        });
    }
 }
@@ -22,18 +22,51 @@
 package fun.asgc.neutrino.core.base.event;

 import com.alibaba.fastjson.JSONObject;
+import fun.asgc.neutrino.core.base.TagMatcher;
+import fun.asgc.neutrino.core.base.TopicMatcher;
+import fun.asgc.neutrino.core.util.CollectionUtil;
+import fun.asgc.neutrino.core.util.StringUtil;
+import fun.asgc.neutrino.core.web.AntPathMatcher;
 import lombok.extern.slf4j.Slf4j;

+import java.util.Set;
+
 /**
 * @author: aoshiguchen
 * @date: 2022/9/29
 */
@Slf4j
-public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>> {
+public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>>,TagMatcher,TopicMatcher {
+    private String topic;
+    private Set<String> tags;
+    private static final AntPathMatcher antPathMatcher = new AntPathMatcher();
+
+    @Override
+    public boolean tagMatch(String tag) {
+        if (CollectionUtil.isEmpty(this.tags)) {
+            return true;
+        }
+        return this.tags.contains(tag);
+    }
+
+    @Override
+    public boolean topicMatch(String topic) {
+        if (StringUtil.isEmpty(this.topic)) {
+            return true;
+        }
+        return antPathMatcher.match(this.topic, topic == null ? "" : topic);
+    }

    @Override
    public void receive(ApplicationEvent<D> msg) {
        log.debug("ApplicationEventReceiver receive {}", JSONObject.toJSONString(msg));
    }

+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public void setTags(Set<String> tags) {
+        this.tags = tags;
+    }
 }
@@ -21,9 +21,12 @@
 */
 package fun.asgc.neutrino.core.base.event;

+import com.alibaba.fastjson.JSONObject;
 import com.google.common.collect.Sets;
+import fun.asgc.neutrino.core.util.SystemUtil;
 import lombok.Data;
 import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
 import org.junit.Test;

 /**
@@ -36,6 +39,7 @@ import org.junit.Test;
 * @author: aoshiguchen
 * @date: 2022/10/3
 */
+@Slf4j
 public class Test1 {
    private ApplicationEventChannel<Student> channel = new ApplicationEventChannel<>();
    private ApplicationEventPublisher<Student> publisher = new ApplicationEventPublisher<>();
@@ -49,17 +53,21 @@ public class Test1 {
        ApplicationEventReceiver<Student> receiver = new ApplicationEventReceiver<Student>() {
            @Override
            public void receive(ApplicationEvent<Student> msg) {
-                System.out.println("msg:" + msg);
+                log.info("data:{}", JSONObject.toJSONString(msg.data()));
            }
        };
+//        receiver.setTopic("/**");
+//        receiver.setTags(Sets.newHashSet("create"));
        channel.registerReceiver(receiver);

        ApplicationEvent<Student> event = new ApplicationEvent<>();
        event.context().setId("123");
-        event.context().setTopic("student");
+        event.context().setTopic("/student/create");
        event.context().setTags(Sets.newHashSet("create"));
        event.setData(new Student().setId("1").setName("张三").setAge(28).setSex("男"));
        publisher.publish(event);
+
+        SystemUtil.waitProcessDestroy().sync();
    }

    @Accessors(chain = true)
@@ -49,8 +49,6 @@ public class InterceptorRegistryTest {
 		System.out.println(antPathMatcher.match("**/**.html", "/a/11/22/33/a.html"));
 		System.out.println(antPathMatcher.match("/**/*.html", "/a/11/22/33/a.html"));
 		System.out.println(antPathMatcher.match("/**/*.html", "/a.html"));
-
-
 	}

 }