//
// MessagePack-RPC for Java
//
// Copyright (C) 2010 Kazuki Ohta
//
//    Licensed under the Apache License, Version 2.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://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
//
package org.msgpack.rpc.client.transport;

import java.util.ArrayList;
import java.util.List;

import org.msgpack.rpc.client.EventLoop;
import org.msgpack.rpc.client.Session;

/**
 * TCPTransport sends/receives the data through TCP, by using underlying
 * TCPSocket layer.
 *
 * This class also hides the latency of establishing the connection. If the
 * connection is not established. the sending messages are temporarily queued.
 * Then, they are actually sent to the network when it's connected.
 */
public class TCPTransport extends Transport {
    protected Boolean isConnecting;
    protected Boolean isConnected;
    protected TCPSocket socket;
    protected List<Object> pendingMessages;

    public TCPTransport(Session session, EventLoop loop) {
        super(session, loop);
        this.isConnecting = false;
        this.isConnected = false;
        this.socket = new TCPSocket(session.getAddress(), loop, this);
        this.pendingMessages = new ArrayList<Object>();
    }

    /**
     * Send the message to the remote server.
     *
     * This method tries to hide the connection latency.
     * If it's already connected, then the message is sent immediately.
     * Otherwise, the message is temporarily buffered in the
     * this.pendingMessages. Once it's connected, the buffered messages are
     * sent to the server.
     * 
     * @param msg the message to send.
     * @throws Exception
     */
    @Override
    public synchronized void sendMessage(Object msg) throws Exception {
        if (isConnected) {
            socket.trySend(msg);
        } else {
            if (!isConnecting) {
                socket.tryConnect();
                isConnecting = true;
            }
            pendingMessages.add(msg);
        }
    }

    /**
     * Send the pending messages.
     * @throws Exception
     */
    protected void trySendPending() throws Exception {
        Object[] msgs;
        synchronized(this) {
            msgs = pendingMessages.toArray();
            pendingMessages.clear();
        }
        for (Object msg : msgs)
            socket.trySend(msg);
    }

    /**
     * Close the connection associated with this transport.
     */
    @Override
    public synchronized void tryClose() {
        if (socket != null)
            socket.tryClose();
        isConnecting = false;
        isConnected = false;
        pendingMessages.clear();
    }
    
    /**
     * The callback function, called when the connection is established.
     * @throws Exception
     */
    public synchronized void onConnected() throws Exception {
        isConnecting = false;
        isConnected = true;
        trySendPending();
    }
    
    /**
     * The callback called when the message arrives
     * @param replyObject the received object, already unpacked.
     * @throws Exception
     */
    public void onMessageReceived(Object replyObjects) throws Exception {
        session.onMessageReceived(replyObjects);
    }

    /**
     * The callback function, called when the connection failed.
     */
    public void onConnectFailed() {
        tryClose();
        session.onConnectFailed();
    }
    
    /**
     * The callback called when the connection closed.
     */
    public void onClosed() {
        tryClose();
        session.onClosed();
    }
    
    /**
     * The callback called when the error occurred.
     * @param e occurred exception.
     */
    public void onFailed(Exception e) {
        tryClose();
        session.onFailed(e);
    }
}

@@ -1198,16 +1198,54 @@
 sport %7B%0A
+    protected final TCPSocket socket;%0A
     prot
@@ -1311,40 +1311,8 @@
 ed;%0A
-    protected TCPSocket socket;%0A
     
@@ -2099,37 +2099,24 @@
 %0A    public 
-synchronized 
 void sendMes
@@ -2163,60 +2163,171 @@
     
-if (isConnected) %7B%0A            socket.trySend(msg);%0A
+boolean isTryConnect = false;%0A        boolean isTrySend = false;%0A        synchronized(this) %7B%0A            if (isConnected) %7B%0A                isTrySend = true;%0A    
     
@@ -2339,16 +2339,20 @@
  else %7B%0A
+    
         
@@ -2392,24 +2392,23 @@
         
+ 
-socket.t
+   isT
 ryConnec
@@ -2408,20 +2408,29 @@
 yConnect
-();%0A
+ = true;%0A    
         
@@ -2462,34 +2462,42 @@
 ue;%0A            
+ 
-%7D%0A
+   %7D%0A    
             pend
@@ -2530,322 +2530,104 @@
     
-%7D%0A
     %7D%0A
-%0A
     
-/**%0A     * Send the pending messages.%0A     * @throws Exception%0A     */%0A    protected void trySendPending() throws Exception %7B%0A        Object%5B%5D msgs;%0A        synchronized(this) %7B%0A            msgs = pendingMessages.toArray();%0A            pendingMessages.clear();%0A        %7D%0A        for (Object msg : msgs
+    %7D%0A        if (isTryConnect)%0A            socket.tryConnect();%0A        if (isTrySend
 )%0A  
@@ -2761,37 +2761,24 @@
 %0A    public 
-synchronized 
 void tryClos
@@ -2795,59 +2795,33 @@
     
-if (socket != null)%0A            socket.tryClose();%0A
+synchronized(this) %7B%0A    
     
@@ -2838,32 +2838,36 @@
 ecting = false;%0A
+    
         isConnec
@@ -2879,32 +2879,36 @@
  false;%0A        
+    
 pendingMessages.
@@ -2912,24 +2912,61 @@
 es.clear();%0A
+        %7D%0A        socket.tryClose();%0A
     %7D%0A    %0A 
@@ -3093,21 +3093,8 @@
 lic 
-synchronized 
 void
@@ -3119,32 +3119,125 @@
 ows Exception %7B%0A
+        // send the pending messages%0A        Object%5B%5D msgs;%0A        synchronized(this) %7B%0A    
         isConnec
@@ -3242,32 +3242,36 @@
 ecting = false;%0A
+    
         isConnec
@@ -3294,23 +3294,155 @@
     
-trySendPending(
+    msgs = pendingMessages.toArray();%0A            pendingMessages.clear();%0A        %7D%0A        for (Object msg : msgs)%0A            socket.trySend(msg
 );%0A 

remove unnecessary locks from TCPTransport.