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