SONAR-2649 Add server property 'sonar.notifications.delay'

Can be defined in conf/sonar.properties and specifies delay in seconds
between processing of notifications queue, default value is 60.

3 files changed, 65 insertions, 42 deletions

diff --git a/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java b/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
index b408a05bfd7..bdd2c3434f2 100644
--- a/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
+++ b/sonar-server/src/main/java/org/sonar/server/notifications/NotificationService.java
@@ -19,11 +19,16 @@
  */
 package org.sonar.server.notifications;
 
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.SetMultimap;
-import com.google.common.collect.Sets;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.configuration.Configuration;
 import org.sonar.api.ServerComponent;
 import org.sonar.api.database.DatabaseSession;
 import org.sonar.api.database.configuration.Property;
@@ -31,26 +36,28 @@ import org.sonar.api.database.model.User;
 import org.sonar.api.notifications.Notification;
 import org.sonar.api.notifications.NotificationChannel;
 import org.sonar.api.notifications.NotificationDispatcher;
+import org.sonar.api.utils.Logs;
 import org.sonar.api.utils.TimeProfiler;
 import org.sonar.core.notifications.DefaultNotificationManager;
 import org.sonar.jpa.entity.NotificationQueueElement;
 import org.sonar.jpa.session.DatabaseSessionFactory;
 
-import java.util.*;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.SetMultimap;
+import com.google.common.collect.Sets;
 
 /**
  * @since 2.10
  */
 public class NotificationService implements ServerComponent {
 
-  private static final Logger LOG = LoggerFactory.getLogger(NotificationService.class);
-  private static final TimeProfiler TIME_PROFILER = new TimeProfiler(LOG);
+  private static final TimeProfiler TIME_PROFILER = new TimeProfiler(Logs.INFO).setLevelToDebug();
+
+  private static final String DELAY = "sonar.notifications.delay";
+  private static final long DELAY_DEFAULT = 60;
 
   private ScheduledExecutorService executorService;
-  private long period = 10; // FIXME small value just for tests
+  private long delay;
 
   private DatabaseSessionFactory sessionFactory;
   private DefaultNotificationManager manager;
@@ -60,61 +67,58 @@ public class NotificationService implements ServerComponent {
   /**
    * Default constructor when no channels.
    */
-  public NotificationService(DatabaseSessionFactory sessionFactory, DefaultNotificationManager manager, NotificationDispatcher[] dispatchers) {
-    this(sessionFactory, manager, dispatchers, new NotificationChannel[0]);
-    LOG.warn("There is no channels - all notifications would be ignored!");
+  public NotificationService(Configuration configuration, DatabaseSessionFactory sessionFactory, DefaultNotificationManager manager, NotificationDispatcher[] dispatchers) {
+    this(configuration, sessionFactory, manager, dispatchers, new NotificationChannel[0]);
+    Logs.INFO.warn("There is no channels - all notifications would be ignored!");
   }
 
-  public NotificationService(DatabaseSessionFactory sessionFactory, DefaultNotificationManager manager, NotificationDispatcher[] dispatchers, NotificationChannel[] channels) {
+  public NotificationService(Configuration configuration, DatabaseSessionFactory sessionFactory, DefaultNotificationManager manager, NotificationDispatcher[] dispatchers,
+      NotificationChannel[] channels) {
+    delay = configuration.getLong(DELAY, DELAY_DEFAULT);
     this.sessionFactory = sessionFactory;
     this.manager = manager;
     this.channels = channels;
     this.dispatchers = dispatchers;
   }
 
-  /**
-   * Visibility has been relaxed for tests.
-   */
-  void setPeriod(long milliseconds) {
-    this.period = milliseconds;
-  }
-
   public void start() {
     executorService = Executors.newSingleThreadScheduledExecutor();
     executorService.scheduleWithFixedDelay(new Runnable() {
       public void run() {
         processQueue();
       }
-    }, 0, period, TimeUnit.MILLISECONDS);
-    LOG.info("Notification service started");
+    }, 0, delay, TimeUnit.SECONDS);
+    Logs.INFO.info("Notification service started (delay {} sec.)", delay);
   }
 
   public void stop() {
     try {
-      executorService.awaitTermination(period, TimeUnit.MILLISECONDS);
+      executorService.awaitTermination(delay, TimeUnit.SECONDS);
       executorService.shutdown();
     } catch (InterruptedException e) {
-      LOG.error("Error during stop of notification service", e);
+      Logs.INFO.error("Error during stop of notification service", e);
     }
-    LOG.info("Notification service stopped");
+    Logs.INFO.info("Notification service stopped");
   }
 
   /**
    * Visibility has been relaxed for tests.
    */
   void processQueue() {
+    TIME_PROFILER.start("Processing notifications queue");
     NotificationQueueElement queueElement = manager.getFromQueue();
     while (queueElement != null) {
       deliver(queueElement.getNotification());
       queueElement = manager.getFromQueue();
     }
+    TIME_PROFILER.stop();
   }
 
   /**
    * Visibility has been relaxed for tests.
    */
   void deliver(Notification notification) {
-    TIME_PROFILER.start("Delivering notification " + notification);
+    Logs.INFO.debug("Delivering notification " + notification);
     SetMultimap<String, NotificationChannel> recipients = HashMultimap.create();
     for (NotificationChannel channel : channels) {
       for (NotificationDispatcher dispatcher : dispatchers) {
@@ -126,7 +130,11 @@ public class NotificationService implements ServerComponent {
             }
           }
         };
-        dispatcher.dispatch(notification, context);
+        try {
+          dispatcher.dispatch(notification, context);
+        } catch (Exception e) { // catch all exceptions in order to dispatch using other dispatchers
+          Logs.INFO.warn("Unable to dispatch notification " + notification + " using " + dispatcher, e);
+        }
         for (String username : possibleRecipients) {
           if (isEnabled(username, channel, dispatcher)) {
             recipients.put(username, channel);
@@ -137,12 +145,15 @@ public class NotificationService implements ServerComponent {
     for (Map.Entry<String, Collection<NotificationChannel>> entry : recipients.asMap().entrySet()) {
       String username = entry.getKey();
       Collection<NotificationChannel> userChannels = entry.getValue();
-      LOG.info("For user {} via {}", username, userChannels);
+      Logs.INFO.debug("For user {} via {}", username, userChannels);
       for (NotificationChannel channel : userChannels) {
-        channel.deliver(notification, username);
+        try {
+          channel.deliver(notification, username);
+        } catch (Exception e) { // catch all exceptions in order to deliver via other channels
+          Logs.INFO.warn("Unable to deliver notification " + notification + " for user " + username + " via " + channel, e);
+        }
       }
     }
-    TIME_PROFILER.stop();
   }
 
   public List<NotificationDispatcher> getDispatchers() {
diff --git a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceDbTest.java b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceDbTest.java
index a7a57c9044b..82f1c4fbad7 100644
--- a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceDbTest.java
+++ b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceDbTest.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import org.apache.commons.configuration.Configuration;
 import org.junit.Before;
 import org.junit.Test;
 import org.sonar.api.notifications.NotificationChannel;
@@ -37,11 +38,11 @@ public class NotificationServiceDbTest extends AbstractDbUnitTestCase {
   @Before
   public void setUp() {
     setupData("fixture");
-    notificationService = new NotificationService(getSessionFactory(), null, null);
+    notificationService = new NotificationService(mock(Configuration.class), getSessionFactory(), null, null);
   }
 
   @Test
-  public void should() {
+  public void shouldCheckEnablement() {
     NotificationChannel email = mock(NotificationChannel.class);
     when(email.getKey()).thenReturn("EmailNotificationChannel");
     NotificationDispatcher commentOnReviewAssignedToMe = mock(NotificationDispatcher.class);
diff --git a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
index d17ada17562..1bc824b1413 100644
--- a/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
+++ b/sonar-server/src/test/java/org/sonar/server/notifications/NotificationServiceTest.java
@@ -20,8 +20,18 @@
 package org.sonar.server.notifications;
 
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.*;
-
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
@@ -77,7 +87,9 @@ public class NotificationServiceTest {
     NotificationDispatcher[] dispatchers = new NotificationDispatcher[] { commentOnReviewAssignedToMe, commentOnReviewCreatedByMe };
     NotificationChannel[] channels = new NotificationChannel[] { emailChannel, gtalkChannel };
     manager = mock(DefaultNotificationManager.class);
-    service = spy(new NotificationService(null, manager, dispatchers, channels));
+    Configuration configuration = new BaseConfiguration();
+    configuration.setProperty("sonar.notifications.delay", "1"); // delay 1 second
+    service = spy(new NotificationService(configuration, null, manager, dispatchers, channels));
     doReturn(false).when(service).isEnabled(any(String.class), any(NotificationChannel.class), any(NotificationDispatcher.class));
   }
 
@@ -86,15 +98,14 @@ public class NotificationServiceTest {
     NotificationQueueElement queueElement = mock(NotificationQueueElement.class);
     Notification notification = mock(Notification.class);
     when(queueElement.getNotification()).thenReturn(notification);
-    when(manager.getFromQueue()).thenReturn(queueElement).thenReturn(null);
+    when(manager.getFromQueue()).thenReturn(queueElement).thenReturn(null).thenReturn(queueElement).thenReturn(null).thenReturn(queueElement).thenReturn(null);
     doNothing().when(service).deliver(any(Notification.class));
 
-    service.setPeriod(10);
     service.start();
-    Thread.sleep(50);
+    Thread.sleep(1500); // sleep 1.5 second to process queue
     service.stop();
 
-    verify(service).deliver(notification);
+    verify(service, times(3)).deliver(notification); // 3 times - 1 on start, 1 after delay, 1 on stop
   }
 
   /**