package com.github.davidmoten.rtree;

import java.util.concurrent.atomic.AtomicLong;

import com.github.davidmoten.guavamini.annotations.VisibleForTesting;
import com.github.davidmoten.rtree.geometry.Geometry;
import com.github.davidmoten.rtree.internal.util.ImmutableStack;
import com.github.davidmoten.rx.util.BackpressureUtils;

import rx.Observable.OnSubscribe;
import rx.Producer;
import rx.Subscriber;
import rx.functions.Func1;

final class OnSubscribeSearch<T, S extends Geometry> implements OnSubscribe<Entry<T, S>> {

    private final Node<T, S> node;
    private final Func1<? super Geometry, Boolean> condition;

    OnSubscribeSearch(Node<T, S> node, Func1<? super Geometry, Boolean> condition) {
        this.node = node;
        this.condition = condition;
    }

    @Override
    public void call(Subscriber<? super Entry<T, S>> subscriber) {
        subscriber.setProducer(new SearchProducer<T, S>(node, condition, subscriber));
    }

    @VisibleForTesting
    static class SearchProducer<T, S extends Geometry> implements Producer {

        private final Subscriber<? super Entry<T, S>> subscriber;
        private final Node<T, S> node;
        private final Func1<? super Geometry, Boolean> condition;
        private volatile ImmutableStack<NodePosition<T, S>> stack;
        private final AtomicLong requested = new AtomicLong(0);

        SearchProducer(Node<T, S> node, Func1<? super Geometry, Boolean> condition,
                Subscriber<? super Entry<T, S>> subscriber) {
            this.node = node;
            this.condition = condition;
            this.subscriber = subscriber;
            stack = ImmutableStack.create(new NodePosition<T, S>(node, 0));
        }

        @Override
        public void request(long n) {
            try {
                if (n <= 0 || requested.get() == Long.MAX_VALUE)
                    // none requested or already started with fast path
                    return;
                else if (n == Long.MAX_VALUE && requested.compareAndSet(0, Long.MAX_VALUE)) {
                    // fast path
                    requestAll();
                } else
                    requestSome(n);
            } catch (RuntimeException e) {
                subscriber.onError(e);
            }
        }

        private void requestAll() {
            node.searchWithoutBackpressure(condition, subscriber);
            if (!subscriber.isUnsubscribed())
                subscriber.onCompleted();
        }

        private void requestSome(long n) {
            // back pressure path
            // this algorithm copied roughly from
            // rxjava-core/OnSubscribeFromIterable.java

            // rxjava used AtomicLongFieldUpdater instead of AtomicLong
            // but benchmarks showed no benefit here so reverted to AtomicLong
            long previousCount = BackpressureUtils.getAndAddRequest(requested, n);
            if (previousCount == 0) {
                // don't touch stack every time during the loop because
                // is a volatile and every write forces a thread memory
                // cache flush
                ImmutableStack<NodePosition<T, S>> st = stack;
                while (true) {
                    // minimize atomic reads by assigning to a variable here
                    long r = requested.get();
                    st = Backpressure.search(condition, subscriber, st, r);
                    if (st.isEmpty()) {
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onCompleted();
                            break;
                        } else
                            break;
                    } else if (requested.addAndGet(-r) == 0)
                        break;
                }
                stack = st;
            }
        }
    }

}

@@ -3425,24 +3425,162 @@
 sEmpty()) %7B%0A
+                        // release some state for gc (although empty stack so not very significant)%0A                        stack = null;%0A
             
@@ -3701,33 +3701,56 @@
             
-    break
+%7D%0A                        return
 ;%0A          
@@ -3755,34 +3755,32 @@
             
- 
-   
 %7D else
+ %7B
 %0A           
@@ -3788,33 +3788,34 @@
             
-    break
+stack = st
 ;%0A          
@@ -3824,22 +3824,19 @@
         
-%7D else
+   
  if (req
@@ -3890,15 +3890,24 @@
     
+ 
-break;%0A
+   return;%0A    
     
@@ -3932,35 +3932,26 @@
             
-stack = st;
+%7D%0A
 %0A           

fix backpressure race condition and release volatile state (null it) before calling onComplete in OnSubscribeSearch