package org.jruby.jubilee;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.jruby.*;
import org.jruby.anno.JRubyClass;
import org.jruby.anno.JRubyMethod;
import org.jruby.runtime.ObjectAllocator;
import org.jruby.runtime.ThreadContext;
import org.jruby.runtime.builtin.IRubyObject;
import org.jruby.util.ByteList;
import org.vertx.java.core.Handler;
import org.vertx.java.core.VoidHandler;
import org.vertx.java.core.buffer.Buffer;
import org.vertx.java.core.net.NetSocket;
import org.vertx.java.core.streams.WriteStream;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * This class create a ruby IO interface by wrapping a Vertx NetSocket object.
 * <p/>
 * Not threadsafe.
 */
@JRubyClass(name="NetSocket")
public class RubyNetSocket extends RubyObject {
    private NetSocket sock;
    private ByteBuf buf;
    private AtomicBoolean eof;
    private boolean readClosed = false;
    private boolean writeClosed = false;
    private boolean closed = false;

    private final static int BUFSIZE = 4096 * 2;

    public static RubyClass createNetSocketClass(final Ruby runtime) {
        RubyModule jubilee = runtime.getOrCreateModule("Jubilee");
        RubyClass klazz = jubilee.defineClassUnder("NetSocket", runtime.getObject(), new ObjectAllocator() {
            @Override
            public IRubyObject allocate(Ruby ruby, RubyClass rubyClass) {
                return new RubyNetSocket(ruby, rubyClass);
            }
        });
        klazz.defineAnnotatedMethods(RubyNetSocket.class);
        return klazz;
    }

    public RubyNetSocket(Ruby ruby, RubyClass rubyClass) {
        super(ruby, rubyClass);
    }

    public RubyNetSocket(Ruby ruby, RubyClass rubyClass, NetSocket socket) {
        super(ruby, rubyClass);
        this.sock = socket;
        this.buf = Unpooled.buffer(BUFSIZE);
        this.eof = new AtomicBoolean(false);
        this.sock.dataHandler(new Handler<Buffer>() {
            @Override
            public void handle(Buffer buffer) {
                if (buf.isWritable(buffer.length()))
                    buf.writeBytes(buffer.getByteBuf());
                else sock.pause();
            }
        });

        this.sock.endHandler(new VoidHandler() {
            @Override
            protected void handle() {
                eof.set(true);
            }
        });
    }

    /**
     * Both of the calls block
     *
     * @param context
     * @param args
     * @return
     */
    @JRubyMethod(name = {"read", "read_nonblock"}, required = 1, optional = 1)
    public IRubyObject read(ThreadContext context, IRubyObject[] args) {
        if (this.readClosed) throw context.runtime.newIOError("closed stream");
        int length = RubyNumeric.num2int(args[0]);
        byte[] data;
        if (args.length == 1)
            data = new byte[length];
        else data = ((RubyString) args[1]).getBytes();
        if (!(eof.get() || buf.isReadable())) {
            this.buf.clear();
            this.sock.resume();
        }
        waitReadable(this.buf);
        while (!eof.get() && length > 0) {
            int readedLength = Math.min(this.buf.readableBytes(), length);
            this.buf.readBytes(data, this.buf.readerIndex(), readedLength);
            length -= readedLength;
        }
        return context.runtime.newString(new ByteList(data));
    }

    /**
     * Though required by rack spec to impelement write_nonblock, it's just easier to block both of the calls.
     *
     * @param context the calling threadcontext
     * @param str     the string to write to the underline stream
     * @return the length written
     */
    @JRubyMethod(name = {"write", "write_nonblock"}, required = 1)
    public IRubyObject write(ThreadContext context, IRubyObject str) {
        if (this.writeClosed) throw context.runtime.newIOError("closed stream");
        RubyString data;
        if (str instanceof RubyString)
            data = (RubyString) str;
        else
            data = (RubyString) str.callMethod(context, "to_s");
        if (this.sock.writeQueueFull())
            waitWritable(this.sock);
        this.sock.write(data.asJavaString());
        // TODO return the length actually written
        return data.length();
    }

    @JRubyMethod(name = "close_write")
    public IRubyObject closeWrite(ThreadContext context) {
        this.writeClosed = true;
        return context.runtime.getTrue();
    }

    @JRubyMethod(name = "closeRead")
    public IRubyObject closeRead(ThreadContext context) {
        this.readClosed = true;
        return context.runtime.getTrue();
    }

    @JRubyMethod(name = "close")
    public IRubyObject close(ThreadContext context) {
        this.sock.close();
        this.closed = true;
        return context.runtime.getTrue();
    }

    @JRubyMethod(name = "closed?")
    public IRubyObject isClosed(ThreadContext context) {
        return context.runtime.newBoolean(this.closed);
    }

    @JRubyMethod(name = "flush")
    public IRubyObject flush(ThreadContext context) {
        return context.runtime.getTrue();
    }

    private void waitWritable(WriteStream<?> stream) {
        final AtomicBoolean writable = new AtomicBoolean(false);
        stream.drainHandler(new Handler<Void>() {
            @Override
            public void handle(Void empty) {
                writable.set(true);
            }
        });
        while (!writable.get())
            ;
    }

    private void waitReadable(ByteBuf buf) {
        while (!buf.isReadable())
            ;
    }
}