package org.jvirtanen.parity.client;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import org.jvirtanen.nassau.soupbintcp.SoupBinTCP;
import org.jvirtanen.nassau.soupbintcp.SoupBinTCPClient;
import org.jvirtanen.nassau.soupbintcp.SoupBinTCPClientStatusListener;
import org.jvirtanen.parity.net.poe.POE;
import org.jvirtanen.parity.net.poe.POEClientListener;
import org.jvirtanen.parity.net.poe.POEClientParser;

public class OrderEntry implements Closeable {

    private ByteBuffer buffer;

    private Selector selector;

    private SoupBinTCPClient transport;

    private volatile boolean closed;

    private OrderEntry(Selector selector, SocketChannel channel, POEClientListener listener) {
        this.buffer = ByteBuffer.allocate(POE.MAX_INBOUND_MESSAGE_LENGTH);

        this.selector = selector;

        this.transport = new SoupBinTCPClient(channel, POE.MAX_OUTBOUND_MESSAGE_LENGTH,
                new POEClientParser(listener), new StatusListener());

        this.closed = false;

        new Thread(new Receiver()).start();
    }

    public static OrderEntry open(InetSocketAddress address, POEClientListener listener) throws IOException {
        SocketChannel channel = SocketChannel.open();

        channel.connect(address);
        channel.configureBlocking(false);

        Selector selector = Selector.open();

        channel.register(selector, SelectionKey.OP_READ);

        return new OrderEntry(selector, channel, listener);
    }

    @Override
    public void close() {
        closed = true;
    }

    public SoupBinTCPClient getTransport() {
        return transport;
    }

    public void send(POE.InboundMessage message) throws IOException {
        buffer.clear();
        message.put(buffer);
        buffer.flip();

        transport.send(buffer);
    }

    private class StatusListener implements SoupBinTCPClientStatusListener {

        @Override
        public void heartbeatTimeout() {
            close();
        }

        @Override
        public void loginAccepted(SoupBinTCP.LoginAccepted payload) {
        }

        @Override
        public void loginRejected(SoupBinTCP.LoginRejected payload) {
            close();
        }

        @Override
        public void endOfSession() {
        }

    }

    private class Receiver implements Runnable {

        private static final long TIMEOUT_MILLIS = 100;

        @Override
        public void run() {
            try {
                while (!closed) {
                    int numKeys = selector.select(TIMEOUT_MILLIS);
                    if (numKeys > 0) {
                        if (transport.receive() < 0)
                            break;

                        selector.selectedKeys().clear();
                    }

                    transport.keepAlive();
                }
            } catch (IOException e) {
            }

            try {
                transport.close();
            } catch (IOException e) {
            }

            try {
                selector.close();
            } catch (IOException e) {
            }
        }

    }

}

@@ -782,16 +782,44 @@
 losed;%0A%0A
+    private Object rxLock;%0A%0A
     priv
@@ -901,24 +901,24 @@
 listener) %7B%0A
-
         this
@@ -1205,16 +1205,53 @@
 false;%0A%0A
+        this.rxLock = new Object();%0A%0A
         
@@ -1998,24 +1998,60 @@
 er.flip();%0A%0A
+        synchronized (rxLock) %7B%0A    
         tran
@@ -2066,24 +2066,34 @@
 nd(buffer);%0A
+        %7D%0A
     %7D%0A%0A    p
@@ -3020,32 +3020,80 @@
              %7D%0A%0A
+                    synchronized (rxLock) %7B%0A    
                 
@@ -3115,24 +3115,46 @@
 eepAlive();%0A
+                    %7D%0A
             

Add SoupBinTCP transmission lock to terminal client