/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package reactor.ipc.netty.http.multipart;

import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.MultiProducer;
import reactor.core.Producer;
import reactor.core.Receiver;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.core.Trackable;
import reactor.core.publisher.Operators;
import reactor.core.Exceptions;
import reactor.util.concurrent.QueueSupplier;
import reactor.ipc.netty.common.ByteBufEncodedFlux;

/**
 * @author Ben Hale
 * @author Stephane Maldini
 */

final class MultipartParser
		implements Subscriber<MultipartTokenizer.Token>, Subscription, Runnable, Producer,
		           MultiProducer, Receiver, Trackable {

	final Subscriber<? super ByteBufEncodedFlux> actual;
	final ByteBufAllocator                       alloc;

	volatile int wip;
	@SuppressWarnings("rawtypes")
	static final AtomicIntegerFieldUpdater<MultipartParser> WIP =
			AtomicIntegerFieldUpdater.newUpdater(MultipartParser.class, "wip");

	volatile int once;
	@SuppressWarnings("rawtypes")
	static final AtomicIntegerFieldUpdater<MultipartParser> ONCE =
			AtomicIntegerFieldUpdater.newUpdater(MultipartParser.class, "once");

	Subscription s;

	UnicastProcessor<ByteBuf> window;

	boolean done;

	MultipartParser(Subscriber<? super ByteBufEncodedFlux> actual, ByteBufAllocator alloc) {
		this.actual = actual;
		this.wip = 1;
		this.alloc = alloc;
	}

	@Override
	public void onSubscribe(Subscription s) {
		if (Operators.validate(this.s, s)) {
			this.s = s;
			actual.onSubscribe(this);
		}
	}

	@Override
	public void onNext(MultipartTokenizer.Token token) {
		if (done) {
			Operators.onNextDropped(token);
			return;
		}

		UnicastProcessor<ByteBuf> w = window;

		switch (token.getKind()) {
			case BODY:
				if (window != null) {
					window.onNext(token.getByteBuf().retain());
					return;
				}
				s.cancel();
				actual.onError(new IllegalStateException("Body received before " +
						"delimiter"));
				break;
			case CLOSE_DELIMITER:
				onComplete();
				break;
			case DELIMITER:
				if (window != null) {
					window = null;
					w.onComplete();
				}

				WIP.getAndIncrement(this);

				w = UnicastProcessor.create(QueueSupplier.<ByteBuf>unbounded().get(), this);

				window = w;

				actual.onNext(ByteBufEncodedFlux.encoded(
						w.flatMap(b -> Flux.using(() -> b, Flux::just,
								ByteBuf::release)), alloc));
		}
	}

	@Override
	public void onError(Throwable t) {
		if (done) {
			Operators.onErrorDropped(t);
			return;
		}
		UnicastProcessor<ByteBuf> w = window;
		if (w != null) {
			window = null;
			w.onError(t);
		}

		actual.onError(t);
	}

	@Override
	public void onComplete() {
		if (done) {
			return;
		}

		UnicastProcessor<ByteBuf> w = window;
		if (w != null) {
			window = null;
			w.onComplete();
		}

		actual.onComplete();
	}

	@Override
	public void request(long n) {
		s.request(n);
	}

	@Override
	public void cancel() {
		if (ONCE.compareAndSet(this, 0, 1)) {
			run();
		}
	}

	@Override
	public void run() {
		if (WIP.decrementAndGet(this) == 0) {
			s.cancel();
		}
	}

	@Override
	public Object downstream() {
		return actual;
	}

	@Override
	public boolean isStarted() {
		return s != null && !done;
	}

	@Override
	public boolean isTerminated() {
		return done;
	}

	@Override
	public Object upstream() {
		return s;
	}

	@Override
	public Iterator<?> downstreams() {
		return Arrays.asList(window)
		             .iterator();
	}

	@Override
	public long downstreamCount() {
		return window != null ? 1L : 0L;
	}
}
@@ -942,16 +942,50 @@
 iption;%0A
+import reactor.core.Cancellation;%0A
 import r
@@ -1512,16 +1512,20 @@
 on, 
-Runnable
+Cancellation
 , Pr
@@ -3853,19 +3853,23 @@
 )) %7B%0A%09%09%09
-run
+dispose
 ();%0A%09%09%7D%0A
@@ -3900,11 +3900,15 @@
 oid 
-run
+dispose
 () %7B

fix to latest snapshot