package net.ihiroky.niotty.nio;

import net.ihiroky.niotty.DefaultLoadPipeline;
import net.ihiroky.niotty.DefaultStorePipeline;
import net.ihiroky.niotty.DefaultTransportFuture;
import net.ihiroky.niotty.EventLoop;
import net.ihiroky.niotty.TransportConfig;
import net.ihiroky.niotty.TransportFuture;
import net.ihiroky.niotty.buffer.BufferSink;
import net.ihiroky.niotty.event.MessageEvent;
import net.ihiroky.niotty.event.TransportState;
import net.ihiroky.niotty.event.TransportStateEvent;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.LinkedList;
import java.util.Queue;

/**
 * Created on 13/01/15, 16:50
 *
 * TODO abstracts ByteBuffer to to support 'zero copy' writing / reading (FileChannel)
 * @author Hiroki Itoh
 */
public class NioChildChannelTransport extends NioSocketTransport<MessageIOSelector> {

    private boolean remainingPreviousData_;
    private ByteBuffer writeBuffer_;
    private Queue<BufferSink> pendingQueue_;

    NioChildChannelTransport(TransportConfig config, String name, int writeBufferSize, boolean directWriteBuffer) {
        DefaultLoadPipeline loadPipeline = DefaultLoadPipeline.createPipeline(name);
        DefaultStorePipeline storePipeline = DefaultStorePipeline.createPipeline(name);
        config.getPipelineInitializer().setUpPipeline(loadPipeline, storePipeline);

        loadPipeline.regulate();
        loadPipeline.verifyStageContextType();

        storePipeline.regulate();
        storePipeline.getLastContext().addListener(MessageIOSelector.MESSAGE_IO_STORE_CONTEXT_LISTENER);
        storePipeline.verifyStageContextType();

        ByteBuffer writeBuffer = directWriteBuffer
                ? ByteBuffer.allocateDirect(writeBufferSize) : ByteBuffer.allocate(writeBufferSize);

        setLoadPipeline(loadPipeline);
        setStorePipeline(storePipeline);
        this.writeBuffer_ = writeBuffer;
        this.pendingQueue_ = new LinkedList<>();
    }

    @Override
    public void write(Object message) {
        fire(new MessageEvent<>(this, message));
    }

    @Override
    public void write(Object message, SocketAddress remote) {
        throw new UnsupportedOperationException("write");
    }

    @Override
    public InetSocketAddress localAddress() {
        SelectableChannel channel = getSelectionKey().channel();
        try {
            return (InetSocketAddress) ((SocketChannel) channel).getLocalAddress();
        } catch (IOException ioe) {
            throw new RuntimeException(ioe);
        }
    }

    @Override
    public InetSocketAddress remoteAddress() {
        SelectableChannel channel = getSelectionKey().channel();
        try {
            return (InetSocketAddress) ((SocketChannel) channel).getRemoteAddress();
        } catch (IOException ioe) {
            throw new RuntimeException(ioe);
        }
    }

    @Override
    public boolean isOpen() {
        return getSelectionKey().channel().isOpen();
    }

    @Override
    public TransportFuture bind(SocketAddress socketAddress) {
        throw new UnsupportedOperationException("bind");
    }

    @Override
    public TransportFuture connect(SocketAddress remoteAddress) {
        throw new UnsupportedOperationException("connect");
    }

    @Override
    public TransportFuture close() {
        DefaultTransportFuture future = new DefaultTransportFuture(this);
        fire(new TransportStateEvent(this, TransportState.CONNECTED, future, null));
        return future;
    }

    @Override
    public TransportFuture join(InetAddress group, NetworkInterface networkInterface, InetAddress source) {
        throw new UnsupportedOperationException("join");
    }

    @Override
    protected void writeDirect(final BufferSink buffer) {
        EventLoop<MessageIOSelector> loop = getEventLoop();
        if (loop != null) {
            loop.offerTask(new EventLoop.Task<MessageIOSelector>() {
                @Override
                public boolean execute(MessageIOSelector eventLoop) throws Exception {
                    writeBufferSink(buffer);
                    eventLoop.flushLater(NioChildChannelTransport.this);
                    return true;
                }
            });
        }
    }

    void writeBufferSink(BufferSink buffer) throws IOException {
        WritableByteChannel channel = (WritableByteChannel) getSelectionKey().channel();
        ByteBuffer localWriteBuffer = writeBuffer_;

        if (remainingPreviousData_) {
            channel.write(localWriteBuffer);
            if (localWriteBuffer.hasRemaining()) {
                pendingQueue_.offer(buffer);
                return;
            }
            remainingPreviousData_ = false;
            localWriteBuffer.clear();
        }

        if (pendingQueue_.isEmpty()) {
            if (buffer.transferTo(channel, localWriteBuffer)) {
                localWriteBuffer.clear();
            } else {
                remainingPreviousData_ = true;
                pendingQueue_.offer(buffer);
            }
        }
    }

    boolean flush() throws IOException {
        WritableByteChannel channel = (WritableByteChannel) getSelectionKey().channel();
        ByteBuffer localWriteBuffer = writeBuffer_;

        if (remainingPreviousData_) {
            channel.write(localWriteBuffer);
            if (localWriteBuffer.hasRemaining()) {
                return false;
            }
            remainingPreviousData_ = false;
            localWriteBuffer.clear();
        }

        for (BufferSink pendingBuffer; (pendingBuffer = pendingQueue_.peek()) != null; ) {
            if (pendingBuffer.transferTo(channel, localWriteBuffer)) {
                localWriteBuffer.clear();
                pendingQueue_.poll();
            } else {
                remainingPreviousData_ = true;
                return false;
            }
        }
        return true;
    }

    private void fire(MessageEvent<Object> event) {
        getStorePipeline().fire(event);
    }

    private void fire(TransportStateEvent event) {
        getStorePipeline().fire(event);
    }

    void loadEvent(MessageEvent<Object> event) {
        getLoadPipeline().fire(event);
    }

    void loadEventLater(final TransportStateEvent event) {
        offerTask(new EventLoop.Task<MessageIOSelector>() {
            @Override
            public boolean execute(MessageIOSelector eventLoop) throws Exception {
                getLoadPipeline().fire(event);
                return true;
            }
        });
    }
}

@@ -4499,762 +4499,46 @@
 er) 
-throws IOException %7B%0A        WritableByteChannel channel = (WritableByteChannel) getSelectionKey().channel();%0A        ByteBuffer localWriteBuffer = writeBuffer_;%0A%0A        if (remainingPreviousData_) %7B%0A            channel.write(localWriteBuffer);%0A            if (localWriteBuffer.hasRemaining()) %7B%0A                pendingQueue_.offer(buffer);%0A                return;%0A            %7D%0A            remainingPreviousData_ = false;%0A            localWriteBuffer.clear();%0A        %7D%0A%0A        if (pendingQueue_.isEmpty()) %7B%0A            if (buffer.transferTo(channel, localWriteBuffer)) %7B%0A                localWriteBuffer.clear();%0A            %7D else %7B%0A                remainingPreviousData_ = true;%0A                pendingQueue_.offer(buffer);%0A            %7D%0A        %7D
+%7B%0A        pendingQueue_.offer(buffer);
 %0A   
@@ -4773,16 +4773,30 @@
         
+ int written =
  channel
@@ -4908,32 +4908,136 @@
 ;%0A            %7D%0A
+            if (written == -1) %7B%0A                throw new IOException(%22end of stream.%22);%0A            %7D%0A
             rema
@@ -5126,16 +5126,34 @@
    for (
+;;) %7B%0A            
 BufferSi
@@ -5172,24 +5172,8 @@
 ffer
-; (pendingBuffer
  = p
@@ -5195,23 +5195,88 @@
 ek()
-) !
+;%0A            if (pendingBuffer =
 = null
-; 
 ) %7B%0A
+                break;%0A            %7D%0A
     

Fix write operation on non I/O thread