package org.handwerkszeug.riak.transport.internal;

import static org.handwerkszeug.riak.util.Validation.notNull;

import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicBoolean;

import org.handwerkszeug.riak.Markers;
import org.handwerkszeug.riak.RiakException;
import org.handwerkszeug.riak._;
import org.handwerkszeug.riak.model.AbstractRiakResponse;
import org.handwerkszeug.riak.model.RiakContentsResponse;
import org.handwerkszeug.riak.model.RiakFuture;
import org.handwerkszeug.riak.nls.Messages;
import org.handwerkszeug.riak.op.RiakResponseHandler;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author taichi
 */
public class CompletionSupport implements ChannelFutureListener {

	static Logger LOG = LoggerFactory.getLogger(CompletionSupport.class);

	final Channel channel;

	final Set<String> inProgress = new ConcurrentSkipListSet<String>();
	final Queue<Command> waitQueue = new ConcurrentLinkedQueue<Command>();
	final AtomicBoolean complete = new AtomicBoolean(false);

	public CompletionSupport(Channel channel) {
		notNull(channel, "channel");
		this.channel = channel;
	}

	@Override
	public void operationComplete(ChannelFuture future) throws Exception {
		if (future.isDone() && this.waitQueue.size() < 1
				&& this.inProgress.size() < 1 && this.complete.get()) {
			LOG.debug(Markers.BOUNDARY, Messages.CloseChannel);
			future.getChannel().close();
		}
	}

	public void complete() {
		this.complete.compareAndSet(false, true);
	}

	public void decrementProgress(String name) {
		this.inProgress.remove(name);
	}

	public CountDownRiakFuture newRiakFuture(String name) {
		ChannelPipeline pipeline = this.channel.getPipeline();
		return new CountDownRiakFuture(name, pipeline);
	}

	public <T> RiakFuture handle(String name, Object send,
			RiakResponseHandler<T> users, MessageHandler handler) {
		CountDownRiakFuture future = newRiakFuture(name);
		ChannelHandler ch = new DefaultCompletionChannelHandler<T>(this, name,
				users, handler, future);
		return handle(name, send, users, ch, future);
	}

	public <T> RiakFuture handle(String name, Object send,
			RiakResponseHandler<T> users, ChannelHandler handler,
			RiakFuture future) {
		ChannelPipeline pipeline = this.channel.getPipeline();

		Command cmd = new Command(this.channel, send, name, handler);
		try {
			if (this.inProgress.add(cmd.name)) {
				cmd.execute();
			} else {
				if (LOG.isDebugEnabled()) {
					LOG.debug(Markers.LIFECYCLE, Messages.Queue, name);
				}
				this.waitQueue.add(cmd);
			}
			return future;
		} catch (Exception e) {
			complete(name, pipeline);
			throw new RiakException(e);
		} catch (Error e) {
			complete(name, pipeline);
			throw e;
		}
	}

	protected void complete(String name, ChannelPipeline pipeline) {
		pipeline.remove(name);
		complete();
		this.channel.close();
	}

	protected void invokeNext() {
		for (Iterator<Command> i = this.waitQueue.iterator(); i.hasNext();) {
			Command cmd = i.next();
			if (this.inProgress.add(cmd.name)) {
				i.remove();
				cmd.execute();
			}
		}
	}

	public RiakContentsResponse<_> newResponse() {
		return newResponse(_._);
	}

	public <T> RiakContentsResponse<T> newResponse(final T contents) {
		return new AbstractCompletionRiakResponse<T>() {
			@Override
			public T getContents() {
				return contents;
			}
		};
	}

	public abstract class AbstractCompletionRiakResponse<T> extends
			AbstractRiakResponse implements RiakContentsResponse<T> {
		@Override
		public void operationComplete() {
			complete();
		}
	}
}

@@ -1642,108 +1642,456 @@
 if (
-future.isDone() && this.waitQueue.size() %3C 1%0D%0A%09%09%09%09&& this.inProgress.size() %3C 1 && this.complete.get
+LOG.isDebugEnabled()) %7B%0D%0A%09%09%09LOG.debug(Markers.DETAIL,%0D%0A%09%09%09%09%09%22done:%7B%7D queue:%7B%7D inprocess:%7B%7D complete:%7B%7D%22, new Object%5B%5D %7B%0D%0A%09%09%09%09%09%09%09future.isDone(), this.waitQueue.size(),%0D%0A%09%09%09%09%09%09%09this.inProgress.size(), this.complete.get() %7D);%0D%0A%09%09%7D%0D%0A%09%09if (future.isDone() && this.waitQueue.size() %3C 1%0D%0A%09%09%09%09&& this.inProgress.size() %3C 1 && this.complete.get()) %7B%0D%0A%09%09%09close(future.getChannel());%0D%0A%09%09%7D%0D%0A%09%7D%0D%0A%0D%0A%09protected void close(Channel channel) %7B%0D%0A%09%09if (channel.isOpen
 ()) 
@@ -2152,35 +2152,23 @@
 l);%0D%0A%09%09%09
-future.getC
+c
 hannel
-()
 .close()
@@ -3255,17 +3255,14 @@
 ers.
-LIFECYCLE
+DETAIL
 , Me
@@ -3617,24 +3617,30 @@
 plete();%0D%0A%09%09
+close(
 this.channel
@@ -3635,31 +3635,24 @@
 this.channel
-.close(
 );%0D%0A%09%7D%0D%0A%0D%0A%09p
@@ -3856,24 +3856,36 @@
 execute();%0D%0A
+%09%09%09%09break;%0D%0A
 %09%09%09%7D%0D%0A%09%09%7D%0D%0A%09

fix design problem.