JVMCBC-560: Do not mark Node as disconnected if service is degraded

Motivation
----------
Before this change, if a Node has a service pool that is degraded
(and no other service in a non-idle state), then the Node would
mark itself in a DISCONNECTED state which also would cause a
NodeDisconnectedEvent to be sent into the event bus.

The node might end up in a situation like the following: the node
only has the config service and a query service enabled. The
config service is idle since it is not used, and if the query
pool goes from 1 to 2 endpoints it might temporarily go into
a degraded state until the endpoint is connected as well. Since
the service is now degraded and the other service is idle, without
any changes the node ends up in a DISCONNECTED state.

Modifications
-------------
The ServiceStateZipper did not take the degraded state into account,
which caused it to switch into the disconnected branch and this
is not correct.

This change makes the zipper recognize the degraded state from the
upstream components and it will put the Node into a degraded
state as well - this is largely an internal state and does not
emit a disconnected event.

Result
------
The SDK does not emit a NodeDisconnectedEvent if the node isn't
actually disconnected.

Change-Id: I7341e02d7c4fbfa81122e100855718b072838d3c
Reviewed-on: http://review.couchbase.org/97197
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Graham Pople <grahampople@gmail.com>

src/main/java/com/couchbase/client/core/node/ServiceStateZipper.java

@@ -46,6 +46,7 @@ protected LifecycleState zipWith(Collection<LifecycleState> states) {
         int connecting = 0;
         int disconnecting = 0;
         int idle = 0;
+        int degraded = 0;
         for (LifecycleState serviceState : states) {
             switch (serviceState) {
                 case CONNECTED:
@@ -57,16 +58,23 @@ protected LifecycleState zipWith(Collection<LifecycleState> states) {
                 case DISCONNECTING:
                     disconnecting++;
                     break;
+                case DEGRADED:
+                    degraded++;
+                    break;
                 case IDLE:
                     idle++;
+                case DISCONNECTED:
+                    // Intentionally ignored.
                     break;
+                default:
+                    throw new IllegalStateException("Unknown unhandled state " + serviceState + ", this is a bug!");
             }
         }
         if (states.size() == idle) {
             return LifecycleState.IDLE;
         } else if (states.size() == (connected + idle)) {
             return LifecycleState.CONNECTED;
-        } else if (connected > 0) {
+        } else if (connected > 0 || degraded > 0) {
             return LifecycleState.DEGRADED;
         } else if (connecting > 0) {
             return LifecycleState.CONNECTING;

src/test/java/com/couchbase/client/core/node/ServiceStateZipperTest.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2018 Couchbase, Inc.
+ *
+ * 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 com.couchbase.client.core.node;
+
+import com.couchbase.client.core.service.Service;
+import com.couchbase.client.core.state.LifecycleState;
+import org.junit.Test;
+import rx.Observable;
+import rx.subjects.BehaviorSubject;
+import rx.subjects.Subject;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Verifies the state transitions of the {@link ServiceStateZipper}.
+ */
+public class ServiceStateZipperTest {
+
+    @Test
+    public void shouldHandleDegradedService() {
+        ServiceStateZipper zipper = new ServiceStateZipper(LifecycleState.DISCONNECTED);
+
+        assertEquals(LifecycleState.DISCONNECTED, zipper.state());
+
+        Service configService = mock(Service.class);
+        when(configService.states()).thenReturn(Observable.just(LifecycleState.IDLE));
+        zipper.register(configService, configService);
+
+        assertEquals(LifecycleState.IDLE, zipper.state());
+
+        Service queryService = mock(Service.class);
+        when(queryService.states()).thenReturn(Observable.just(LifecycleState.DEGRADED));
+        zipper.register(queryService, queryService);
+
+        assertEquals(LifecycleState.DEGRADED, zipper.state());
+    }
+}