yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

新增应用事件管理器封装、测试，客户端接入应用生命周期事件监听

Browse Source
This commit is contained in:
aoshiguchen
2022-10-10 18:48:44 +08:00
parent b15d1f1b91
commit aa1c08fd54
15 changed files with 322 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-core/src/main/java/fun/asgc/neutrino/core/annotation/Subscribe.java
+18
View File
@@ -0,0 +1,18 @@
package fun.asgc.neutrino.core.annotation;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* @author: aoshiguchen
* @date: 2022/10/10
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface Subscribe {
boolean enable() default true;
String topic() default "";
String[] tags() default {};
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/Publisher.java
+4 -4
View File
@@ -33,14 +33,14 @@ public interface Publisher<D,Ch extends Channel> {
void publish(D msg);
/**
* 注册渠道
* 绑定渠道
* @param channel 渠道
*/
void registerChannel(Ch channel);
void bindChannel(Ch channel);
/**
* 注销渠道
* 解绑渠道
* @param channel 渠道
*/
void unRegisterChannel(Ch channel);
void unbindChannel(Ch channel);
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventPublisher.java
+2 -2
View File
@@ -44,7 +44,7 @@ public class ApplicationEventPublisher<D> implements EventPublisher<D,Applicatio
}
@Override
public void registerChannel(ApplicationEventChannel<D> channel) {
public void bindChannel(ApplicationEventChannel<D> channel) {
Assert.notNull(channel, "channel不能为空!");
if (!this.channelList.contains(channel)) {
this.channelList.add(channel);
@@ -52,7 +52,7 @@ public class ApplicationEventPublisher<D> implements EventPublisher<D,Applicatio
}
@Override
public void unRegisterChannel(ApplicationEventChannel<D> channel) {
public void unbindChannel(ApplicationEventChannel<D> channel) {
Assert.notNull(channel, "channel不能为空!");
this.channelList.remove(channel);
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/base/event/ApplicationEventReceiver.java
+14
View File
@@ -34,6 +34,20 @@ import java.util.Set;
public class ApplicationEventReceiver<D> implements EventReceiver<ApplicationEventContext,D,ApplicationEvent<D>> {
private String topic;
private Set<String> tags;
public ApplicationEventReceiver() {
}
public ApplicationEventReceiver(String topic) {
this.topic = topic;
}
public ApplicationEventReceiver(String topic, Set<String> tags) {
this.topic = topic;
this.tags = tags;
}
@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/SimpleApplicationEventManager.java
+81
View File
@@ -0,0 +1,81 @@
/**
* 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 java.util.Set;
/**
* 简单的应用事件管理器
* @author: wen.u
* @date: 2022/10/10
*/
public class SimpleApplicationEventManager<D> {
private ApplicationEventChannel<D> channel;
private ApplicationEventPublisher<D> publisher;
private Object source;
public SimpleApplicationEventManager() {
this.channel = new ApplicationEventChannel<>();
this.publisher = new ApplicationEventPublisher<>();
this.publisher.bindChannel(channel);
}
public SimpleApplicationEventManager(Object source) {
this.channel = new ApplicationEventChannel<>();
this.publisher = new ApplicationEventPublisher<>();
this.publisher.bindChannel(channel);
this.source = source;
}
public void publish(D data) {
this.publish(null, null, data);
}
public void publish(String topic, D data) {
this.publish(topic, null, data);
}
public void publish(String topic, Set<String> tags, D data) {
ApplicationEvent<D> event = new ApplicationEvent<>();
event.setData(data);
event.context().setSource(source);
event.context().setTopic(topic);
event.context().setTags(tags);
this.publisher.publish(event);
}
public void registerReceiver(ApplicationEventReceiver<D> receiver) {
this.channel.registerReceiver(receiver);
}
public ApplicationEventChannel<D> getChannel() {
return channel;
}
public ApplicationEventPublisher<D> getPublisher() {
return publisher;
}
public Object getSource() {
return source;
}
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/BeanWrapper.java
+4 -1
View File
@@ -25,7 +25,10 @@ package fun.asgc.neutrino.core.bean;
import fun.asgc.neutrino.core.annotation.*;
import fun.asgc.neutrino.core.context.ApplicationRunner;
import fun.asgc.neutrino.core.context.LifeCycle;
import fun.asgc.neutrino.core.util.*;
import fun.asgc.neutrino.core.util.ClassUtil;
import fun.asgc.neutrino.core.util.CollectionUtil;
import fun.asgc.neutrino.core.util.LockUtil;
import fun.asgc.neutrino.core.util.ReflectUtil;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/AbstractBeanFactory.java
+35 -6
View File
@@ -21,7 +21,10 @@
*/
package fun.asgc.neutrino.core.bean.factory;
import fun.asgc.neutrino.core.annotation.Subscribe;
import fun.asgc.neutrino.core.base.CustomThreadFactory;
import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver;
import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager;
import fun.asgc.neutrino.core.bean.*;
import fun.asgc.neutrino.core.context.Environment;
import fun.asgc.neutrino.core.context.LifeCycle;
@@ -31,12 +34,13 @@ import fun.asgc.neutrino.core.exception.BeanException;
import fun.asgc.neutrino.core.util.*;
import lombok.extern.slf4j.Slf4j;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 抽象的bean工厂
@@ -371,7 +375,32 @@ public abstract class AbstractBeanFactory implements BeanFactory, BeanRegistry,
parent.init();
}
if (CollectionUtil.notEmpty(beanCache)) {
beanCache.values().stream().filter(b -> BeanStatus.INJECT == b.getStatus()).forEach(bean -> bean.init());
beanCache.values().stream().forEach(bean -> {
SimpleApplicationEventManager defaultApplicationEventManager = getEnvironment().getDefaultApplicationEventManager();
if (null != defaultApplicationEventManager && ApplicationEventReceiver.class.isAssignableFrom(bean.getType())) {
boolean enable = true;
String topic = null;
Set<String> tags = null;
Subscribe subscribe = bean.getType().getAnnotation(Subscribe.class);
if (null != subscribe) {
enable = subscribe.enable();
topic = subscribe.topic();
if (ArrayUtil.notEmpty(subscribe.tags())) {
tags = Stream.of(subscribe.tags()).collect(Collectors.toSet());
}
}
if (enable) {
ApplicationEventReceiver receiver = (ApplicationEventReceiver) bean.getInstance();
receiver.setTopic(topic);
receiver.setTags(tags);
defaultApplicationEventManager.registerReceiver(receiver);
}
}
if (BeanStatus.INJECT == bean.getStatus()) {
bean.init();
}
});
}
log.info("bean工厂[{}]初始化.", getName());
scheduledExecutor.scheduleWithFixedDelay(this::run, 0, 1, TimeUnit.SECONDS);
neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/factory/SimpleBeanFactory.java
+2
View File
@@ -28,6 +28,7 @@ import fun.asgc.neutrino.core.aop.interceptor.ExceptionHandler;
import fun.asgc.neutrino.core.aop.interceptor.Filter;
import fun.asgc.neutrino.core.aop.interceptor.Interceptor;
import fun.asgc.neutrino.core.aop.interceptor.ResultAdvice;
import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver;
import fun.asgc.neutrino.core.bean.*;
import fun.asgc.neutrino.core.exception.BeanException;
import fun.asgc.neutrino.core.context.ApplicationRunner;
@@ -374,6 +375,7 @@ public class SimpleBeanFactory extends AbstractBeanFactory {
|| Filter.class.isAssignableFrom(item)
|| ExceptionHandler.class.isAssignableFrom(item)
|| ResultAdvice.class.isAssignableFrom(item)
|| ApplicationEventReceiver.class.isAssignableFrom(item)
)
.forEach(clazz -> {
String beanName = TypeUtil.getDefaultVariableName(clazz);
neutrino-core/src/main/java/fun/asgc/neutrino/core/constant/MetaDataConstant.java
+4
View File
@@ -80,4 +80,8 @@ public interface MetaDataConstant {
* 服务版本
*/
String SERVER_VS = "Neutrino-1.0";
/**
* app生命周期主题
*/
String TOPIC_APP_LIFE_CYCLE = "TP_APP_LIFE_CYCLE";
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/context/Environment.java
+5
View File
@@ -22,6 +22,7 @@
package fun.asgc.neutrino.core.context;
import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager;
import fun.asgc.neutrino.core.util.SystemUtil;
import lombok.Data;
import lombok.experimental.Accessors;
@@ -64,4 +65,8 @@ public class Environment {
* 运行上下文
*/
private SystemUtil.RunContext runContext;
/**
* 默认的应用事件管理器
*/
private SimpleApplicationEventManager defaultApplicationEventManager;
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/context/NeutrinoLauncher.java
+30 -2
View File
@@ -25,6 +25,8 @@ package fun.asgc.neutrino.core.context;
import com.google.common.collect.Lists;
import fun.asgc.neutrino.core.annotation.EnableJob;
import fun.asgc.neutrino.core.annotation.NeutrinoApplication;
import fun.asgc.neutrino.core.base.event.SimpleApplicationEventManager;
import fun.asgc.neutrino.core.constant.AppLifeCycleStatusEnum;
import fun.asgc.neutrino.core.constant.MetaDataConstant;
import fun.asgc.neutrino.core.util.*;
import lombok.extern.slf4j.Slf4j;
@@ -52,26 +54,44 @@ public class NeutrinoLauncher {
private NeutrinoLauncher(Class<?> clazz, String[] args) {
this.environment = new Environment()
.setMainClass(clazz)
.setMainArgs(args);
.setMainArgs(args)
.setDefaultApplicationEventManager(new SimpleApplicationEventManager(this))
;
}
private SystemUtil.RunContext launch() {
// 已创建
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_CREATE);
StopWatch stopWatch = new StopWatch();
stopWatch.start();
environmentInit();
// 应用已初始化
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_INIT);
ApplicationContext context = new ApplicationContext(environment);
SystemUtil.RunContext runContext = SystemUtil.waitProcessDestroy(() -> {
// 应用准备销毁
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_PRE_DESTROY);
context.destroy();
log.info("Application already stop.");
// 应用已销毁
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_DESTROY);
});
environment.setRunContext(runContext);
context.run();
// 容器已初始化
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.CONTAINER_INIT);
stopWatch.stop();
printLog(environment, stopWatch);
// 应用已启动完成
publishAppLifeCycleEvent(AppLifeCycleStatusEnum.APP_STARTUP);
return runContext;
}
@@ -133,4 +153,12 @@ public class NeutrinoLauncher {
}
log.info(environment.getBanner());
}
/**
* 发布应用生命周期事件
* @param appLifeCycleStatusEnum 应用生命周期事件
*/
private void publishAppLifeCycleEvent(AppLifeCycleStatusEnum appLifeCycleStatusEnum) {
this.environment.getDefaultApplicationEventManager().publish(MetaDataConstant.TOPIC_APP_LIFE_CYCLE, appLifeCycleStatusEnum);
}
}
neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test1.java
+1 -1
View File
@@ -45,7 +45,7 @@ public class Test1 {
private ApplicationEventPublisher<Student> publisher = new ApplicationEventPublisher<>();
{
publisher.registerChannel(channel);
publisher.bindChannel(channel);
}
@Test
neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test2.java
+1 -1
View File
@@ -47,7 +47,7 @@ public class Test2 {
{
channel.connectChannel(myTestChannel);
publisher.registerChannel(channel);
publisher.bindChannel(channel);
}
@Test
neutrino-core/src/test/java/fun/asgc/neutrino/core/base/event/Test3.java
+80
View File
@@ -0,0 +1,80 @@
/**
* 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 fun.asgc.neutrino.core.util.SystemUtil;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
/**
* 应用事件测试
* 1、异步执行
* 2、业务解耦
* 3、topic订阅
* 4、支持多种模式无缝切换（本地模式、redis模式、rocketMQ模式、MQTT模式等）
* 5、不支持事务消息
* @author: aoshiguchen
* @date: 2022/10/3
*/
@Slf4j
public class Test3 {
private SimpleApplicationEventManager<Student> simpleApplicationEventManager = new SimpleApplicationEventManager<>(this);
@Test
public void test1() {
ApplicationEventReceiver<Student> receiver1 = new ApplicationEventReceiver<Student>() {
@Override
public void receive(ApplicationEvent<Student> msg) {
log.info("receiver1 data:{}", JSONObject.toJSONString(msg.data()));
}
};
receiver1.setTopic("/*");
ApplicationEventReceiver<Student> receiver2 = new ApplicationEventReceiver<Student>() {
@Override
public void receive(ApplicationEvent<Student> msg) {
log.info("receiver2 data:{}", JSONObject.toJSONString(msg.data()));
}
};
simpleApplicationEventManager.registerReceiver(receiver1);
simpleApplicationEventManager.registerReceiver(receiver2);
simpleApplicationEventManager.publish("/aaa", null, new Student().setId("1").setName("张三").setAge(28).setSex("男"));
SystemUtil.waitProcessDestroy().sync();
}
@Accessors(chain = true)
@Data
public static class Student {
private String id;
private String name;
private Integer age;
private String sex;
}
public static class MyTestChannel extends ApplicationEventChannel<Student> {
}
}
neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/ApplicationLifeCycleListener.java
+41
View File
@@ -0,0 +1,41 @@
/**
* 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.proxy.client.core;
import fun.asgc.neutrino.core.base.event.ApplicationEvent;
import fun.asgc.neutrino.core.base.event.ApplicationEventReceiver;
import fun.asgc.neutrino.core.constant.AppLifeCycleStatusEnum;
import lombok.extern.slf4j.Slf4j;
/**
* @author: aoshiguchen
* @date: 2022/10/10
*/
@Slf4j
public class ApplicationLifeCycleListener extends ApplicationEventReceiver<AppLifeCycleStatusEnum> {
@Override
public void receive(ApplicationEvent<AppLifeCycleStatusEnum> msg) {
log.info("ApplicationLifeCycleListener:{}", msg.data().getDesc());
}
}