package com.github.davidmoten.rtree;

import java.util.concurrent.atomic.AtomicLongFieldUpdater;

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

import com.github.davidmoten.rtree.geometry.Geometry;
import com.github.davidmoten.util.ImmutableStack;

final class OnSubscribeSearch<T> implements OnSubscribe<Entry<T>> {

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

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

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

    private static class SearchProducer<T> implements Producer {

        private final Subscriber<? super Entry<T>> subscriber;
        private final Node<T> node;
        private final Func1<? super Geometry, Boolean> condition;
        private volatile ImmutableStack<NodePosition<T>> stack;

        private volatile long requested = 0;
        @SuppressWarnings("rawtypes")
        private static final AtomicLongFieldUpdater<SearchProducer> REQUESTED_UPDATER = AtomicLongFieldUpdater
                .newUpdater(SearchProducer.class, "requested");

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

        @Override
        public void request(long n) {
            try {
                if (REQUESTED_UPDATER.get(this) == Long.MAX_VALUE)
                    // already started with fast path
                    return;
                else if (n == Long.MAX_VALUE) {
                    // fast path
                    requestAll();
                } else {
                    requestSome(n);
                }
            } catch (RuntimeException e) {
                subscriber.onError(e);
            }
        }

        private void requestAll() {
            REQUESTED_UPDATER.set(this, Long.MAX_VALUE);
            node.search(condition, subscriber);
            if (!subscriber.isUnsubscribed())
                subscriber.onCompleted();
        }

        private void requestSome(long n) {
            // back pressure path
            // this algorithm copied roughly from
            // rxjava-core/OnSubscribeFromIterable.java
            long previousCount = REQUESTED_UPDATER.getAndAdd(this, n);
            if (previousCount == 0) {
                while (true) {
                    long r = requested;
                    long numToEmit = r;

                    stack = Backpressure.search(condition, subscriber, stack, numToEmit);
                    if (stack.isEmpty()) {
                        if (!subscriber.isUnsubscribed())
                            subscriber.onCompleted();
                        else
                            return;
                    } else if (REQUESTED_UPDATER.addAndGet(this, -r) == 0)
                        return;
                }
            }
        }
    }

}

@@ -80,22 +80,10 @@
 Long
-FieldUpdater
 ;%0A
+
 %0Aimp
@@ -1067,17 +1067,16 @@
  stack;%0A
-%0A
         
@@ -1087,247 +1087,53 @@
 ate 
-volatile long requested = 0;%0A        @SuppressWarnings(%22rawtypes%22)%0A        private static final AtomicLongFieldUpdater%3CSearchProducer%3E REQUESTED_UPDATER = AtomicLongFieldUpdater%0A                .newUpdater(SearchProducer.class, %22requested%22
+final AtomicLong requested = new AtomicLong(0
 );%0A%0A
@@ -1566,34 +1566,22 @@
 if (
-REQUESTED_UPDATER
+requested
 .get(
-this
 ) ==
@@ -2032,36 +2032,22 @@
     
-REQUESTED_UPDATER.set(this, 
+requested.set(
 Long
@@ -2422,33 +2422,25 @@
 Count = 
-REQUESTED_UPDATER
+requested
 .getAndA
@@ -2442,22 +2442,16 @@
 tAndAdd(
-this, 
 n);%0A    
@@ -2553,16 +2553,22 @@
 equested
+.get()
 ;%0A      
@@ -2947,25 +2947,17 @@
 if (
-REQUESTED_UPDATER
+requested
 .add
@@ -2967,14 +2967,8 @@
 Get(
-this, 
 -r) 

revert to AtomicLong because benchmarks not improved with use of AtomicLongFieldUpdater