User observable for paging

azure-client-runtime/src/main/java/com/microsoft/azure/AzureServiceCall.java

@@ -0,0 +1,69 @@
+/**
+ *
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See License.txt in the project root for license information.
+ *
+ */
+
+package com.microsoft.azure;
+
+import com.microsoft.rest.ServiceCall;
+import com.microsoft.rest.ServiceResponse;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Action1;
+import rx.functions.Func1;
+
+/**
+ * An instance of this class provides access to the underlying REST call invocation.
+ * This class wraps around the Retrofit Call object and allows updates to it in the
+ * progress of a long running operation or a paging operation.
+ *
+ * @param <T> the type of the returning object
+ */
+public class AzureServiceCall<T> extends ServiceCall<T> {
+    private AzureServiceCall() {
+    }
+
+    public static <T extends Page<V>, V> ServiceCall<T> create(Observable<ServiceResponse<T>> first, final Func1<String, Observable<ServiceResponse<T>>> next, final ListOperationCallback<V> callback) {
+        final AzureServiceCall<T> serviceCall = new AzureServiceCall<>();
+        Subscriber<ServiceResponse<T>> subscriber = new Subscriber<ServiceResponse<T>>() {
+            private ServiceResponse<T> lastResponse;
+
+            @Override
+            public void onCompleted() {
+                // do nothing
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                serviceCall.setException(e);
+                if (callback != null) {
+                    callback.failure(e);
+                }
+            }
+
+            @Override
+            public void onNext(ServiceResponse<T> serviceResponse) {
+                lastResponse = serviceResponse;
+                ListOperationCallback.PagingBehavior behavior = ListOperationCallback.PagingBehavior.CONTINUE;
+                if (callback != null) {
+                    behavior = callback.progress(serviceResponse.getBody().getItems());
+                    if (behavior == ListOperationCallback.PagingBehavior.STOP || serviceResponse.getBody().getNextPageLink() == null) {
+                        callback.success();
+                    }
+                }
+                if (behavior == ListOperationCallback.PagingBehavior.STOP || serviceResponse.getBody().getNextPageLink() == null) {
+                    serviceCall.set(lastResponse);
+                } else {
+                    serviceCall.setSubscription(next.call(serviceResponse.getBody().getNextPageLink()).single().subscribe());
+                }
+            }
+        };
+        serviceCall.setSubscription(first
+            .single()
+            .subscribe(subscriber));
+        return serviceCall;
+    }
+}

azure-client-runtime/src/main/java/com/microsoft/azure/ListOperationCallback.java

@@ -8,6 +8,7 @@
 package com.microsoft.azure;
 
 import com.microsoft.rest.ServiceCallback;
+import com.microsoft.rest.ServiceResponse;
 
 import java.util.List;
 
@@ -36,16 +37,14 @@ public ListOperationCallback() {
 
     /**
      * Override this method to handle progressive results.
-     * The user is responsible for returning a {@link PagingBahavior} Enum to indicate
+     * The user is responsible for returning a {@link PagingBehavior} Enum to indicate
      * whether the client should continue loading or stop.
      *
      * @param partial the list of resources from the current request.
      * @return CONTINUE if you want to go on loading, STOP otherwise.
      *
      */
-    public PagingBahavior progress(List<E> partial) {
-        return PagingBahavior.CONTINUE;
-    }
+    public abstract PagingBehavior progress(List<E> partial);
 
     /**
      * Get the list result that stores the accumulated resources loaded from server.
@@ -71,6 +70,13 @@ public void load(List<E> result) {
         }
     }
 
+    @Override
+    public void success(ServiceResponse<List<E>> result) {
+        success();
+    }
+
+    public abstract void success();
+
     /**
      * Get the number of loaded pages.
      *
@@ -83,7 +89,7 @@ public int pageCount() {
     /**
      * An enum to indicate whether the client should continue loading or stop.
      */
-    public enum PagingBahavior {
+    public enum PagingBehavior {
         /**
          * Indicates that the client should continue loading.
          */

client-runtime/src/main/java/com/microsoft/rest/ServiceCall.java

@@ -9,7 +9,10 @@
 
 import com.google.common.util.concurrent.AbstractFuture;
 
+import java.util.List;
+
 import rx.Observable;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action1;
 
@@ -26,7 +29,7 @@ public class ServiceCall<T> extends AbstractFuture<ServiceResponse<T>> {
      */
     private Subscription subscription;
 
-    private ServiceCall() {
+    protected ServiceCall() {
     }
 
     public static <T> ServiceCall<T> create(final Observable<ServiceResponse<T>> observable) {
@@ -95,6 +98,14 @@ public void call(Throwable throwable) {
         return serviceCall;
     }
 
+    public Subscription getSubscription() {
+        return subscription;
+    }
+
+    protected void setSubscription(Subscription subscription) {
+        this.subscription = subscription;
+    }
+
     /**
      * Cancel the Retrofit call if possible. Parameter
      * 'mayInterruptIfRunning is ignored.