Merge pull request #191 from splitio/occupancy-sec-reg

Adding second region

Author: ldecheverz-split

client/src/main/java/io/split/engine/sse/PushStatusTrackerImp.java

@@ -1,5 +1,6 @@
 package io.split.engine.sse;
 
+import com.google.common.collect.Maps;
 import io.split.engine.common.PushManager;
 import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.ControlNotification;
@@ -9,6 +10,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -20,6 +22,7 @@ public class PushStatusTrackerImp implements PushStatusTracker {
     private final AtomicReference<SSEClient.StatusMessage> _sseStatus = new AtomicReference<>(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
     private final AtomicReference<ControlType> _backendStatus = new AtomicReference<>(ControlType.STREAMING_RESUMED);
     private final LinkedBlockingQueue<PushManager.Status> _statusMessages;
+    private final ConcurrentMap<String, Integer> regions = Maps.newConcurrentMap();
 
     public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessages) {
         _statusMessages = statusMessages;
@@ -98,9 +101,11 @@ public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotifica
         _log.debug(String.format("handleIncomingOccupancyEvent: publishers=%d", occupancyNotification.getMetrics().getPublishers()));
 
         int publishers = occupancyNotification.getMetrics().getPublishers();
-        if (publishers <= 0 && _publishersOnline.compareAndSet(true, false) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
+        regions.put(occupancyNotification.getChannel(), publishers);
+        boolean isPublishers = isPublishers();
+        if (!isPublishers && _publishersOnline.compareAndSet(true, false) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
             _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
-        } else if (publishers >= 1 && _publishersOnline.compareAndSet(false, true) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
+        } else if (isPublishers && _publishersOnline.compareAndSet(false, true) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
             _statusMessages.offer(PushManager.Status.STREAMING_READY);
         }
     }
@@ -114,6 +119,7 @@ public void handleIncomingAblyError(ErrorNotification notification) {
         }
         if (notification.getCode() >= 40140 && notification.getCode() <= 40149) {
             _statusMessages.offer(PushManager.Status.STREAMING_BACKOFF);
+            return;
         }
         if (notification.getCode() >= 40000 && notification.getCode() <= 49999) {
             _statusMessages.offer(PushManager.Status.STREAMING_OFF);
@@ -129,4 +135,13 @@ public synchronized void forcePushDisable() {
         _backendStatus.set(ControlType.STREAMING_DISABLED);
         _statusMessages.offer(PushManager.Status.STREAMING_OFF);
     }
+
+    private boolean isPublishers() {
+        for(Integer publisher : regions.values()) {
+            if (publisher > 0) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

client/src/main/java/io/split/engine/sse/dtos/OccupancyNotification.java

@@ -4,7 +4,6 @@
 import io.split.engine.sse.NotificationProcessor;
 
 public class OccupancyNotification extends IncomingNotification implements StatusNotification {
-    private static final String CONTROL_PRI_CHANNEL = "control_pri";
     private final OccupancyMetrics metrics;
 
     public OccupancyNotification(GenericNotificationData genericNotificationData) {
@@ -23,9 +22,7 @@ public void handler(NotificationProcessor notificationProcessor) {
 
     @Override
     public void handlerStatus(PushStatusTracker notificationManagerKeeper) {
-        if (CONTROL_PRI_CHANNEL.equals(getChannel())) {
-            notificationManagerKeeper.handleIncomingOccupancyEvent(this);
-        }
+        notificationManagerKeeper.handleIncomingOccupancyEvent(this);
     }
 
     @Override

client/src/test/java/io/split/engine/sse/PushStatusTrackerTest.java

@@ -37,7 +37,7 @@ public void HandleControlEventStreamingResumedShouldNotifyEvent() throws Interru
     @Test
     public void HandleControlEventStreamingResumedShouldNotNotifyEvent() {
         LinkedBlockingQueue<PushManager.Status> messages = new LinkedBlockingQueue<>();
-        OccupancyNotification occupancyNotification = buildOccupancyNotification(0);
+        OccupancyNotification occupancyNotification = buildOccupancyNotification(0, null);
         ControlNotification controlNotification = buildControlNotification(ControlType.STREAMING_RESUMED);
 
         PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(messages);
@@ -65,7 +65,7 @@ public void HandleControlEventStreamingDisabledShouldNotifyShutdownEvent() {
     @Test
     public void HandleOccupancyEventWithPublishersFirstTimeShouldNotNotifyEvent() {
         LinkedBlockingQueue<PushManager.Status> messages = new LinkedBlockingQueue<>();
-        OccupancyNotification occupancyNotification = buildOccupancyNotification(2);
+        OccupancyNotification occupancyNotification = buildOccupancyNotification(2, null);
 
         PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(messages);
         pushStatusTracker.handleIncomingOccupancyEvent(occupancyNotification);
@@ -76,8 +76,23 @@ public void HandleOccupancyEventWithPublishersFirstTimeShouldNotNotifyEvent() {
     public void HandleOccupancyEventWithPublishersAndWithStreamingDisabledShouldNotifyEvent() throws InterruptedException {
         LinkedBlockingQueue<PushManager.Status> messages = new LinkedBlockingQueue<>();
         PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(messages);
-        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(0));
-        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(2));
+        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(0, null));
+        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(2, null));
+
+        assertThat(messages.size(), is(equalTo(2)));
+        PushManager.Status m1 = messages.take();
+        assertThat(m1, is(equalTo(PushManager.Status.STREAMING_DOWN)));
+
+        PushManager.Status m2 = messages.take();
+        assertThat(m2, is(equalTo(PushManager.Status.STREAMING_READY)));
+    }
+
+    @Test
+    public void HandleOccupancyEventWithDifferentChannelsPublishersShouldNotifyEvent() throws InterruptedException {
+        LinkedBlockingQueue<PushManager.Status> messages = new LinkedBlockingQueue<>();
+        PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(messages);
+        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(0, "control_pri"));
+        pushStatusTracker.handleIncomingOccupancyEvent(buildOccupancyNotification(2, "control_sec"));
 
         assertThat(messages.size(), is(equalTo(2)));
         PushManager.Status m1 = messages.take();
@@ -88,14 +103,14 @@ public void HandleOccupancyEventWithPublishersAndWithStreamingDisabledShouldNoti
     }
 
     private ControlNotification buildControlNotification(ControlType controlType) {
-        return new ControlNotification(buildGenericData(controlType, IncomingNotification.Type.CONTROL,null));
+        return new ControlNotification(buildGenericData(controlType, IncomingNotification.Type.CONTROL,null, null));
     }
 
-    private OccupancyNotification buildOccupancyNotification(int publishers) {
-        return new OccupancyNotification(buildGenericData(null, IncomingNotification.Type.OCCUPANCY, publishers));
+    private OccupancyNotification buildOccupancyNotification(int publishers, String channel) {
+        return new OccupancyNotification(buildGenericData(null, IncomingNotification.Type.OCCUPANCY, publishers, channel));
     }
 
-    private GenericNotificationData buildGenericData(ControlType controlType, IncomingNotification.Type type, Integer publishers) {
+    private GenericNotificationData buildGenericData(ControlType controlType, IncomingNotification.Type type, Integer publishers, String channel) {
         return new GenericNotificationData(
                 null,
                 null,
@@ -104,6 +119,6 @@ private GenericNotificationData buildGenericData(ControlType controlType, Incomi
                 publishers != null ? new OccupancyMetrics(publishers) : null,
                 null,
                 type,
-                "channel-test");
+                channel == null ? "channel-test" : channel);
     }
 }