/*
 * 
 * Copyright 2004 Sun Microsystems, Inc. All Rights Reserved.
 * 
 * The contents of this file are subject to the Sun Community Source
 * License v3.0/Jini Technology Specific Attachment v 1.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://wwws.sun.com/software/jini/licensing/SCSL3_JiniTSA1.html.
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Reference Code is Jini Technology Core Platform code, v2.0. The
 * Developer of the Reference Code is Sun Microsystems, Inc.
 * 
 * Contributor(s): Sun Microsystems, Inc.
 * 
 * The contents of this file comply with the Jini Technology Core
 * Platform Compatibility Kit, v2.0A.
 * 
 * Tester(s): Sun Microsystems, Inc.
 * 
 * Test Platform(s):
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Solaris(TM)SPARC/x86
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Linux (Intel x86)
 * Java 2 SDK, Standard Edition, Version 1.4.2_03 for Windows (Intel Platform)
 * 
 */

package com.sun.jini.jeri.internal.mux;

import com.sun.jini.jeri.internal.runtime.SelectionManager;
import com.sun.jini.logging.Levels;
import com.sun.jini.thread.Executor;
import com.sun.jini.thread.GetThreadPoolAction;
import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.security.AccessController;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * SocketChannelConnectionIO implements the ConnectionIO abstraction for a
 * connection accessible through a java.nio.channels.SocketChannel, and thus
 * supports non-blocking I/O.
 *
 * @author Sun Microsystems, Inc.
 **/
final class SocketChannelConnectionIO extends ConnectionIO {

    private static final int RECEIVE_BUFFER_SIZE = 4096;
    private static final int IOV_MAX = 16; // max writev iovcnt on Solaris...

    /**
     * pool of threads for executing tasks in system thread group:
     * used for I/O (reader and writer) threads
     */
    private static final Executor systemThreadPool =
	(Executor) AccessController.doPrivileged(
	    new GetThreadPoolAction(false));

    /** mux logger */
    private static final Logger logger =
	Logger.getLogger("net.jini.jeri.connection.mux");

    /** selection manager used by this implementation */
    private static final SelectionManager selectionManager;
    static {					// REMIND: share more widely?
	try {
	    selectionManager = new SelectionManager();
	} catch (IOException e) {
	    throw new ExceptionInInitializerError(e);
	}
    }

    /*
     * Work around 4496906: sun.nio.ch.IOVecWrapper.<clinit> requires
     * permission to read the system property "sun.arch.data.model".
     */
    static {
	java.security.AccessController.doPrivileged(
	new java.security.PrivilegedAction() { public Object run() {
	    try {
		Class.forName("sun.nio.ch.IOVecWrapper");
	    } catch (ClassNotFoundException e) {
	    }
	    return null;
	} });
    }

    /** socket channel for underlying connection */
    private final SocketChannel channel;

    private final SelectionManager.Key key;

    /**
     * queue of buffers of data to be sent over connection
     */
    private final LinkedList sendQueue = new LinkedList();

    /**
     * queue of alternating buffers (that are in sendQueue) and IOFuture
     * objects that need to be notified when those buffers are written
     */
    private final LinkedList notifyQueue = new LinkedList();

    /** buffer for reading incoming data from connection */
    private final ByteBuffer inputBuffer =
	ByteBuffer.allocateDirect(RECEIVE_BUFFER_SIZE);	// ready for reading

    private final ByteBuffer[] bufferPair = new ByteBuffer[2];

    private final ByteBuffer[] preallocBufferArray = new ByteBuffer[IOV_MAX];

    /**
     * Creates a new SocketChannelConnectionIO for the connection represented
     * by the supplied SocketChannel.
     */
    SocketChannelConnectionIO(Mux mux, SocketChannel channel)
	throws IOException
    {
	super(mux);
	channel.configureBlocking(false);
	this.channel = channel;
	key = selectionManager.register(channel, new Handler());
    }

    /**
     * Starts processing connection data.
     */
    void start() throws IOException {
	key.renewInterestMask(SelectionKey.OP_READ);
    }

    void asyncSend(ByteBuffer buffer) {
	synchronized (mux.muxLock) {
	    if (mux.muxDown) {
		return;
	    }
	    try {
		if (sendQueue.isEmpty()) {
		    channel.write(buffer);
		}
		if (buffer.hasRemaining()) {
		    sendQueue.addLast(buffer);
		    key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
		}
	    } catch (IOException e) {
		mux.setDown("I/O error writing to mux connection: " +
			    e.toString(), e);
		try {
		    channel.close();
		} catch (IOException ignore) {
		}
	    }
	}
    }

    void asyncSend(ByteBuffer first, ByteBuffer second) {
	synchronized (mux.muxLock) {
	    if (mux.muxDown) {
		return;
	    }
	    try {
		if (sendQueue.isEmpty()) {
		    bufferPair[0] = first;
		    bufferPair[1] = second;
		    channel.write(bufferPair);
		}
		if (!first.hasRemaining()) {
		    if (second.hasRemaining()) {
			sendQueue.addLast(second);
			key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
		    }
		} else {
		    sendQueue.addLast(first);
		    sendQueue.addLast(second);
		    key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
		}
	    } catch (IOException e) {
		mux.setDown("I/O error writing to mux connection: " +
			    e.toString(), e);
		try {
		    channel.close();
		} catch (IOException ignore) {
		}
	    } finally {
		bufferPair[0] = null;
		bufferPair[1] = null;
	    }
	}
    }

    IOFuture futureSend(ByteBuffer first, ByteBuffer second) {
	synchronized (mux.muxLock) {
	    IOFuture future = new IOFuture();
	    if (mux.muxDown) {
		IOException ioe = new IOException(mux.muxDownMessage);
		ioe.initCause(mux.muxDownCause);
		future.done(ioe);
		return future;
	    }
	    try {
		if (sendQueue.isEmpty()) {
		    bufferPair[0] = first;
		    bufferPair[1] = second;
		    channel.write(bufferPair);
		}
		if (!first.hasRemaining()) {
		    if (second.hasRemaining()) {
			sendQueue.addLast(second);
			key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
			notifyQueue.addLast(second);
			notifyQueue.addLast(future);
		    } else {
			future.done();
		    }
		} else {
		    sendQueue.addLast(first);
		    sendQueue.addLast(second);
		    key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
		    notifyQueue.addLast(second);
		    notifyQueue.addLast(future);
		}
	    } catch (IOException e) {
		mux.setDown("I/O error writing to mux connection: " +
			    e.toString(), e);
		future.done(e);
		try {
		    channel.close();
		} catch (IOException ignore) {
		}
	    } finally {
		bufferPair[0] = first;
		bufferPair[1] = second;
	    }
	    return future;
	}
	/*
	 * REMIND: Can/should we implement any sort of
	 * priority inversion avoidance scheme here?
	 */
    }

    private void handleWriteReady() {
	try {
	    synchronized (mux.muxLock) {
//		ByteBuffer[] buffers =
//		    (ByteBuffer[]) sendQueue.toArray(preallocBufferArray);
//		channel.write(buffers);
//		while (!sendQueue.isEmpty()) {
//		    ByteBuffer bb = (ByteBuffer) sendQueue.getFirst();
//		    if (!bb.hasRemaining()) {
//			sendQueue.removeFirst();
//			if (!notifyQueue.isEmpty() &&
//			    bb == notifyQueue.getFirst())
//			{
//			    notifyQueue.removeFirst();
//			    IOFuture future =
//				(IOFuture) notifyQueue.removeFirst();
//			    future.done();
//			}
//		    } else {
//			key.renewInterestMask(SelectionKey.OP_WRITE);	// ###
//			break;
//		    }
//		}

		/*
		 * Work around 4481573: must manually break sequence of
		 * buffers to write into chunks no larger than IOV_MAX.
		 */
	      gatherLoop:
		while (!sendQueue.isEmpty()) {
		    /*
		     * Copy up to the first IOV_MAX buffers of the send queue
		     * into the preallocated ByteBuffer array.
		     */
		    ByteBuffer[] bufs = preallocBufferArray; // IOV_MAX length
		    int len = sendQueue.size();
		    if (len <= bufs.length) {			// optimization
			bufs = (ByteBuffer[]) sendQueue.toArray(bufs);
		    } else {
			Iterator iter = sendQueue.iterator();	// sufficient
			len = 0;
			while (iter.hasNext() && len < bufs.length) {
			    bufs[len++] = (ByteBuffer) iter.next();
			}
		    }
		    channel.write(bufs, 0, len);
		    for (int i = 0; i < len; i++) {
			ByteBuffer bb = bufs[i];
			assert bb == sendQueue.getFirst();
			if (!bb.hasRemaining()) {
			    sendQueue.removeFirst();
			    if (!notifyQueue.isEmpty() &&
				bb == notifyQueue.getFirst())
			    {
				notifyQueue.removeFirst();
				IOFuture future =
				    (IOFuture) notifyQueue.removeFirst();
				future.done();
			    }
			} else {
			    key.renewInterestMask(SelectionKey.OP_WRITE);// ###
			    break gatherLoop;
			}
		    }
		}
	    }
	} catch (IOException e) {
	    logger.log(Levels.HANDLED,
		       "mux writer thread dying, I/O error", e);
	    mux.setDown("I/O error writing to mux connection: " +
			e.toString(), e);
	    drainNotifyQueue();
	    try {
		channel.close();
	    } catch (IOException ignore) {
	    }
	} catch (Throwable t) {
	    logger.log(Level.INFO,
		       "mux writer thread dying, unexpected exception", t);
	    mux.setDown("unexpected exception in mux writer thread: " +
			t.toString(), t);
	    drainNotifyQueue();
	    try {
		channel.close();
	    } catch (IOException ignore) {
	    }
	}
    }

    private void drainNotifyQueue() {
	synchronized (mux.muxLock) {
	    assert mux.muxDown;
	    while (!notifyQueue.isEmpty()) {
		notifyQueue.removeFirst();
		IOFuture future = (IOFuture) notifyQueue.removeFirst();
		IOException ioe = new IOException(mux.muxDownMessage);
		ioe.initCause(mux.muxDownCause);
		future.done(ioe);
	    }
	}
    }

    private void handleReadReady() {
	try {
	    int n = channel.read(inputBuffer);
	    if (n == -1) {
		throw new EOFException();
	    }
	    if (n > 0) {
		mux.processIncomingData(inputBuffer);
	    }
	    assert inputBuffer.hasRemaining();
	    key.renewInterestMask(SelectionKey.OP_READ);
	} catch (ProtocolException e) {
	    IOFuture future = null;
	    synchronized (mux.muxLock) {
		/*
		 * If mux connection is already down, then we probably got
		 * here because of the receipt of a normal protocol-ending
		 * message, like Shutdown or Error, or else something else
		 * went wrong anyway.  Otherwise, a real protocol violation
		 * was detected, so respond with an Error message before
		 * taking down the whole mux connection.
		 */
		if (!mux.muxDown) {
		    logger.log(Levels.HANDLED,
			       "mux reader thread dying, protocol error", e);
		    future = mux.futureSendError(e.getMessage());
		    mux.setDown("protocol violation detected: " +
				e.getMessage(), null);
		} else {
		    logger.log(Level.FINEST,
			       "mux reader thread dying: " + e.getMessage());
		}
	    }
	    if (future != null) {
		try {
		    future.waitUntilDone();
		} catch (IOException ignore) {
		} catch (InterruptedException interrupt) {
		    Thread.currentThread().interrupt();
		}
	    }
	    try {
		channel.close();
	    } catch (IOException ignore) {
	    }
	} catch (IOException e) {
	    logger.log(Levels.HANDLED,
		       "mux reader thread dying, I/O error", e);
	    mux.setDown("I/O error reading from mux connection: " +
			e.toString(), e);
	    try {
		channel.close();
	    } catch (IOException ignore) {
	    }
	} catch (Throwable t) {
	    logger.log(Level.INFO,
		       "mux reader thread dying, unexpected exception", t);
	    mux.setDown("unexpected exception in mux reader thread: " +
			t.toString(), t);
	    try {
		channel.close();
	    } catch (IOException ignore) {
	    }
	}
    }

    private class Handler implements SelectionManager.SelectionHandler {
	public void handleSelection(int readyMask, SelectionManager.Key key) {
	    if ((readyMask & SelectionKey.OP_WRITE) != 0) {
		handleWriteReady();
	    }
	    if ((readyMask & SelectionKey.OP_READ) != 0) {
		handleReadReady();
	    }
	}
    }
}
