/**
 *
 * 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 java.util.List;

import rx.Observable;
import rx.Subscriber;
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 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;
    }

    /**
     * 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));
            }
        }
    }
}

@@ -302,16 +302,70 @@
 .List;%0A%0A
+import com.microsoft.rest.ServiceResponseWithHeaders;%0A
 import r
@@ -1647,32 +1647,1225 @@
 iceCall;%0A    %7D%0A%0A
+    public static %3CE, V%3E ServiceCall%3CList%3CE%3E%3E createWithHeaders(Observable%3CServiceResponseWithHeaders%3CPage%3CE%3E, V%3E%3E first, final Func1%3CString, Observable%3CServiceResponseWithHeaders%3CPage%3CE%3E, V%3E%3E%3E next, final ListOperationCallback%3CE%3E callback) %7B%0A        final AzureServiceCall%3CList%3CE%3E%3E serviceCall = new AzureServiceCall%3C%3E();%0A        final PagingSubscriber%3CE%3E subscriber = new PagingSubscriber%3C%3E(serviceCall, new Func1%3CString, Observable%3CServiceResponse%3CPage%3CE%3E%3E%3E%3E() %7B%0A            @Override%0A            public Observable%3CServiceResponse%3CPage%3CE%3E%3E%3E call(String s) %7B%0A                return next.call(s)%0A                        .map(new Func1%3CServiceResponseWithHeaders%3CPage%3CE%3E, V%3E, ServiceResponse%3CPage%3CE%3E%3E%3E() %7B%0A                            @Override%0A                            public ServiceResponse%3CPage%3CE%3E%3E call(ServiceResponseWithHeaders%3CPage%3CE%3E, V%3E pageVServiceResponseWithHeaders) %7B%0A                                return pageVServiceResponseWithHeaders;%0A                            %7D%0A                        %7D);%0A            %7D%0A        %7D, callback);%0A        serviceCall.setSubscription(first%0A                .single()%0A                .subscribe(subscriber));%0A        return serviceCall;%0A    %7D%0A%0A
     /**%0A     * T

Fix pagings with headers