JAV-546 register an anonymous subscriber to listen to instance registry event and trigger AFTER_REGISTRY event

core/src/main/java/io/servicecomb/core/CseApplicationListener.java

@@ -31,6 +31,9 @@
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.core.Ordered;
+import org.springframework.util.StringUtils;
+
+import com.google.common.eventbus.Subscribe;
 
 import io.servicecomb.core.BootListener.BootEvent;
 import io.servicecomb.core.BootListener.EventType;
@@ -41,9 +44,11 @@
 import io.servicecomb.core.provider.consumer.ReferenceConfigUtils;
 import io.servicecomb.core.provider.producer.ProducerProviderManager;
 import io.servicecomb.core.transport.TransportManager;
+import io.servicecomb.foundation.common.event.EventManager;
 import io.servicecomb.foundation.common.utils.BeanUtils;
 import io.servicecomb.foundation.common.utils.FortifyUtils;
 import io.servicecomb.serviceregistry.RegistryUtils;
+import io.servicecomb.serviceregistry.task.MicroserviceInstanceRegisterTask;
 
 public class CseApplicationListener
     implements ApplicationListener<ApplicationEvent>, Ordered, ApplicationContextAware {
@@ -124,9 +129,10 @@ public void onApplicationEvent(ApplicationEvent event) {
           schemaListenerManager.notifySchemaListener();
 
           triggerEvent(EventType.BEFORE_REGISTRY);
+
+          triggerAfterRegistryEvent();
+
           RegistryUtils.run();
-          ReferenceConfigUtils.setReady(true);
-          triggerEvent(EventType.AFTER_REGISTRY);
 
           // 当程序退出时，进行相关清理，注意：kill -9 {pid}下无效
           // 1. 去注册实例信息
@@ -145,4 +151,30 @@ public void onApplicationEvent(ApplicationEvent event) {
       isInit = false;
     }
   }
+
+  /**
+   * <p>As the process of instance registry is asynchronous, the {@code AFTER_REGISTRY}
+   * event should not be sent immediately after {@link RegistryUtils#run()} is invoked.
+   * When the instance registry succeeds, {@link MicroserviceInstanceRegisterTask} will be posted in {@link EventManager},
+   * register a subscriber to watch this event and send {@code AFTER_REGISTRY}.</p>
+   *
+   * <p>This method should be called before {@link RegistryUtils#run()} to avoid that the registry process is too quick
+   * that the event is not watched by this subscriber.</p>
+   *
+   * <p>Check if {@code InstanceId} is null to judge whether the instance registry has succeeded.</p>
+   */
+  private void triggerAfterRegistryEvent() {
+    EventManager.register(new Object() {
+      @Subscribe
+      public void afterRegistryInstance(MicroserviceInstanceRegisterTask microserviceInstanceRegisterTask) {
+        LOGGER.info("receive MicroserviceInstanceRegisterTask event, check instance Id...");
+        if (!StringUtils.isEmpty(RegistryUtils.getMicroserviceInstance().getInstanceId())) {
+          LOGGER.info("instance registry succeeds for the first time, will send AFTER_REGISTRY event.");
+          ReferenceConfigUtils.setReady(true);
+          triggerEvent(EventType.AFTER_REGISTRY);
+          EventManager.unregister(this);
+        }
+      }
+    });
+  }
 }

core/src/test/java/io/servicecomb/core/TestCseApplicationListener.java

@@ -16,21 +16,37 @@
 
 package io.servicecomb.core;
 
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.support.AbstractApplicationContext;
 
+import io.servicecomb.core.BootListener.BootEvent;
+import io.servicecomb.core.definition.loader.SchemaListenerManager;
 import io.servicecomb.core.endpoint.AbstractEndpointsCache;
 import io.servicecomb.core.provider.consumer.ConsumerProviderManager;
+import io.servicecomb.core.provider.consumer.ReferenceConfigUtils;
 import io.servicecomb.core.provider.producer.ProducerProviderManager;
 import io.servicecomb.core.transport.TransportManager;
+import io.servicecomb.foundation.common.event.EventManager;
 import io.servicecomb.foundation.common.utils.ReflectUtils;
 import io.servicecomb.serviceregistry.RegistryUtils;
+import io.servicecomb.serviceregistry.api.registry.MicroserviceInstance;
+import io.servicecomb.serviceregistry.task.MicroserviceInstanceRegisterTask;
+import mockit.Deencapsulation;
 import mockit.Expectations;
 import mockit.Injectable;
 import mockit.Mocked;
@@ -41,49 +57,66 @@ public static void teardown() {
     AbstractEndpointsCache.init(null, null);
   }
 
+  @After
+  public void cleanup() {
+    Deencapsulation.setField(ReferenceConfigUtils.class, "ready", false);
+  }
+
   @Test
   public void testCseApplicationListenerNormal(@Injectable ContextRefreshedEvent event,
       @Injectable AbstractApplicationContext context,
-      @Injectable BootListener listener,
       @Injectable ProducerProviderManager producerProviderManager,
       @Injectable ConsumerProviderManager consumerProviderManager,
       @Injectable TransportManager transportManager,
-      @Mocked RegistryUtils ru) throws Exception {
+      @Mocked RegistryUtils ru) {
     Map<String, BootListener> listeners = new HashMap<>();
+    BootListener listener = Mockito.mock(BootListener.class);
     listeners.put("test", listener);
 
+    SchemaListenerManager schemaListenerManager = Mockito.mock(SchemaListenerManager.class);
+    MicroserviceInstance microserviceInstance = Mockito.mock(MicroserviceInstance.class);
+
     new Expectations() {
       {
         context.getBeansOfType(BootListener.class);
         result = listeners;
       }
     };
+    new Expectations(RegistryUtils.class) {
+      {
+        RegistryUtils.init();
+        RegistryUtils.getInstanceCacheManager();
+        RegistryUtils.run();
+        RegistryUtils.getMicroserviceInstance();
+        result = microserviceInstance;
+      }
+    };
+    Mockito.when(microserviceInstance.getInstanceId()).thenReturn("testInstanceId");
 
     CseApplicationListener cal = new CseApplicationListener();
+    Deencapsulation.setField(cal, "schemaListenerManager", schemaListenerManager);
     cal.setInitEventClass(ContextRefreshedEvent.class);
     cal.setApplicationContext(context);
     ReflectUtils.setField(cal, "producerProviderManager", producerProviderManager);
     ReflectUtils.setField(cal, "consumerProviderManager", consumerProviderManager);
     ReflectUtils.setField(cal, "transportManager", transportManager);
 
     cal.onApplicationEvent(event);
+
+    EventManager.post(Mockito.mock(MicroserviceInstanceRegisterTask.class));
+    verify(schemaListenerManager).notifySchemaListener();
+    verify(listener, times(10)).onBootEvent(Mockito.any(BootEvent.class));
   }
 
   @Test
   public void testCseApplicationListenerThrowException(@Injectable ContextRefreshedEvent event,
       @Injectable AbstractApplicationContext context,
       @Injectable BootListener listener,
       @Injectable ProducerProviderManager producerProviderManager,
-      @Mocked RegistryUtils ru) throws Exception {
+      @Mocked RegistryUtils ru) {
     Map<String, BootListener> listeners = new HashMap<>();
     listeners.put("test", listener);
 
-    new Expectations() {
-      {
-        context.getBeansOfType(BootListener.class);
-        result = listeners;
-      }
-    };
     CseApplicationListener cal = new CseApplicationListener();
     cal.setApplicationContext(context);
     ReflectUtils.setField(cal, "producerProviderManager", producerProviderManager);
@@ -94,7 +127,7 @@ public void testCseApplicationListenerThrowException(@Injectable ContextRefreshe
   public void testCseApplicationListenerParentNotnull(@Injectable ContextRefreshedEvent event,
       @Injectable AbstractApplicationContext context,
       @Injectable AbstractApplicationContext pContext,
-      @Mocked RegistryUtils ru) throws Exception {
+      @Mocked RegistryUtils ru) {
 
     CseApplicationListener cal = new CseApplicationListener();
     cal.setApplicationContext(context);
@@ -103,7 +136,7 @@ public void testCseApplicationListenerParentNotnull(@Injectable ContextRefreshed
 
   @Test
   public void testCseApplicationListenerShutdown(@Injectable ContextClosedEvent event,
-      @Mocked RegistryUtils ru) throws Exception {
+      @Mocked RegistryUtils ru) {
     new Expectations() {
       {
         RegistryUtils.destory();
@@ -112,4 +145,69 @@ public void testCseApplicationListenerShutdown(@Injectable ContextClosedEvent ev
     CseApplicationListener cal = new CseApplicationListener();
     cal.onApplicationEvent(event);
   }
+
+  @Test
+  public void testTriggerAfterRegistryEvent() {
+    CseApplicationListener cal = new CseApplicationListener();
+
+    Collection<BootListener> listeners = new ArrayList<>(1);
+    BootListener listener = Mockito.mock(BootListener.class);
+    listeners.add(listener);
+    Deencapsulation.setField(cal, "bootListenerList", listeners);
+
+    MicroserviceInstance microserviceInstance = Mockito.mock(MicroserviceInstance.class);
+    new Expectations(RegistryUtils.class) {
+      {
+        RegistryUtils.getMicroserviceInstance();
+        result = microserviceInstance;
+      }
+    };
+    Mockito.when(microserviceInstance.getInstanceId()).thenReturn("testInstanceId");
+
+    Deencapsulation.invoke(cal, "triggerAfterRegistryEvent");
+
+    EventManager.post(Mockito.mock(MicroserviceInstanceRegisterTask.class));
+
+    Deencapsulation.invoke(ReferenceConfigUtils.class, "assertIsReady");
+
+    // AFTER_REGISTRY event should only be sent at the first time of registry success.
+    EventManager.post(Mockito.mock(MicroserviceInstanceRegisterTask.class));
+    verify(listener, times(1)).onBootEvent(Mockito.any(BootEvent.class));
+  }
+
+  @Test
+  public void testTriggerAfterRegistryEventOnInstanceIdIsNull() {
+    CseApplicationListener cal = new CseApplicationListener();
+
+    Collection<BootListener> listeners = new ArrayList<>(1);
+    BootListener listener = Mockito.mock(BootListener.class);
+    listeners.add(listener);
+    Deencapsulation.setField(cal, "bootListenerList", listeners);
+
+    MicroserviceInstance microserviceInstance = Mockito.mock(MicroserviceInstance.class);
+    new Expectations(RegistryUtils.class) {
+      {
+        RegistryUtils.getMicroserviceInstance();
+        result = microserviceInstance;
+      }
+    };
+    Mockito.when(microserviceInstance.getInstanceId()).thenReturn(null).thenReturn("testInstanceId");
+
+    Deencapsulation.invoke(cal, "triggerAfterRegistryEvent");
+
+    EventManager.post(Mockito.mock(MicroserviceInstanceRegisterTask.class));
+
+    try {
+      Deencapsulation.invoke(ReferenceConfigUtils.class, "assertIsReady");
+      fail("an exception is expected.");
+    } catch (Exception e) {
+      Assert.assertEquals(IllegalStateException.class, e.getClass());
+    }
+    verify(listener, times(0)).onBootEvent(Mockito.any(BootEvent.class));
+
+    // AFTER_REGISTRY event should only be sent at the first time of registry success.
+    EventManager.post(Mockito.mock(MicroserviceInstanceRegisterTask.class));
+    Deencapsulation.invoke(ReferenceConfigUtils.class, "assertIsReady");
+    verify(listener, times(1)).onBootEvent(Mockito.any(BootEvent.class));
+  }
 }