REFA: Trigger the notification updates via reactive fluxxes

spring-boot-admin-samples/spring-boot-admin-sample-zookeeper/pom.xml

@@ -34,6 +34,10 @@
             <artifactId>spring-boot-admin-server-ui</artifactId>
         </dependency-->
         <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-zookeeper-discovery</artifactId>
         </dependency>

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/config/HazelcastStoreConfiguration.java

@@ -23,7 +23,6 @@ import de.codecentric.boot.admin.server.model.ApplicationId;
 import de.codecentric.boot.admin.server.registry.store.ApplicationStore;
 import de.codecentric.boot.admin.server.registry.store.HazelcastApplicationStore;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
@@ -31,7 +30,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
 import org.springframework.boot.autoconfigure.hazelcast.HazelcastAutoConfiguration;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import com.hazelcast.core.HazelcastInstance;
@@ -50,15 +48,9 @@ public class HazelcastStoreConfiguration {
     @Value("${spring.boot.admin.hazelcast.event-store:spring-boot-admin-event-store}")
     private String eventListName;
 
-    @Autowired
-    private ApplicationEventPublisher publisher;
-
-    @Autowired
-    private HazelcastInstance hazelcastInstance;
-
     @Bean
     @ConditionalOnMissingBean
-    public ApplicationStore applicationStore() {
+    public ApplicationStore applicationStore(HazelcastInstance hazelcastInstance) {
         IMap<ApplicationId, Application> map = hazelcastInstance.getMap(hazelcastMapName);
         map.addIndex("registration.name", false);
         return new HazelcastApplicationStore(map);
@@ -66,7 +58,7 @@ public class HazelcastStoreConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
-    public ClientApplicationEventStore eventStore() {
+    public ClientApplicationEventStore eventStore(HazelcastInstance hazelcastInstance) {
         IList<ClientApplicationEvent> list = hazelcastInstance.getList(eventListName);
         return new HazelcastEventStore(list);
     }

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/config/NotifierConfiguration.java

 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2014-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,19 +15,20 @@
  */
 package de.codecentric.boot.admin.server.config;
 
+import de.codecentric.boot.admin.server.event.ClientApplicationEvent;
 import de.codecentric.boot.admin.server.notify.CompositeNotifier;
 import de.codecentric.boot.admin.server.notify.HipchatNotifier;
 import de.codecentric.boot.admin.server.notify.LetsChatNotifier;
 import de.codecentric.boot.admin.server.notify.MailNotifier;
+import de.codecentric.boot.admin.server.notify.NotificationTrigger;
 import de.codecentric.boot.admin.server.notify.Notifier;
-import de.codecentric.boot.admin.server.notify.NotifierListener;
 import de.codecentric.boot.admin.server.notify.SlackNotifier;
 import de.codecentric.boot.admin.server.notify.filter.FilteringNotifier;
 import de.codecentric.boot.admin.server.notify.filter.web.NotificationFilterController;
 import de.codecentric.boot.admin.server.web.PrefixHandlerMapping;
 
 import java.util.List;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.reactivestreams.Publisher;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
@@ -48,20 +49,17 @@ public class NotifierConfiguration {
 
     @Configuration
     @ConditionalOnBean(Notifier.class)
-    public static class NotifierListenerConfiguration {
-        @Autowired
-        private Notifier notifier;
-
-        @Bean
+    public static class NotifierTriggerConfiguration {
+        @Bean(initMethod = "start", destroyMethod = "stop")
         @ConditionalOnMissingBean
-        public NotifierListener notifierListener() {
-            return new NotifierListener(notifier);
+        public NotificationTrigger notificationTrigger(Notifier notifier, Publisher<ClientApplicationEvent> events) {
+            return new NotificationTrigger(notifier, events);
         }
     }
 
     @Configuration
     @ConditionalOnBean(Notifier.class)
-    @AutoConfigureBefore({NotifierListenerConfiguration.class})
+    @AutoConfigureBefore({NotifierTriggerConfiguration.class})
     public static class CompositeNotifierConfiguration {
         @Bean
         @Primary
@@ -84,12 +82,14 @@ public class NotifierConfiguration {
     @Configuration
     @ConditionalOnSingleCandidate(FilteringNotifier.class)
     public static class FilteringNotifierWebConfiguration {
+        private final FilteringNotifier filteringNotifier;
+        private final AdminServerProperties adminServer;
 
-        @Autowired
-        private FilteringNotifier filteringNotifier;
-
-        @Autowired
-        private AdminServerProperties adminServer;
+        public FilteringNotifierWebConfiguration(FilteringNotifier filteringNotifier,
+                                                 AdminServerProperties adminServer) {
+            this.filteringNotifier = filteringNotifier;
+            this.adminServer = adminServer;
+        }
 
         @Bean
         public NotificationFilterController notificationFilterController() {
@@ -107,22 +107,19 @@ public class NotifierConfiguration {
     @Configuration
     @ConditionalOnBean(MailSender.class)
     @AutoConfigureAfter({MailSenderAutoConfiguration.class})
-    @AutoConfigureBefore({NotifierListenerConfiguration.class, CompositeNotifierConfiguration.class})
+    @AutoConfigureBefore({NotifierTriggerConfiguration.class, CompositeNotifierConfiguration.class})
     public static class MailNotifierConfiguration {
-        @Autowired
-        private MailSender mailSender;
-
         @Bean
         @ConditionalOnMissingBean
         @ConfigurationProperties("spring.boot.admin.notify.mail")
-        public MailNotifier mailNotifier() {
+        public MailNotifier mailNotifier(MailSender mailSender) {
             return new MailNotifier(mailSender);
         }
     }
 
     @Configuration
     @ConditionalOnProperty(prefix = "spring.boot.admin.notify.hipchat", name = "url")
-    @AutoConfigureBefore({NotifierListenerConfiguration.class, CompositeNotifierConfiguration.class})
+    @AutoConfigureBefore({NotifierTriggerConfiguration.class, CompositeNotifierConfiguration.class})
     public static class HipchatNotifierConfiguration {
         @Bean
         @ConditionalOnMissingBean
@@ -134,7 +131,7 @@ public class NotifierConfiguration {
 
     @Configuration
     @ConditionalOnProperty(prefix = "spring.boot.admin.notify.slack", name = "webhook-url")
-    @AutoConfigureBefore({NotifierListenerConfiguration.class, CompositeNotifierConfiguration.class})
+    @AutoConfigureBefore({NotifierTriggerConfiguration.class, CompositeNotifierConfiguration.class})
     public static class SlackNotifierConfiguration {
         @Bean
         @ConditionalOnMissingBean
@@ -146,7 +143,7 @@ public class NotifierConfiguration {
 
     @Configuration
     @ConditionalOnProperty(prefix = "spring.boot.admin.notify.letschat", name = "url")
-    @AutoConfigureBefore({NotifierListenerConfiguration.class, CompositeNotifierConfiguration.class})
+    @AutoConfigureBefore({NotifierTriggerConfiguration.class, CompositeNotifierConfiguration.class})
     public static class LetsChatNotifierConfiguration {
         @Bean
         @ConditionalOnMissingBean

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/notify/NotificationTrigger.java

+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package de.codecentric.boot.admin.server.notify;
+
+import de.codecentric.boot.admin.server.event.ClientApplicationEvent;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import java.util.logging.Level;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationTrigger {
+    private static final Logger log = LoggerFactory.getLogger(NotificationTrigger.class);
+    private final Notifier notifier;
+    private final Publisher<ClientApplicationEvent> events;
+    private Disposable subscription;
+
+    public NotificationTrigger(Notifier notifier, Publisher<ClientApplicationEvent> events) {
+        this.notifier = notifier;
+        this.events = events;
+    }
+
+    public void start() {
+        log.debug("Subscribed to {} events for notifications", ClientApplicationEvent.class);
+        subscription = Flux.from(events)
+                           .log(log.getName(), Level.FINEST)
+                           .subscribeOn(Schedulers.newSingle("notifications"))
+                           .doOnNext(this::sendNotifications)
+                           .retry()
+                           .subscribe();
+    }
+
+    public void stop() {
+        if (subscription != null) {
+            subscription.dispose();
+        }
+    }
+
+    protected void sendNotifications(ClientApplicationEvent event) {
+        notifier.notify(event);
+    }
+}

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/notify/NotifierListener.java

-package de.codecentric.boot.admin.server.notify;
-
-import de.codecentric.boot.admin.server.event.ClientApplicationEvent;
-
-import org.springframework.context.event.EventListener;
-
-public class NotifierListener {
-    private final Notifier notifier;
-
-    public NotifierListener(Notifier notifier) {
-        this.notifier = notifier;
-    }
-
-    @EventListener
-    public void onClientApplicationEvent(ClientApplicationEvent event) {
-        notifier.notify(event);
-    }
-}

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/config/NotifierConfigurationTest.java

 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -24,8 +24,8 @@ import de.codecentric.boot.admin.server.model.StatusInfo;
 import de.codecentric.boot.admin.server.notify.CompositeNotifier;
 import de.codecentric.boot.admin.server.notify.HipchatNotifier;
 import de.codecentric.boot.admin.server.notify.MailNotifier;
+import de.codecentric.boot.admin.server.notify.NotificationTrigger;
 import de.codecentric.boot.admin.server.notify.Notifier;
-import de.codecentric.boot.admin.server.notify.NotifierListener;
 import de.codecentric.boot.admin.server.notify.SlackNotifier;
 
 import java.util.ArrayList;
@@ -33,6 +33,7 @@ import java.util.List;
 import org.junit.After;
 import org.junit.Test;
 import org.springframework.boot.autoconfigure.mail.MailSenderAutoConfiguration;
+import org.springframework.boot.autoconfigure.web.client.RestTemplateAutoConfiguration;
 import org.springframework.boot.test.util.TestPropertyValues;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Primary;
@@ -65,7 +66,7 @@ public class NotifierConfigurationTest {
     @Test
     public void test_no_notifierListener() {
         load(null);
-        assertThat(context.getBeansOfType(NotifierListener.class)).isEmpty();
+        assertThat(context.getBeansOfType(NotificationTrigger.class)).isEmpty();
     }
 
     @Test
@@ -105,6 +106,8 @@ public class NotifierConfigurationTest {
         if (config != null) {
             context.register(config);
         }
+        context.register(RestTemplateAutoConfiguration.class);
+        context.register(AdminServerCoreConfiguration.class);
         context.register(MailSenderAutoConfiguration.class);
         context.register(NotifierConfiguration.class);
 

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/notify/NotificationTriggerTest.java

+/*
+ * Copyright 2014-2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package de.codecentric.boot.admin.server.notify;
+
+import de.codecentric.boot.admin.server.event.ClientApplicationEvent;
+import de.codecentric.boot.admin.server.event.ClientApplicationRegisteredEvent;
+import de.codecentric.boot.admin.server.event.ClientApplicationStatusChangedEvent;
+import de.codecentric.boot.admin.server.model.Application;
+import de.codecentric.boot.admin.server.model.ApplicationId;
+import de.codecentric.boot.admin.server.model.Registration;
+import de.codecentric.boot.admin.server.model.StatusInfo;
+import reactor.test.publisher.TestPublisher;
+
+import org.junit.Test;
+
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class NotificationTriggerTest {
+    private final Application application = Application.create(ApplicationId.of("id-1"),
+            Registration.create("foo", "http://health-1").build()).build();
+
+    @Test
+    public void should_notify_on_event() throws InterruptedException {
+        //given
+        Notifier notifier = mock(Notifier.class);
+        TestPublisher<ClientApplicationEvent> events = TestPublisher.create();
+        NotificationTrigger trigger = new NotificationTrigger(notifier, events);
+        trigger.start();
+        doNothing().when(notifier).notify(isA(ClientApplicationEvent.class));
+
+        //when registered event is emitted
+        ClientApplicationStatusChangedEvent event = new ClientApplicationStatusChangedEvent(application,
+                StatusInfo.ofUp(), StatusInfo.ofDown());
+        events.next(event);
+        //then should notify
+        verify(notifier, times(1)).notify(event);
+
+        //when registered event is emitted but the trigger has been stopped
+        trigger.stop();
+        reset(notifier);
+        events.next(new ClientApplicationRegisteredEvent(application, application.getRegistration()));
+        //then should not notify
+        verify(notifier, never()).notify(event);
+    }
+}
\ No newline at end of file

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/registry/InfoUpdateTriggerTest.java

@@ -27,6 +27,8 @@ import reactor.test.publisher.TestPublisher;
 
 import org.junit.Test;
 
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
@@ -44,6 +46,7 @@ public class InfoUpdateTriggerTest {
         TestPublisher<ClientApplicationEvent> events = TestPublisher.create();
         InfoUpdateTrigger trigger = new InfoUpdateTrigger(updater, events);
         trigger.start();
+        doNothing().when(updater).updateInfo(isA(Application.class));
 
         //when some non-registered event is emitted
         events.next(new ClientApplicationRegisteredEvent(application, application.getRegistration()));

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/registry/StatusUpdateTriggerTest.java

@@ -27,7 +27,9 @@ import reactor.test.publisher.TestPublisher;
 
 import org.junit.Test;
 
+import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
@@ -67,6 +69,7 @@ public class StatusUpdateTriggerTest {
         TestPublisher<ClientApplicationEvent> events = TestPublisher.create();
         StatusUpdateTrigger trigger = new StatusUpdateTrigger(updater, events);
         trigger.start();
+        doNothing().when(updater).updateStatus(isA(Application.class));
 
         //when some non-registered event is emitted
         events.next(new ClientApplicationInfoChangedEvent(application, Info.empty()));