Add unit tests for Watcher, some code cleanup

Author: russgold

pom.xml

@@ -463,6 +463,12 @@
             <version>1.3</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>18.0</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.httpunit</groupId>
             <artifactId>httpunit</artifactId>

src/main/java/oracle/kubernetes/operator/DomainWatcher.java

@@ -5,10 +5,11 @@
 
 import io.kubernetes.client.ApiException;
 import io.kubernetes.client.util.Watch;
+import oracle.kubernetes.operator.builders.WatchBuilder;
+import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.domain.model.oracle.kubernetes.weblogic.domain.v1.Domain;
 import oracle.kubernetes.operator.helpers.ClientHelper;
 import oracle.kubernetes.operator.helpers.ClientHolder;
-import oracle.kubernetes.operator.builders.WatchBuilder;
 import oracle.kubernetes.operator.watcher.Watcher;
 import oracle.kubernetes.operator.watcher.Watching;
 import oracle.kubernetes.operator.watcher.WatchingEventDestination;
@@ -50,7 +51,7 @@ public void run() {
     ClientHolder client = helper.take();
     try {
       Watching<Domain> w = createWatching(client);
-      Watcher<Domain> watcher = new Watcher<Domain>(w, null, initialResourceVersion);
+      Watcher<Domain> watcher = new Watcher<>(w, initialResourceVersion);
 
       // invoke watch on current Thread.  Won't return until watch stops
       watcher.doWatch();
@@ -66,13 +67,12 @@ protected Watching<Domain> createWatching(ClientHolder client) {
       /**
        * Watcher callback to issue the list Domain changes. It is driven by the
        * Watcher wrapper to issue repeated watch requests.
-       * @param context user defined contact object or null
        * @param resourceVersion resource version to omit older events
        * @return Watch object or null if the operation should end
        * @throws ApiException if there is an API error.
        */
       @Override
-      public Watch<Domain> initiateWatch(Object context, String resourceVersion) throws ApiException {
+      public WatchI<Domain> initiateWatch(String resourceVersion) throws ApiException {
         return new WatchBuilder(client)
                   .withResourceVersion(resourceVersion)
                 .createDomainWatch(ns);

src/main/java/oracle/kubernetes/operator/IngressWatcher.java

@@ -7,6 +7,7 @@
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.models.V1beta1Ingress;
 import io.kubernetes.client.util.Watch;
+import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.helpers.ClientHelper;
 import oracle.kubernetes.operator.helpers.ClientHolder;
 import oracle.kubernetes.operator.builders.WatchBuilder;
@@ -52,7 +53,7 @@ public void run() {
     ClientHolder client = helper.take();
     try {
       Watching<V1beta1Ingress> w = createWatching(client);
-      Watcher<V1beta1Ingress> watcher = new Watcher<V1beta1Ingress>(w, null, initialResourceVersion);
+      Watcher<V1beta1Ingress> watcher = new Watcher<>(w, initialResourceVersion);
 
       // invoke watch on current Thread.  Won't return until watch stops
       watcher.doWatch();
@@ -68,13 +69,12 @@ protected Watching<V1beta1Ingress> createWatching(ClientHolder client) {
       /**
        * Watcher callback to issue the list Ingress changes. It is driven by the
        * Watcher wrapper to issue repeated watch requests.
-       * @param context user defined contact object or null
        * @param resourceVersion resource version to omit older events
        * @return Watch object or null if the operation should end
        * @throws ApiException if there is an API error.
        */
       @Override
-      public Watch<V1beta1Ingress> initiateWatch(Object context, String resourceVersion) throws ApiException {
+      public WatchI<V1beta1Ingress> initiateWatch(String resourceVersion) throws ApiException {
         return new WatchBuilder(client)
                   .withResourceVersion(resourceVersion)
                   .withLabelSelector(LabelConstants.DOMAINUID_LABEL) // Any Ingress with a domainUID label

src/main/java/oracle/kubernetes/operator/PodWatcher.java

@@ -8,14 +8,15 @@
 import io.kubernetes.client.models.V1PodCondition;
 import io.kubernetes.client.models.V1PodStatus;
 import io.kubernetes.client.util.Watch;
+import oracle.kubernetes.operator.builders.WatchBuilder;
+import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.helpers.CallBuilder;
 import oracle.kubernetes.operator.helpers.ClientHelper;
 import oracle.kubernetes.operator.helpers.ClientHolder;
 import oracle.kubernetes.operator.helpers.ResponseStep;
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
 import oracle.kubernetes.operator.logging.MessageKeys;
-import oracle.kubernetes.operator.builders.WatchBuilder;
 import oracle.kubernetes.operator.watcher.Watcher;
 import oracle.kubernetes.operator.watcher.Watching;
 import oracle.kubernetes.operator.watcher.WatchingEventDestination;
@@ -77,7 +78,7 @@ public void run() {
     ClientHolder client = helper.take();
     try {
       Watching<V1Pod> w = createWatching(client);
-      Watcher<V1Pod> watcher = new Watcher<V1Pod>(w, null, initialResourceVersion);
+      Watcher<V1Pod> watcher = new Watcher<>(w, initialResourceVersion);
 
       // invoke watch on current Thread.  Won't return until watch stops
       watcher.doWatch();
@@ -93,13 +94,12 @@ private Watching<V1Pod> createWatching(ClientHolder client) {
       /**
        * Watcher callback to issue the list Pod changes. It is driven by the
        * Watcher wrapper to issue repeated watch requests.
-       * @param context user defined contact object or null
        * @param resourceVersion resource version to omit older events
        * @return Watch object or null if the operation should end
        * @throws ApiException if there is an API error.
        */
       @Override
-      public Watch<V1Pod> initiateWatch(Object context, String resourceVersion) throws ApiException {
+      public WatchI<V1Pod> initiateWatch(String resourceVersion) throws ApiException {
         return new WatchBuilder(client)
                   .withResourceVersion(resourceVersion)
                   .withLabelSelector(LabelConstants.DOMAINUID_LABEL)  // Any Pod with a domainUID label

src/main/java/oracle/kubernetes/operator/ServiceWatcher.java

@@ -7,9 +7,10 @@
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.models.V1Service;
 import io.kubernetes.client.util.Watch;
+import oracle.kubernetes.operator.builders.WatchBuilder;
+import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.helpers.ClientHelper;
 import oracle.kubernetes.operator.helpers.ClientHolder;
-import oracle.kubernetes.operator.builders.WatchBuilder;
 import oracle.kubernetes.operator.watcher.Watcher;
 import oracle.kubernetes.operator.watcher.Watching;
 import oracle.kubernetes.operator.watcher.WatchingEventDestination;
@@ -52,7 +53,7 @@ public void run() {
     ClientHolder client = helper.take();
     try {
       Watching<V1Service> w = createWatching(client);
-      Watcher<V1Service> watcher = new Watcher<V1Service>(w, null, initialResourceVersion);
+      Watcher<V1Service> watcher = new Watcher<>(w, initialResourceVersion);
 
       // invoke watch on current Thread.  Won't return until watch stops
       watcher.doWatch();
@@ -68,13 +69,12 @@ protected Watching<V1Service> createWatching(ClientHolder client) {
       /**
        * Watcher callback to issue the list Service changes. It is driven by the
        * Watcher wrapper to issue repeated watch requests.
-       * @param context user defined contact object or null
        * @param resourceVersion resource version to omit older events
        * @return Watch object or null if the operation should end
        * @throws ApiException if there is an API error.
        */
       @Override
-      public Watch<V1Service> initiateWatch(Object context, String resourceVersion) throws ApiException {
+      public WatchI<V1Service> initiateWatch(String resourceVersion) throws ApiException {
         return new WatchBuilder(client)
                   .withResourceVersion(resourceVersion)
                   .withLabelSelector(LabelConstants.DOMAINUID_LABEL)   // Any Service with a domainUID label

src/main/java/oracle/kubernetes/operator/builders/WatchBuilder.java

@@ -32,7 +32,7 @@ public class WatchBuilder {
     private CallParamsImpl callParams = new CallParamsImpl();
 
     public interface WatchFactory {
-        <T> Watch<T> createWatch(ClientHolder clientHolder, CallParams callParams, Class<?> responseBodyType, BiFunction<ClientHolder, CallParams, Call> function) throws ApiException;
+        <T> WatchI<T> createWatch(ClientHolder clientHolder, CallParams callParams, Class<?> responseBodyType, BiFunction<ClientHolder, CallParams, Call> function) throws ApiException;
     }
 
     public WatchBuilder(ClientHolder clientHolder) {
@@ -63,7 +63,7 @@ public Type getOwnerType() {
      * @return the active web hook
      * @throws ApiException if there is an error on the call that sets up the web hook.
      */
-    public Watch<V1Service> createServiceWatch(String namespace) throws ApiException {
+    public WatchI<V1Service> createServiceWatch(String namespace) throws ApiException {
         return FACTORY.createWatch(clientHolder, callParams, V1Service.class, new ListNamespacedServiceCall(namespace));
     }
 
@@ -92,7 +92,7 @@ public Call apply(ClientHolder clientHolder, CallParams callParams) {
      * @return the active web hook
      * @throws ApiException if there is an error on the call that sets up the web hook.
      */
-    public Watch<V1Pod> createPodWatch(String namespace) throws ApiException {
+    public WatchI<V1Pod> createPodWatch(String namespace) throws ApiException {
         return FACTORY.createWatch(clientHolder, callParams, V1Pod.class, new ListPodCall(namespace));
     }
 
@@ -121,7 +121,7 @@ public Call apply(ClientHolder clientHolder, CallParams callParams) {
      * @return the active web hook
      * @throws ApiException if there is an error on the call that sets up the web hook.
      */
-    public Watch<V1beta1Ingress> createIngressWatch(String namespace) throws ApiException {
+    public WatchI<V1beta1Ingress> createIngressWatch(String namespace) throws ApiException {
         return FACTORY.createWatch(clientHolder, callParams, V1beta1Ingress.class, new ListIngressCall(namespace));
     }
 
@@ -151,7 +151,7 @@ public Call apply(ClientHolder clientHolder, CallParams callParams) {
      * @return the active web hook
      * @throws ApiException if there is an error on the call that sets up the web hook.
      */
-    public Watch<Domain> createDomainWatch(String namespace) throws ApiException {
+    public WatchI<Domain> createDomainWatch(String namespace) throws ApiException {
         return FACTORY.createWatch(clientHolder, callParams, Domain.class, new ListDomainsCall(namespace));
     }
 
@@ -175,45 +175,6 @@ public Call apply(ClientHolder clientHolder, CallParams callParams) {
         }
     }
 
-    /**
-     * Creates a web hook object to track changes to custom objects
-     * @param namespace the namespace in which to track custom objects.
-     * @param responseBodyType the type of objects returned in the events
-     * @param group the custom resource group name
-     * @param version the custom resource version
-     * @param plural the custom resource plural name    @return the active web hook
-     * @throws ApiException if there is an error on the call that sets up the web hook.
-     */
-    public <T> Watch<T> createCustomObjectWatch(String namespace, Class<?> responseBodyType, String group, String version, String plural) throws ApiException {
-        return FACTORY.createWatch(clientHolder, callParams, responseBodyType, new ListCustomObjectsInNamespaceCall(namespace, group, version, plural));
-    }
-
-    private class ListCustomObjectsInNamespaceCall implements BiFunction<ClientHolder, CallParams, Call> {
-        private String namespace;
-        private String group;
-        private String version;
-        private String plural;
-
-        ListCustomObjectsInNamespaceCall(String namespace, String group, String version, String plural) {
-            this.namespace = namespace;
-            this.group = group;
-            this.version = version;
-            this.plural = plural;
-        }
-
-        @Override
-        public Call apply(ClientHolder clientHolder, CallParams callParams) {
-            try {
-                return clientHolder.getCustomObjectsApiClient().listNamespacedCustomObjectCall(
-                            group, version, namespace, plural,
-                            callParams.getPretty(), callParams.getLabelSelector(), callParams.getResourceVersion(),
-                            WATCH, null, null);
-            } catch (ApiException e) {
-                throw new UncheckedApiException(e);
-            }
-        }
-    }
-
     /**
      * Sets a value for the fieldSelector parameter for the call that will set up this watch. Defaults to null.
      * @param fieldSelector the desired value
@@ -268,9 +229,9 @@ public WatchBuilder withProgressRequestListener(ProgressRequestBody.ProgressRequ
 
     static class WatchFactoryImpl implements WatchFactory {
         @Override
-        public <T> Watch<T> createWatch(ClientHolder clientHolder, CallParams callParams, Class<?> responseBodyType, BiFunction<ClientHolder, CallParams, Call> function) throws ApiException {
+        public <T> WatchI<T> createWatch(ClientHolder clientHolder, CallParams callParams, Class<?> responseBodyType, BiFunction<ClientHolder, CallParams, Call> function) throws ApiException {
             try {
-                return Watch.createWatch(clientHolder.getApiClient(), function.apply(clientHolder, callParams), getType(responseBodyType));
+                return new WatchImpl<T>(Watch.createWatch(clientHolder.getApiClient(), function.apply(clientHolder, callParams), getType(responseBodyType)));
             } catch (UncheckedApiException e) {
                 throw e.getCause();
             }

src/main/java/oracle/kubernetes/operator/builders/WatchI.java

@@ -0,0 +1,16 @@
+// Copyright 2018 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Licensed under the Universal Permissive License v 1.0 as shown at http://oss.oracle.com/licenses/upl.
+package oracle.kubernetes.operator.builders;
+
+import io.kubernetes.client.util.Watch;
+
+import java.util.Iterator;
+
+/**
+ * An interface that allows test-stubbing of the Kubernetes Watch class.
+ * @param <T> the generic object type
+ */
+public interface WatchI<T>
+        extends Iterable<Watch.Response<T>>, Iterator<Watch.Response<T>>,
+				 java.io.Closeable {
+}

src/main/java/oracle/kubernetes/operator/builders/WatchImpl.java

@@ -0,0 +1,40 @@
+// Copyright 2018 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Licensed under the Universal Permissive License v 1.0 as shown at http://oss.oracle.com/licenses/upl.
+package oracle.kubernetes.operator.builders;
+
+import io.kubernetes.client.util.Watch;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * A pass-through implementation of the Kubernetes Watch class which implements a facade
+ * interface.
+ */
+public class WatchImpl<T> implements WatchI<T> {
+    private Watch<T> impl;
+
+    WatchImpl(Watch<T> impl) {
+        this.impl = impl;
+    }
+
+    @Override
+    public void close() throws IOException {
+        impl.close();
+    }
+
+    @Override
+    public Iterator<Watch.Response<T>> iterator() {
+        return impl.iterator();
+    }
+
+    @Override
+    public boolean hasNext() {
+        return impl.hasNext();
+    }
+
+    @Override
+    public Watch.Response<T> next() {
+        return impl.next();
+    }
+}