/*
 * Copyright (c) 2010 Remko Tronçon
 * All rights reserved.
 */
/*
 * Copyright (c) 2010-2012, Isode Limited, London, England.
 * All rights reserved.
 */
package com.isode.stroke.network;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import com.isode.stroke.base.ByteArray;
import com.isode.stroke.eventloop.Event.Callback;
import com.isode.stroke.eventloop.EventLoop;
import com.isode.stroke.eventloop.EventOwner;

public class JavaConnection extends Connection implements EventOwner {

    private class Worker implements Runnable {

        private final HostAddressPort address_;
        private final List<ByteArray> writeBuffer_ = Collections.synchronizedList(new ArrayList<ByteArray>());

        public Worker(HostAddressPort address) {
            address_ = address;
        }

        public void run() {
            try {
                socketChannel_ = SocketChannel.open(
                        new InetSocketAddress(address_.getAddress().getInetAddress(),address_.getPort()));
                
                /* By default, SocketChannels start off in blocking mode, which
                 * isn't what we want
                 */
                socketChannel_.configureBlocking(false);
            } catch (IOException ex) {
                handleConnected(true);
                return;
            }
            handleConnected(false);
            while (!disconnecting_) {
                while (!writeBuffer_.isEmpty()) {
                    ByteArray data = writeBuffer_.get(0);
                    ByteBuffer byteBuffer = ByteBuffer.wrap(data.getData());
                    try {
                        /* Because the SocketChannel is non-blocking, we have to
                         * be prepared to cope with the write operation not
                         * consuming all of the data
                         */
                        boolean finishedWriting = false;
                        while (!finishedWriting && !disconnecting_) {                     
                            socketChannel_.write(byteBuffer);
                            finishedWriting = (byteBuffer.remaining() == 0);
                            if (!finishedWriting) {
                                try {
                                    /* Give the output buffer a chance to empty */
                                    Thread.sleep(100);
                                }
                                catch (InterruptedException e) {
                                    /* Perhaps someone has set disconnecting_ */
                                }
                            }
                        }
                    } catch (IOException ex) {
                        disconnecting_ = true;
                        handleDisconnected(Error.WriteError);                    
                    }
                    writeBuffer_.remove(0);
                }

                ByteArray data = new ByteArray();
                try {
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                   
                    int count = socketChannel_.read(byteBuffer);
                    while (count > 0) {
                        byteBuffer.flip();
                        byte[] result = new byte[byteBuffer.remaining()];
                        byteBuffer.get(result);
                        byteBuffer.compact();
                        for (int i=0; i<result.length; i++) {
                            data.append(result[i]);
                        }
                        count = socketChannel_.read(byteBuffer);
                   }
                   if (count == -1) {
                       /* socketChannel input has reached "end-of-stream", which
                        * we regard as meaning that the socket has been closed 
                        */
                       throw new IOException("socketChannel_.read returned -1");
                   }
                } catch (IOException ex) {
                    handleDisconnected(Error.ReadError);
                    return;
                }
                
                if (!data.isEmpty()) {
                    handleDataRead(data);
                }

                try {
                    Thread.sleep(100);
                } catch (InterruptedException ex) {
                    /* We've been woken up, probably to force us to do something.*/
                }
            }
            try {
                socketChannel_.close();
            } catch (IOException ex) {
                /* Do we need to return an error if we're already trying to close? */
            }
        }

        private void handleConnected(final boolean error) {
            eventLoop_.postEvent(new Callback() {
                public void run() {
                    onConnectFinished.emit(error);
                }
            });
        }

        private void handleDisconnected(final Error error) {
            eventLoop_.postEvent(new Callback() {
                public void run() {
                    onDisconnected.emit(error);
                }
            });
        }

        private void handleDataRead(final ByteArray data) {
            eventLoop_.postEvent(new Callback() {
                public void run() {
                    onDataRead.emit(data);
                }
            });
        }

        public void write(ByteArray data) {
            writeBuffer_.add(data);
        }
    }

    private JavaConnection(EventLoop eventLoop) {
        eventLoop_ = eventLoop;
    }

    public static JavaConnection create(EventLoop eventLoop) {
        return new JavaConnection(eventLoop);
    }

    @Override
    public void listen() {
        //TODO: needed for server, not for client.
        throw new UnsupportedOperationException("Not supported yet.");
    }

    @Override
    public void connect(HostAddressPort address) {
        worker_ = new Worker(address);
        Thread workerThread = new Thread(worker_);
        workerThread.setDaemon(true);
        workerThread.start();
    }

    @Override
    public void disconnect() {
        disconnecting_ = true;
    }

    @Override
    public void write(ByteArray data) {
        worker_.writeBuffer_.add(data);
    }

    @Override
    public HostAddressPort getLocalAddress() {
        if (socketChannel_ == null) {
            return null;
        }
        Socket socket = socketChannel_.socket();
        if (socket == null) {
            return null;
        }
        return new HostAddressPort(new HostAddress(socket.getLocalAddress()), socket.getLocalPort());        
    }
    
    @Override
    public String toString()
    {
        return "JavaConnection " + 
        (socketChannel_ == null ? "with no socket configured" : "for " + getLocalAddress()) +
        (disconnecting_ ? " (disconnecting)" : "");
    }
    
    private final EventLoop eventLoop_;
    private boolean disconnecting_ = false;
    private SocketChannel socketChannel_;
    private Worker worker_;

}