package au.gov.amsa.streams;

import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.regex.Pattern;

import rx.Observable.Operator;
import rx.Producer;
import rx.Subscriber;
import rx.internal.operators.NotificationLite;

public class StringSplitOperator implements Operator<String, String> {

	private final Pattern pattern;

	public StringSplitOperator(Pattern pattern) {
		this.pattern = pattern;
	}

	@Override
	public Subscriber<? super String> call(Subscriber<? super String> child) {
		final ParentSubscriber parent = new ParentSubscriber(child, pattern);

		child.setProducer(new Producer() {
			@Override
			public void request(long n) {
				parent.requestMore(n);
			}
		});

		return parent;
	}

	private static final class ParentSubscriber extends Subscriber<String> {

		// utility object for reactive events
		private final NotificationLite<String> on = NotificationLite.instance();

		// report events to this downstream subscriber
		private final Subscriber<? super String> child;

		// fast path or not
		private boolean requestAll = false;

		// number of items being waited for (not used if fast path selected)
		private volatile long expected = 0;

		// updater object for the expected field above
		private final AtomicLongFieldUpdater<ParentSubscriber> EXPECTED = AtomicLongFieldUpdater
				.newUpdater(ParentSubscriber.class, "expected");

		// queue to hold messages till they are requested
		private final Deque<Object> queue = new LinkedList<Object>();

		// pattern to split upon
		private final Pattern pattern;

		private ParentSubscriber(Subscriber<? super String> child,
				Pattern pattern) {
			this.child = child;
			this.pattern = pattern;
		}

		private void requestMore(long n) {
			// this method may be run concurrent with any of the event methods
			// below so watch out for thread safety

			if (requestAll)
				// ignore request if all items have already been requested
				return;
			if (n == Long.MAX_VALUE)
				requestAll = true;
			else
				EXPECTED.addAndGet(this, n);
			request(n);
		}

		@Override
		public void onCompleted() {
			if (requestAll)
				child.onCompleted();
			else {
				queue.add(on.completed());
				drainQueue();
			}
		}

		@Override
		public void onError(Throwable e) {
			if (requestAll)
				child.onError(e);
			else {
				queue.add(on.error(e));
				drainQueue();
			}
		}

		@Override
		public void onNext(String s) {
			String[] parts = pattern.split(s, -1);
			if (requestAll)
				for (String part : parts)
					child.onNext(part);
			else {
				for (String part : parts)
					queue.add(on.next(part));
				drainQueue();
			}
		}

		private void drainQueue() {
			while (true) {
				Object item = queue.peek();
				if (item == null)
					break;
				else if (on.isCompleted(item) || on.isError(item)) {
					on.accept(child, queue.poll());
					break;
				} else {
					if (expected > 0) {
						// expected won't be Long.MAX_VALUE so can safely
						// decrement
						EXPECTED.decrementAndGet(this);
						on.accept(child, queue.poll());
					} else {
						break;
					}
				}
			}
		}
	}

}

@@ -289,16 +289,85 @@
 nLite;%0A%0A
+/**%0A * Splits strings by a regex pattern. Supports backpressure.%0A */%0A
 public c
@@ -435,39 +435,137 @@
 %7B%0A%0A%09
-private final Pattern pattern;%0A
+/**%0A%09 * Pattern to split the strings by.%0A%09 */%0A%09private final Pattern pattern;%0A%0A%09/**%0A%09 * Constructor.%0A%09 * %0A%09 * @param pattern%0A%09 */
 %0A%09pu

add javadoc