Merge commit '794b2abafedf65c2052f2a7a5932155662de8ab1' into rx

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

@@ -9,6 +9,7 @@
 
 import com.microsoft.rest.ServiceCall;
 import com.microsoft.rest.ServiceResponse;
+import com.microsoft.rest.ServiceResponseWithHeaders;
 
 import java.util.List;
 
@@ -45,6 +46,36 @@ public static <E> ServiceCall<List<E>> create(Observable<ServiceResponse<Page<E>
         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.
      *

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

@@ -169,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();

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

@@ -21,6 +21,9 @@
  */
 public abstract class TaskGroupBase<T, U extends TaskItem<T>>
     implements TaskGroup<T, U> {
+    /**
+     * Stores the tasks in this group and their dependency information.
+     */
     private DAGraph<U, DAGNode<U>> dag;
 
     /**
@@ -67,7 +70,7 @@ public void execute() throws Exception {
 
     @Override
     public Observable<T> executeAsync() {
-        return executeReadyTasksAsync();
+        return executeReadyTasksAsync().last();
     }
 
     @Override
@@ -98,6 +101,6 @@ public Observable<T> call(T t) {
                     }));
             nextNode = dag.getNext();
         }
-        return Observable.merge(observables).last();
+        return Observable.merge(observables);
     }
 }

client-runtime/build.gradle

@@ -30,7 +30,6 @@ dependencies {
     compile 'com.squareup.retrofit2:converter-jackson:2.0.2'
     compile 'com.squareup.retrofit2:adapter-rxjava:2.0.2'
     compile 'io.reactivex:rxjava:1.1.8'
-    compile 'io.reactivex:rxjava-computation-expressions:0.21.0'
     compile 'com.fasterxml.jackson.datatype:jackson-datatype-joda:2.7.2'
     compile 'org.apache.commons:commons-lang3:3.4'
     testCompile 'junit:junit:4.12'

client-runtime/pom.xml

@@ -83,6 +83,10 @@
       <groupId>io.reactivex</groupId>
       <artifactId>rxjava</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.squareup.retrofit2</groupId>
+      <artifactId>adapter-rxjava</artifactId>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

pom.xml

@@ -126,6 +126,11 @@
         <artifactId>rxjava</artifactId>
         <version>1.1.8</version>
       </dependency>
+      <dependency>
+        <groupId>com.squareup.retrofit2</groupId>
+        <artifactId>adapter-rxjava</artifactId>
+        <version>2.0.2</version>
+      </dependency>
       <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>