package com.github.davidmoten.rx2.internal.flowable;

import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicLong;

import org.reactivestreams.Publisher;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.LongConsumer;
import io.reactivex.internal.subscriptions.SubscriptionHelper;
import io.reactivex.subjects.ReplaySubject;
import io.reactivex.subjects.Subject;

public final class FlowableFetchPagesByRequest {

    private FlowableFetchPagesByRequest() {
        // prevent instantiation
    }

    public static <T> Flowable<T> create(final BiFunction<? super Long, ? super Long, ? extends Flowable<T>> fetch,
            final long start, final int maxConcurrency) {
        return Flowable.defer(new Callable<Flowable<T>>() {
            @Override
            public Flowable<T> call() throws Exception {
                // need a ReplaySubject because requests can come through before
                // concatEager has established subscriptions to the subject
                final ReplaySubject<Flowable<T>> subject = ReplaySubject.create();
                final AtomicLong position = new AtomicLong(start);
                LongConsumer request = new LongConsumer() {
                    @Override
                    public void accept(final long n) throws Exception {
                        final long pos = position.getAndAdd(n);
                        if (SubscriptionHelper.validate(n)) {
                            Flowable<T> flowable;
                            try {
                                flowable = fetch.apply(pos, n);
                            } catch (Throwable e) {
                                Exceptions.throwIfFatal(e);
                                subject.onError(e);
                                return;
                            }
                            flowable = notifySubjectOfCompletionIfFullRequestNotReturned(flowable, subject, n);
                            subject.onNext(flowable);
                        }
                    }

                };
                return Flowable //
                        .concatEager(subject.serialize() //
                                .toFlowable(BackpressureStrategy.BUFFER), maxConcurrency, 128) //
                        .doOnRequest(request);
            }
        });
    }

    private static <T> Flowable<T> notifySubjectOfCompletionIfFullRequestNotReturned(final Flowable<T> flowable,
            final Subject<Flowable<T>> subject, final long n) {
        return Flowable.defer(new Callable<Publisher<T>>() {
            long count;

            @Override
            public Flowable<T> call() throws Exception {
                return flowable.doOnNext(new Consumer<T>() {

                    @Override
                    public void accept(T t) throws Exception {
                        count++;
                    }
                }).doOnComplete(new Action() {

                    @Override
                    public void run() throws Exception {
                        if (count < n) {
                            subject.onComplete();
                        }
                    }
                });
            }
        });

    }

}

@@ -137,47 +137,8 @@
 g;%0A%0A
-import org.reactivestreams.Publisher;%0A%0A
 impo
@@ -2014,89 +2014,406 @@
     
-flowable = notifySubjectOfCompletionIfFullRequestNotReturned(flowable, subject, n
+// reduce allocations by incorporating the onNext%0A                            // and onComplete actions into the mutable count%0A                            // object%0A                            final Count count = new Count(subject, n);%0A                            flowable = flowable //%0A                                    .doOnNext(count) //%0A                                    .doOnComplete(count
 );%0A 
@@ -2509,25 +2509,24 @@
           %7D%0A
-%0A
             
@@ -2828,405 +2828,261 @@
 tic 
-%3CT%3E Flowable%3CT%3E notifySubjectOfCompletionIfFullRequestNotReturned(final Flowable%3CT%3E flowable,%0A            final Subject%3CFlowable%3CT%3E%3E subject, final long n) %7B%0A        return Flowable.defer(new Callable%3CPublisher%3CT%3E%3E() %7B%0A            long count;%0A%0A            @Override%0A            public Flowable%3CT%3E call() throws Exception %7B%0A                return flowable.doOnNext(new Consumer%3CT%3E() %7B%0A%0A    
+final class Count implements Consumer%3CObject%3E, Action %7B%0A        final Subject%3C?%3E subject;%0A        private final long n;%0A        long count;%0A%0A        Count(Subject%3C?%3E subject, long n) %7B%0A            this.subject = subject;%0A            this.n = n;%0A
         
     
@@ -3069,32 +3069,35 @@
 .n = n;%0A        
+%7D%0A%0A
         @Overrid
@@ -3094,36 +3094,24 @@
   @Override%0A
-            
         publ
@@ -3125,17 +3125,22 @@
  accept(
-T
+Object
  t) thro
@@ -3170,111 +3170,28 @@
     
-            count++;%0A                    %7D%0A                %7D).doOnComplete(new Action() %7B%0A%0A            
+count++;%0A        %7D%0A%0A
     
@@ -3200,36 +3200,24 @@
   @Override%0A
-            
         publ
@@ -3257,36 +3257,24 @@
             
-       
-     
 if (count %3C 
@@ -3274,36 +3274,24 @@
 ount %3C n) %7B%0A
-            
             
@@ -3320,76 +3320,8 @@
 ();%0A
-                        %7D%0A                    %7D%0A                %7D);%0A
     
@@ -3339,19 +3339,16 @@
        %7D
-);%0A
 %0A    %7D%0A%0A

save allocations in fetchPagesByRequest