package org.opencloudb.net;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.atomic.AtomicBoolean;

import org.opencloudb.util.TimeUtil;

public class AIOSocketWR extends SocketWR {
	private static final AIOReadHandler aioReadHandler = new AIOReadHandler();
	private static final AIOWriteHandler aioWriteHandler = new AIOWriteHandler();
	private final AsynchronousSocketChannel channel;
	protected final AbstractConnection con;
	private final AtomicBoolean writing = new AtomicBoolean(false);

	public AIOSocketWR(AbstractConnection conn) {
		channel = (AsynchronousSocketChannel) conn.getChannel();
		this.con = conn;
	}

	@Override
	public void asynRead() {

		ByteBuffer theBuffer = con.readBuffer;
		if (theBuffer == null) {
			theBuffer = con.processor.getBufferPool().allocate();
			con.readBuffer = theBuffer;
			channel.read(theBuffer, this, aioReadHandler);

		} else if (theBuffer.hasRemaining()) {
			channel.read(theBuffer, this, aioReadHandler);
		} else {
			throw new java.lang.IllegalArgumentException("full buffer to read ");
		}

	}

	private void asynWrite(ByteBuffer buffer) {
		buffer.flip();
		this.channel.write(buffer, this, aioWriteHandler);
	}

	/**
	 * return true ,means no more data
	 * 
	 * @return
	 */
	private boolean write0() {
		ByteBuffer theBuffer = con.writeBuffer;
		if (theBuffer == null || !theBuffer.hasRemaining()) {// writeFinished,但要区分bufer是否NULL，不NULL，要回收
			if (theBuffer != null) {
				con.recycle(theBuffer);
				con.writeBuffer = null;

			}
			// poll again
			ByteBuffer buffer = con.writeQueue.poll();
			// more data
			if (buffer != null) {
				if (buffer.limit() == 0) {
					con.recycle(buffer);
					con.writeBuffer = null;
					con.close("quit cmd");
					return true;
				} else {
					con.writeBuffer = buffer;
					asynWrite(buffer);
					return false;
				}
			} else {
				// no buffer
				writing.set(false);
				return true;
			}
		} else {
			theBuffer.compact();
			asynWrite(theBuffer);
			return false;
		}

	}

	protected void onWriteFinished(int result) {
		con.netOutBytes += result;
		con.processor.addNetOutBytes(result);
		con.lastWriteTime = TimeUtil.currentTimeMillis();
		boolean noMoreData = this.write0();
		if (noMoreData) {
			this.doNextWriteCheck();
		}

	}

	public void doNextWriteCheck() {
		if (!writing.compareAndSet(false, true)) {
			return;
		}
		boolean noMoreData = this.write0();
		if (noMoreData) {
			if (!con.writeQueue.isEmpty()) {
				this.write0();
			}

		}

	}
}

class AIOWriteHandler implements CompletionHandler<Integer, AIOSocketWR> {

	@Override
	public void completed(final Integer result, final AIOSocketWR wr) {
		try {
			if (result >= 0) {
				wr.onWriteFinished(result);
			} else {
				wr.con.close("write erro " + result);
			}
		} catch (Exception e) {
			AbstractConnection.LOGGER.warn("caught aio process err:", e);
		}

	}

	@Override
	public void failed(Throwable exc, AIOSocketWR wr) {
		wr.con.close("write failed " + exc);
	}

}

class AIOReadHandler implements CompletionHandler<Integer, AIOSocketWR> {
	@Override
	public void completed(final Integer i, final AIOSocketWR wr) {
		// con.getProcessor().getExecutor().execute(new Runnable() {
		// public void run() {
		if (i > 0) {
			try {
				wr.con.onReadData(i);
				wr.con.asynRead();
			} catch (IOException e) {
				wr.con.close("handle err:" + e);
			}
		} else if (i == -1) {
			// System.out.println("read -1 xxxxxxxxx "+con);
			wr.con.close("client closed");
		}
		// }
		// });
	}

	@Override
	public void failed(Throwable exc, AIOSocketWR wr) {
		wr.con.close(exc.toString());

	}
}

@@ -619,16 +619,92 @@
 alse);%0A%0A
+%09protected final AtomicBoolean oneWriteFinished = new AtomicBoolean(true);%0A%0A
 %09public 
@@ -1298,16 +1298,89 @@
 ffer) %7B%0A
+%09%09for (;;) %7B%0A%0A%09%09%09if (oneWriteFinished.compareAndSet(true, false))%0A%09%09%09%7B%0A%09%09
 %09%09buffer
@@ -1384,24 +1384,26 @@
 fer.flip();%0A
+%09%09
 %09%09this.chann
@@ -1443,16 +1443,38 @@
 ndler);%0A
+%09%09%09%09break;%0A%09%09%09%7D%0A%0A%09%09%7D%0A%0A
 %09%7D%0A%0A%09/**
@@ -2930,16 +2930,50 @@
 %09%09try %7B%0A
+%09%09%09wr.oneWriteFinished.set(true);%0A
 %09%09%09if (r
@@ -3233,32 +3233,65 @@
 OSocketWR wr) %7B%0A
+%09%09wr.oneWriteFinished.set(true);%0A
 %09%09wr.con.close(%22

fix AIO的WritePendingException