Merge pull request Azure#51 from jianghaolu/09062016

Sync with SDK repo

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

@@ -0,0 +1,125 @@
+/**
+ *
+ * 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 com.microsoft.rest.ServiceResponseWithHeaders;
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func1;
+
+import java.util.List;
+
+/**
+ * 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 final class AzureServiceCall<T> extends ServiceCall<T> {
+    private AzureServiceCall() {
+    }
+
+    /**
+     * Creates a ServiceCall from a paging operation.
+     *
+     * @param first the observable to the first page
+     * @param next the observable to poll subsequent pages
+     * @param callback the client-side callback
+     * @param <E> the element type
+     * @return the future based ServiceCall
+     */
+    public static <E> ServiceCall<List<E>> create(Observable<ServiceResponse<Page<E>>> first, final Func1<String, Observable<ServiceResponse<Page<E>>>> next, final ListOperationCallback<E> callback) {
+        final AzureServiceCall<List<E>> serviceCall = new AzureServiceCall<>();
+        final PagingSubscriber<E> subscriber = new PagingSubscriber<>(serviceCall, next, callback);
+        serviceCall.setSubscription(first
+            .single()
+            .subscribe(subscriber));
+        return serviceCall;
+    }
+
+    /**
+     * Creates a ServiceCall from a paging operation that returns a header response.
+     *
+     * @param first the observable to the first page
+     * @param next the observable to poll subsequent pages
+     * @param callback the client-side callback
+     * @param <E> the element type
+     * @param <V> the header object type
+     * @return the future based ServiceCall
+     */
+    public static <E, V> ServiceCall<List<E>> createWithHeaders(Observable<ServiceResponseWithHeaders<Page<E>, V>> first, final Func1<String, Observable<ServiceResponseWithHeaders<Page<E>, V>>> next, final ListOperationCallback<E> callback) {
+        final AzureServiceCall<List<E>> serviceCall = new AzureServiceCall<>();
+        final PagingSubscriber<E> subscriber = new PagingSubscriber<>(serviceCall, new Func1<String, Observable<ServiceResponse<Page<E>>>>() {
+            @Override
+            public Observable<ServiceResponse<Page<E>>> call(String s) {
+                return next.call(s)
+                        .map(new Func1<ServiceResponseWithHeaders<Page<E>, V>, ServiceResponse<Page<E>>>() {
+                            @Override
+                            public ServiceResponse<Page<E>> call(ServiceResponseWithHeaders<Page<E>, V> pageVServiceResponseWithHeaders) {
+                                return pageVServiceResponseWithHeaders;
+                            }
+                        });
+            }
+        }, callback);
+        serviceCall.setSubscription(first
+                .single()
+                .subscribe(subscriber));
+        return serviceCall;
+    }
+
+    /**
+     * The subscriber that handles user callback and automatically subscribes to the next page.
+     *
+     * @param <E> the element type
+     */
+    private static class PagingSubscriber<E> extends Subscriber<ServiceResponse<Page<E>>> {
+        private AzureServiceCall<List<E>> serviceCall;
+        private Func1<String, Observable<ServiceResponse<Page<E>>>> next;
+        private ListOperationCallback<E> callback;
+        private ServiceResponse<Page<E>> lastResponse;
+
+        PagingSubscriber(final AzureServiceCall<List<E>> serviceCall, final Func1<String, Observable<ServiceResponse<Page<E>>>> next, final ListOperationCallback<E> callback) {
+            this.serviceCall = serviceCall;
+            this.next = next;
+            this.callback = callback;
+        }
+
+        @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<Page<E>> 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(new ServiceResponse<>(lastResponse.getBody().getItems(), lastResponse.getResponse()));
+            } else {
+                serviceCall.setSubscription(next.call(serviceResponse.getBody().getNextPageLink()).single().subscribe(this));
+            }
+        }
+    }
+}

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,16 @@ public void load(List<E> result) {
         }
     }
 
+    @Override
+    public void success(ServiceResponse<List<E>> result) {
+        success();
+    }
+
+    /**
+     * Override this method to handle successful REST call results.
+     */
+    public abstract void success();
+
     /**
      * Get the number of loaded pages.
      *
@@ -83,7 +92,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.
          */

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

@@ -9,8 +9,6 @@
 
 import com.microsoft.rest.RestException;
 
-import javax.xml.bind.DataBindingException;
-import javax.xml.ws.WebServiceException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -19,6 +17,8 @@
 import java.util.ListIterator;
 import java.util.NoSuchElementException;
 
+import javax.xml.bind.DataBindingException;
+
 /**
  * Defines a list response from a paging operation. The pages are
  * lazy initialized when an instance of this class is iterated.
@@ -47,7 +47,10 @@ public PagedList() {
      */
     public PagedList(Page<E> page) {
         this();
-        items.addAll(page.getItems());
+        List<E> retrievedItems = page.getItems();
+        if (retrievedItems != null && retrievedItems.size() != 0) {
+            items.addAll(retrievedItems);
+        }
         nextPageLink = page.getNextPageLink();
         currentPage = page;
     }
@@ -81,8 +84,6 @@ public void loadNextPage() {
             this.nextPageLink = nextPage.getNextPageLink();
             this.items.addAll(nextPage.getItems());
             this.currentPage = nextPage;
-        } catch (RestException e) {
-            throw new WebServiceException(e.toString(), e);
         } catch (IOException e) {
             throw new DataBindingException(e.getMessage(), e);
         }
@@ -140,14 +141,17 @@ public boolean hasNext() {
         public E next() {
             if (!itemsListItr.hasNext()) {
                 if (!hasNextPage()) {
-                    throw new NoSuchElementException();
+                    throw new NoSuchElementException();                        
                 } else {
                     int size = items.size();
                     loadNextPage();
                     itemsListItr = items.listIterator(size);
                 }
             }
-            return itemsListItr.next();
+            if (itemsListItr.hasNext()) {
+                return itemsListItr.next();
+            }
+            return null;
         }
 
         @Override

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

@@ -100,7 +100,7 @@ public void updateFromResponseOnPutPatch(Response<ResponseBody> response) throws
         }
 
         if (responseContent == null || responseContent.isEmpty()) {
-            CloudException exception = new CloudException("no body");
+            CloudException exception = new CloudException("polling response does not contain a valid body");
             exception.setResponse(response);
             throw exception;
         }

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

@@ -20,6 +20,7 @@
 import okhttp3.OkHttpClient;
 import okhttp3.logging.HttpLoggingInterceptor;
 import retrofit2.Retrofit;
+import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory;
 
 import java.lang.reflect.Field;
 import java.net.CookieManager;
@@ -168,6 +169,7 @@ public Builder(OkHttpClient.Builder httpClientBuilder, Retrofit.Builder retrofit
             // Set up OkHttp client
             this.httpClientBuilder = httpClientBuilder
                     .cookieJar(new JavaNetCookieJar(cookieManager))
+                    .readTimeout(30, TimeUnit.SECONDS)
                     .addInterceptor(userAgentInterceptor);
             this.retrofitBuilder = retrofitBuilder;
             this.buildable = new Buildable();
@@ -338,6 +340,7 @@ public RestClient build() {
                                 .baseUrl(baseUrl)
                                 .client(httpClient)
                                 .addConverterFactory(mapperAdapter.getConverterFactory())
+                                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                                 .build(),
                         credentials,
                         customHeadersInterceptor,

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

@@ -7,8 +7,7 @@
 
 package com.microsoft.azure;
 
-import com.microsoft.rest.ServiceCall;
-import com.microsoft.rest.ServiceCallback;
+import rx.Observable;
 
 /**
  * Represents a group of related tasks.
@@ -48,22 +47,12 @@ public interface TaskGroup<T, U extends TaskItem<T>> {
      */
     void prepare();
 
-    /**
-     * Executes the tasks in the group.
-     * <p>
-     * the order of execution of tasks ensure that a task gets selected for execution only after
-     * the execution of all the tasks it depends on
-     * @throws Exception the exception
-     */
-    void execute() throws Exception;
-
     /**
      * Executes the tasks in the group asynchronously.
      *
-     * @param callback the callback to call on failure or success
      * @return the handle to the REST call
      */
-    ServiceCall executeAsync(ServiceCallback<T> callback);
+    Observable<T> executeAsync();
 
     /**
      * Gets the result of execution of a task in the group.