<commit_before>package com.github.davidmoten.rx.internal.operators;

import java.util.concurrent.atomic.AtomicLong;

import com.github.davidmoten.rx.Actions;

import rx.Observable;
import rx.Observable.Operator;
import rx.Observable.Transformer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.internal.operators.BackpressureUtils;
import rx.schedulers.Schedulers;

public final class TransformerOnBackpressureBufferPassThroughRequests<T> implements Transformer<T, T> {

	private static final TransformerOnBackpressureBufferPassThroughRequests<Object> instance = new TransformerOnBackpressureBufferPassThroughRequests<Object>();

	@SuppressWarnings("unchecked")
	public static final <T> TransformerOnBackpressureBufferPassThroughRequests<T> instance() {
		return (TransformerOnBackpressureBufferPassThroughRequests<T>) instance;
	}

	@Override
	public Observable<T> call(final Observable<T> o) {
		return Observable.defer(new Func0<Observable<T>>() {
			@Override
			public Observable<T> call() {
				final OperatorPassThroughRequest<T> op = new OperatorPassThroughRequest<T>();
				return o.lift(op).onBackpressureBuffer().doOnRequest(new Action1<Long>() {
					@Override
					public void call(Long n) {
						op.requestMore(n);
					}
				});
			}
		});
	}

	/**
	 * Only used with an immediate downstream operator that requests
	 * {@code Long.MAX_VALUE} and should only be subscribed to once (use it
	 * within a {@code defer} block}.
	 *
	 * @param <T>
	 *            stream item type
	 */
	private static final class OperatorPassThroughRequest<T> implements Operator<T, T> {

		private volatile ParentSubscriber<T> parent;
		private volatile long requestedBeforeSubscription = 0;
		private final Object lock = new Object();

		@Override
		public Subscriber<? super T> call(Subscriber<? super T> child) {
			// this method should only be called once for this instance
			// assume child requests MAX_VALUE
			synchronized (lock) {
				parent = new ParentSubscriber<T>(child);
				parent.requestMore(requestedBeforeSubscription);
			}
			child.add(parent);
			return parent;
		}

		public void requestMore(long n) {
			if (parent != null) {
				parent.requestMore(n);
			} else {
				long requestToParent = 0;
				synchronized (lock) {
					if (parent != null) {
						requestToParent = n;
					} else {
						long u = requestedBeforeSubscription + n;
						if (u < 0) {
							requestedBeforeSubscription = Long.MAX_VALUE;
						} else {
							requestedBeforeSubscription += n;
						}
					}
				}
				if (requestToParent > 0) {
					parent.requestMore(requestToParent);
				}
			}
		}

	}

	private static final class ParentSubscriber<T> extends Subscriber<T> {

		private final Subscriber<? super T> child;
		private final AtomicLong emitted = new AtomicLong();
		private final AtomicLong requested = new AtomicLong();

		public ParentSubscriber(Subscriber<? super T> child) {
			this.child = child;
		}

		public void requestMore(long n) {
			long r = requested.get();
			if (r == Long.MAX_VALUE) {
				return;
			} else {
				BackpressureUtils.getAndAddRequest(requested, n);
				r = Math.max(0, requested.get() - emitted.get());
				request(Math.max(n, r));
			}
		}

		@Override
		public void onCompleted() {
			child.onCompleted();
		}

		@Override
		public void onError(Throwable e) {
			child.onError(e);
		}

		@Override
		public void onNext(T t) {
			emitted.incrementAndGet();
			child.onNext(t);
		}

	}

	public static void main(String[] args) throws InterruptedException {
		Observable.range(1, 10000) //
				.doOnRequest(new Action1<Long>() {
					@Override
					public void call(Long n) {
						System.out.println("requested " + n);
					}
				}).doOnUnsubscribe(new Action0() {
					@Override
					public void call() {<commit_msg>buffer pass through dev<commit_after>package com.github.davidmoten.rx.internal.operators;

import java.util.concurrent.atomic.AtomicLong;

import com.github.davidmoten.rx.Actions;

import rx.Observable;
import rx.Observable.Operator;
import rx.Observable.Transformer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.internal.operators.BackpressureUtils;
import rx.schedulers.Schedulers;

public final class TransformerOnBackpressureBufferPassThroughRequests<T> implements Transformer<T, T> {

	private static final TransformerOnBackpressureBufferPassThroughRequests<Object> instance = new TransformerOnBackpressureBufferPassThroughRequests<Object>();

	@SuppressWarnings("unchecked")
	public static final <T> TransformerOnBackpressureBufferPassThroughRequests<T> instance() {
		return (TransformerOnBackpressureBufferPassThroughRequests<T>) instance;
	}

	@Override
	public Observable<T> call(final Observable<T> o) {
		return Observable.defer(new Func0<Observable<T>>() {
			@Override
			public Observable<T> call() {
				final OperatorPassThroughRequest<T> op = new OperatorPassThroughRequest<T>();
				return o.lift(op).onBackpressureBuffer().doOnRequest(new Action1<Long>() {
					@Override
					public void call(Long n) {
						op.requestMore(n);
					}
				});
			}
		});
	}

	/**
	 * Only used with an immediate downstream operator that requests
	 * {@code Long.MAX_VALUE} and should only be subscribed to once (use it
	 * within a {@code defer} block}.
	 *
	 * @param <T>
	 *            stream item type
	 */
	private static final class OperatorPassThroughRequest<T> implements Operator<T, T> {

		private volatile ParentSubscriber<T> parent;
		private volatile long requestedBeforeSubscription = 0;
		private final Object lock = new Object();

		@Override
		public Subscriber<? super T> call(Subscriber<? super T> child) {
			// this method should only be called once for this instance
			// assume child requests MAX_VALUE
			synchronized (lock) {
				parent = new ParentSubscriber<T>(child);
				parent.requestMore(requestedBeforeSubscription);
			}
			child.add(parent);
			return parent;
		}

		public void requestMore(long n) {
			if (parent != null) {
				parent.requestMore(n);
			} else {
				long requestToParent = 0;
				synchronized (lock) {
					if (parent != null) {
						requestToParent = n;
					} else {
						long u = requestedBeforeSubscription + n;
						if (u < 0) {
							requestedBeforeSubscription = Long.MAX_VALUE;
						} else {
							requestedBeforeSubscription += n;
						}
					}
				}
				if (requestToParent > 0) {
					parent.requestMore(requestToParent);
				}
			}
		}

	}

	private static final class ParentSubscriber<T> extends Subscriber<T> {

		private final Subscriber<? super T> child;
		private final AtomicLong emitted = new AtomicLong();
		private final AtomicLong requested = new AtomicLong();

		public ParentSubscriber(Subscriber<? super T> child) {
			this.child = child;
		}

		public void requestMore(long n) {
			long r = requested.get();
			if (r == Long.MAX_VALUE) {
				return;
			} else {
				//TODO use CAS loop
				BackpressureUtils.getAndAddRequest(requested, n);
				r = Math.max(0, requested.get() - emitted.get());
				request(Math.min(n, r));
			}
		}

		@Override
		public void onCompleted() {
			child.onCompleted();
		}

		@Override
		public void onError(Throwable e) {
			child.onError(e);
		}

		@Override
		public void onNext(T t) {
			emitted.incrementAndGet();
			child.onNext(t);
		}

	}

	public static void main(String[] args) throws InterruptedException {
		Observable.range(1, 10000) //
				.doOnRequest(new Action1<Long>() {
					@Override
					public void call(Long n) {
						System.out.println("requested " + n);
					}
				}).doOnUnsubscribe(new Action0() {
					@Override
					public void call() {
{'fraction_non_alphanumeric': 0.10804478000520698, 'fraction_numerical': 0.004686279614683676, 'mean_word_length': 3.6847560975609754, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '382846', 'n_tokens_mistral': 2838, 'n_tokens_neox': 2303, 'n_words': 587}