event

jaychuo · Jul 21, 2022 · 9c5da58 · 9c5da58
1 parent 6b3118f
commit 9c5da58
Show file tree

Hide file tree

Showing 6 changed files with 104 additions and 14 deletions.
diff --git a/...-event-core/src/main/java/com/github/linyuzai/event/core/concept/DefaultEventConcept.java b/...-event-core/src/main/java/com/github/linyuzai/event/core/concept/DefaultEventConcept.java
@@ -6,6 +6,7 @@
 import com.github.linyuzai.event.core.context.EventContextFactory;
 import com.github.linyuzai.event.core.endpoint.EventEndpoint;
 import com.github.linyuzai.event.core.error.EventErrorHandler;
+import com.github.linyuzai.event.core.lifecycle.EventConceptLifecycleListener;
 import com.github.linyuzai.event.core.publisher.EventPublisher;
 import com.github.linyuzai.event.core.engine.EventEngine;
 import com.github.linyuzai.event.core.exchange.EventExchange;
@@ -18,15 +19,19 @@
 import java.lang.reflect.Type;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 @Getter
 @Setter
 public class DefaultEventConcept implements EventConcept {
 
     protected final Map<String, EventEngine> engineMap = new ConcurrentHashMap<>();
 
+    protected final List<EventConceptLifecycleListener> lifecycleListeners = new CopyOnWriteArrayList<>();
+
     private EventContextFactory contextFactory;
 
     private EventExchange exchange;
@@ -37,6 +42,20 @@ public class DefaultEventConcept implements EventConcept {
 
     private EventErrorHandler errorHandler;
 
+    @Override
+    public void initialize() {
+        for (EventConceptLifecycleListener lifecycleListener : lifecycleListeners) {
+            lifecycleListener.onInitialize(this);
+        }
+    }
+
+    @Override
+    public void destroy() {
+        for (EventConceptLifecycleListener lifecycleListener : lifecycleListeners) {
+            lifecycleListener.onDestroy(this);
+        }
+    }
+
     @Override
     public EventOperator event() {
         return new EventOperatorImpl();
@@ -99,12 +118,29 @@ public Collection<EventEngine> getEngines() {
     }
 
     @Override
-    public void add(Collection<? extends EventEngine> engines) {
+    public void addEngines(Collection<? extends EventEngine> engines) {
         for (EventEngine engine : engines) {
             this.engineMap.put(engine.getName(), engine);
         }
     }
 
+    @Override
+    public void removeEngines(Collection<String> engines) {
+        for (String engine : engines) {
+            this.engineMap.remove(engine);
+        }
+    }
+
+    @Override
+    public void addLifecycleListeners(Collection<? extends EventConceptLifecycleListener> lifecycleListeners) {
+        this.lifecycleListeners.addAll(lifecycleListeners);
+    }
+
+    @Override
+    public void removeLifecycleListeners(Collection<? extends EventConceptLifecycleListener> lifecycleListeners) {
+        this.lifecycleListeners.removeAll(lifecycleListeners);
+    }
+
     protected EventExchange useExchange(EventExchange exchange) {
         if (exchange != null) {
             return exchange;

diff --git a/...concept-event-core/src/main/java/com/github/linyuzai/event/core/concept/EventConcept.java b/...concept-event-core/src/main/java/com/github/linyuzai/event/core/concept/EventConcept.java
@@ -5,13 +5,18 @@
 import com.github.linyuzai.event.core.error.EventErrorHandler;
 import com.github.linyuzai.event.core.engine.EventEngine;
 import com.github.linyuzai.event.core.exchange.EventExchange;
+import com.github.linyuzai.event.core.lifecycle.EventConceptLifecycleListener;
 
 import java.lang.reflect.Type;
 import java.util.Arrays;
 import java.util.Collection;
 
 public interface EventConcept {
 
+    void initialize();
+
+    void destroy();
+
     EventOperator event();
 
     EventOperator event(Type type);
@@ -36,9 +41,29 @@ public interface EventConcept {
 
     Collection<EventEngine> getEngines();
 
-    default void add(EventEngine... engines) {
-        add(Arrays.asList(engines));
+    default void addEngines(EventEngine... engines) {
+        addEngines(Arrays.asList(engines));
+    }
+
+    void addEngines(Collection<? extends EventEngine> engines);
+
+    default void removeEngines(String... engines) {
+        removeEngines(Arrays.asList(engines));
+    }
+
+    void removeEngines(Collection<String> engines);
+
+    Collection<EventConceptLifecycleListener> getLifecycleListeners();
+
+    default void addLifecycleListeners(EventConceptLifecycleListener... lifecycleListeners) {
+        addLifecycleListeners(Arrays.asList(lifecycleListeners));
+    }
+
+    void addLifecycleListeners(Collection<? extends EventConceptLifecycleListener> lifecycleListeners);
+
+    default void removeLifecycleListeners(EventConceptLifecycleListener... lifecycleListeners) {
+        removeLifecycleListeners(Arrays.asList(lifecycleListeners));
     }
 
-    void add(Collection<? extends EventEngine> engines);
+    void removeLifecycleListeners(Collection<? extends EventConceptLifecycleListener> lifecycleListeners);
 }
diff --git a/...src/main/java/com/github/linyuzai/event/core/lifecycle/EventConceptLifecycleListener.java b/...src/main/java/com/github/linyuzai/event/core/lifecycle/EventConceptLifecycleListener.java
@@ -0,0 +1,10 @@
+package com.github.linyuzai.event.core.lifecycle;
+
+import com.github.linyuzai.event.core.concept.EventConcept;
+
+public interface EventConceptLifecycleListener {
+
+    void onInitialize(EventConcept concept);
+
+    void onDestroy(EventConcept concept);
+}
diff --git a/...ent-kafka/src/main/java/com/github/linyuzai/event/kafka/AbstractKafkaEventSubscriber.java b/...ent-kafka/src/main/java/com/github/linyuzai/event/kafka/AbstractKafkaEventSubscriber.java
@@ -1,6 +1,9 @@
 package com.github.linyuzai.event.kafka;
 
+import com.github.linyuzai.event.core.concept.EventConcept;
 import com.github.linyuzai.event.core.context.EventContext;
+import com.github.linyuzai.event.core.lifecycle.EventConceptLifecycleListener;
+import lombok.AllArgsConstructor;
 import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.listener.MessageListenerContainer;
 
@@ -13,9 +16,29 @@ public void subscribe(Type type, KafkaEventEndpoint endpoint, EventContext conte
         MessageListenerContainer container = createContainer(type, endpoint, context);
         container.getContainerProperties().setMessageListener(createMessageListener(type, endpoint, context));
         container.start();
+        EventConcept concept = context.get(EventConcept.class);
+        concept.addLifecycleListeners(new MessageListenerContainerStopper(container));
     }
 
     public abstract MessageListenerContainer createContainer(Type type, KafkaEventEndpoint endpoint, EventContext context);
 
     public abstract MessageListener<?, ?> createMessageListener(Type type, KafkaEventEndpoint endpoint, EventContext context);
+
+    @AllArgsConstructor
+    public static class MessageListenerContainerStopper implements EventConceptLifecycleListener {
+
+        private MessageListenerContainer container;
+
+        @Override
+        public void onInitialize(EventConcept concept) {
+
+        }
+
+        @Override
+        public void onDestroy(EventConcept concept) {
+            if (container.isRunning()) {
+                container.stop();
+            }
+        }
+    }
 }
diff --git a/...vent-kafka/src/main/java/com/github/linyuzai/event/kafka/KafkaEventAutoConfiguration.java b/...vent-kafka/src/main/java/com/github/linyuzai/event/kafka/KafkaEventAutoConfiguration.java
@@ -69,7 +69,6 @@ public ProducerListener<Object, Object> kafkaProducerListener() {
         return new DefaultKafkaConsumerFactory<>(Collections.emptyMap());
     }
 
-
     @Bean
     public KafkaAdmin kafkaAdmin() {
         return new KafkaAdmin(Collections.emptyMap());
@@ -145,12 +144,6 @@ public KafkaEventEngine kafkaEventEngine(ConfigurableBeanFactory beanFactory,
                 beanFactory.registerSingleton(key + "KafkaAdmin", endpoint.getAdmin());
             }
 
-            //registry.registerBeanDefinition();
-
-            //ApplicationContext context;
-
-            //context.getBean()
-
             engine.add(endpoint);
             beanFactory.registerSingleton(key + "KafkaEventEndpoint", endpoint);
 

diff --git a/...oot-starter/src/main/java/com/github/linyuzai/event/autoconfigure/EventConfiguration.java b/...oot-starter/src/main/java/com/github/linyuzai/event/autoconfigure/EventConfiguration.java
@@ -10,6 +10,7 @@
 import com.github.linyuzai.event.core.error.EventErrorHandler;
 import com.github.linyuzai.event.core.error.LoggerEventErrorHandler;
 import com.github.linyuzai.event.core.exchange.EventExchange;
+import com.github.linyuzai.event.core.lifecycle.EventConceptLifecycleListener;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.factory.ObjectProvider;
@@ -35,21 +36,23 @@ public EventErrorHandler eventErrorHandler() {
         return new LoggerEventErrorHandler(log::error);
     }
 
-    @Bean
+    @Bean(initMethod = "initialize", destroyMethod = "destroy")
     @ConditionalOnMissingBean
     public EventConcept eventConcept(EventContextFactory contextFactory,
                                      ObjectProvider<EventExchange> exchangeProvider,
                                      ObjectProvider<EventEncoder> encoderProvider,
                                      ObjectProvider<EventDecoder> decoderProvider,
                                      EventErrorHandler errorHandler,
-                                     List<EventEngine> engines) {
+                                     List<EventEngine> engines,
+                                     List<EventConceptLifecycleListener> lifecycleListeners) {
         DefaultEventConcept concept = new DefaultEventConcept();
         concept.setContextFactory(contextFactory);
         concept.setExchange(exchangeProvider.getIfUnique());
         concept.setEncoder(encoderProvider.getIfUnique());
         concept.setDecoder(decoderProvider.getIfUnique());
         concept.setErrorHandler(errorHandler);
-        concept.add(engines);
+        concept.addEngines(engines);
+        concept.addLifecycleListeners(lifecycleListeners);
         return concept;
     }
 }