IOConnection.java

/*
 * socket.io-java-client IOConnection.java
 *
 * Copyright (c) 2012, Enno Boland
 * socket.io-java-client is a implementation of the socket.io protocol in Java.
 * 
 * See LICENSE file for more information
 */
package io.socket;

import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParseException;
import com.google.gson.JsonParser;
import com.google.gson.JsonPrimitive;

import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Scanner;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Logger;

import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;

/**
 * The Class IOConnection.
 */
class IOConnection implements IOCallback {
  /** Debug logger */
  static final Logger logger = Logger.getLogger("io.socket");

  public static final String FRAME_DELIMITER = "\ufffd";

  /** The Constant STATE_INIT. */
  private static final int STATE_INIT = 0;

  /** The Constant STATE_HANDSHAKE. */
  private static final int STATE_HANDSHAKE = 1;

  /** The Constant STATE_CONNECTING. */
  private static final int STATE_CONNECTING = 2;

  /** The Constant STATE_READY. */
  private static final int STATE_READY = 3;

  /** The Constant STATE_INTERRUPTED. */
  private static final int STATE_INTERRUPTED = 4;

  /** The Constant STATE_INVALID. */
  private static final int STATE_INVALID = 6;

  /** The state. */
  private int state = STATE_INIT;

  /** Socket.io path. */
  public static final String SOCKET_IO_1 = "/socket.io/1/";

  /** The SSL socket factory for HTTPS connections */
  private static SSLContext sslContext = null;

  /** All available connections. */
  private static HashMap<String, List<IOConnection>> connections = new HashMap<String, List<IOConnection>>();

  /** The url for this connection. */
  private URL url;

  /** The transport for this connection. */
  private IOTransport transport;

  /** The connection timeout. */
  private int connectTimeout = 10000;

  /** The session id of this connection. */
  private String sessionId;

  /** The heartbeat timeout. Set by the server */
  private long heartbeatTimeout;

  /** The closing timeout. Set By the server */
  private long closingTimeout;

  /** The protocols supported by the server. */
  private List<String> protocols;

  /** The output buffer used to cache messages while (re-)connecting. */
  private ConcurrentLinkedQueue<String> outputBuffer = new ConcurrentLinkedQueue<String>();

  /** The sockets of this connection. */
  private HashMap<String, SocketIO> sockets = new HashMap<String, SocketIO>();

  /** Custom Request headers used while handshaking */
  private Properties headers;
  
  private String queryString;

  /**
   * The first socket to be connected. the socket.io server does not send a
   * connected response to this one.
   */
  private SocketIO firstSocket = null;

  /** The reconnect timer. IOConnect waits a second before trying to reconnect */
  private Timer backgroundTimer = new Timer("backgroundTimer");

  /** A String representation of {@link #url}. */
  private String urlStr;

  /**
   * The last occurred exception, which will be given to the user if
   * IOConnection gives up.
   */
  private Exception lastException;

  /** The next ID to use. */
  private int nextId = 1;

  /** Acknowledges. */
  HashMap<Integer, AckEntry> acknowledge = new HashMap<Integer, AckEntry>();

  /** true if there's already a keepalive in {@link #outputBuffer}. */
  private boolean keepAliveInQueue;

  /**
   * The heartbeat timeout task. Only null before connection has been
   * initialised.
   */
  private HearbeatTimeoutTask heartbeatTimeoutTask;

  /**
   * The Class HearbeatTimeoutTask. Handles dropping this IOConnection if no
   * heartbeat is received within life time.
   */
  private class HearbeatTimeoutTask extends TimerTask {

    /*
     * (non-Javadoc)
     * 
     * @see java.util.TimerTask#run()
     */
    @Override
    public void run() {
      error(new SocketIOException( "Timeout Error. No heartbeat from server within life time of the socket. closing.", lastException));
    }
  }

  /** The reconnect task. Null if no reconnection is in progress. */
  private ReconnectTask reconnectTask = null;

  /**
   * The Class ReconnectTask. Handles reconnect attempts
   */
  private class ReconnectTask extends TimerTask {

    /*
     * (non-Javadoc)
     * 
     * @see java.util.TimerTask#run()
     */
    @Override
    public void run() {
      connectTransport();
      if (!keepAliveInQueue) {
        sendPlain("2::");
        keepAliveInQueue = true;
      }
    }
  }

  /**
   * The Class ConnectThread. Handles connecting to the server with an
   * {@link IOTransport}
   */
  private class ConnectThread extends Thread {
    /**
     * Instantiates a new thread for handshaking/connecting.
     */
    public ConnectThread() {
      super("ConnectThread");
    }

    /**
     * Tries handshaking if necessary and connects with corresponding
     * transport afterwards.
     */
    @Override
    public void run() {
      if (IOConnection.this.getState() == STATE_INIT)
        handshake();
      connectTransport();
    }

  };

  /**
   * Set the socket factory used for SSL connections.
   * 
   * @param sslContext
   */
  public static void setSslContext(SSLContext sslContext) {
    IOConnection.sslContext = sslContext;
  }

  /**
   * Get the socket factory used for SSL connections.
   * 
   * @return socketFactory
   */
  public static SSLContext getSslContext() {
    return sslContext;
  }

  /**
   * Creates a new connection or returns the corresponding one.
   * 
   * @param origin
   *            the origin
   * @param socket
   *            the socket
   * @return a IOConnection object
   */
  static public IOConnection register(String origin, SocketIO socket) {
    List<IOConnection> list = connections.get(origin);
    if (list == null) {
      list = new LinkedList<IOConnection>();
      connections.put(origin, list);
    } else {
      synchronized (list) {
        for (IOConnection connection : list) {
          if (connection.register(socket))
            return connection;
        }
      }
    }

    IOConnection connection = new IOConnection(origin, socket);
    list.add(connection);
    return connection;
  }

  /**
   * Connects a socket to the IOConnection.
   * 
   * @param socket
   *            the socket to be connected
   * @return true, if successfully registered on this transport, otherwise
   *         false.
   */
  public synchronized boolean register(SocketIO socket) {
    String namespace = socket.getNamespace();
    if (sockets.containsKey(namespace))
      return false;
    sockets.put(namespace, socket);
    socket.setHeaders(headers);
    IOMessage connect = new IOMessage(IOMessage.TYPE_CONNECT,
        socket.getNamespace(), "");
    sendPlain(connect.toString());
    return true;
  }

  /**
   * Disconnect a socket from the IOConnection. Shuts down this IOConnection
   * if no further connections are available for this IOConnection.
   * 
   * @param socket
   *            the socket to be shut down
   */
  public synchronized void unregister(SocketIO socket) {
    sendPlain("0::" + socket.getNamespace());
    sockets.remove(socket.getNamespace());
    socket.getCallback().onDisconnect();

    if (sockets.size() == 0) {
      cleanup();
    }
  }

  /**
   * Handshake.
   * 
   */
  private void handshake() {
    URL url;
    String response;
    URLConnection connection;
    try {
      setState(STATE_HANDSHAKE);
      
      String connectionUrl = IOConnection.this.url.toString() + SOCKET_IO_1;
      if ( this.queryString != null ) {
        connectionUrl += "?" + this.queryString;
      }
      
      url = new URL( connectionUrl );
      connection = url.openConnection();
      if (connection instanceof HttpsURLConnection) {
        ((HttpsURLConnection) connection).setSSLSocketFactory(sslContext.getSocketFactory());
      }
      connection.setConnectTimeout(connectTimeout);
      connection.setReadTimeout(connectTimeout);

      /* Setting the request headers */
      for (Entry<Object, Object> entry : headers.entrySet()) {
        connection.setRequestProperty( (String) entry.getKey(), (String) entry.getValue() );
      }

      InputStream stream = connection.getInputStream();
      Scanner in = new Scanner(stream);
      response = in.nextLine();
      String[] data = response.split(":");
      sessionId = data[0];
      heartbeatTimeout = Long.parseLong(data[1]) * 1000;
      closingTimeout = Long.parseLong(data[2]) * 1000;
      protocols = Arrays.asList(data[3].split(","));
    } catch (Exception e) {
      error(new SocketIOException("Error while handshaking", e));
    }
  }

  /**
   * Connect transport.
   */
  private synchronized void connectTransport() {
    if (getState() == STATE_INVALID)
      return;
    setState(STATE_CONNECTING);
    if (protocols.contains(WebsocketTransport.TRANSPORT_NAME))
      transport = WebsocketTransport.create(url, this);
    else if (protocols.contains(XhrTransport.TRANSPORT_NAME))
      transport = XhrTransport.create(url, this);
    else {
      error(new SocketIOException( "Server supports no available transports. You should reconfigure the server to support a available transport") );
      return;
    }
    transport.connect();
  }

  /**
   * Creates a new {@link IOAcknowledge} instance which sends its arguments
   * back to the server.
   * 
   * @param message
   *            the message
   * @return an {@link IOAcknowledge} instance, may be <code>null</code> if
   *         server doesn't request one.
   */
  private IOAcknowledge remoteAcknowledge(IOMessage message) {
    String _id = message.getId();
    if (_id.equals(""))
      return null;
    else if (_id.endsWith("+") == false)
      _id = _id + "+";
    final String id = _id;
    final String endPoint = message.getEndpoint();
    return new IOAcknowledge() {
      @Override
      public void ack(JsonElement... args) {
        JsonArray array = new JsonArray();
        for (JsonElement o : args) {
          try {
            array.add(o);
          } catch (Exception e) {
            error(new SocketIOException("You can only put values in IOAcknowledge.ack() which can be handled by JSONArray.put()", e));
          }
        }
        IOMessage ackMsg = new IOMessage(IOMessage.TYPE_ACK, endPoint, id + array.toString());
        sendPlain(ackMsg.toString());
      }
    };
  }

  private static class AckEntry {
    AckTimeoutTask ackTimeoutTask;
    IOAcknowledge ack;
  }
	
  /**
   * The Class AckTimeoutTask. Handles timeouts on ACK
   */
  private class AckTimeoutTask extends TimerTask {

    int ackId;
    IOAcknowledgeWithTimeout ack;

    AckTimeoutTask(int ackId, IOAcknowledgeWithTimeout ack) {
      this.ackId = ackId;
      this.ack = ack;
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.util.TimerTask#run()
     */
    @Override
    public void run() {
      synchronized (acknowledge) {
        acknowledge.remove(ackId);	
      }
      logger.warning("! ack " + ackId + " timeout");
      ack.timeout();
    }
  }

  /**
   * adds an {@link IOAcknowledge} to an {@link IOMessage}.
   * 
   * @param message
   *            the {@link IOMessage}
   * @param ack
   *            the {@link IOAcknowledge}
   */
  private void synthesizeAck(IOMessage message, IOAcknowledge ack) {
    if (ack != null) {
      int id = nextId++;
      AckEntry entry = new AckEntry();
      entry.ack = ack;
      if (ack instanceof IOAcknowledgeWithTimeout) {
        IOAcknowledgeWithTimeout timeoutAck = (IOAcknowledgeWithTimeout) ack;
        entry.ackTimeoutTask = new AckTimeoutTask(id, timeoutAck);
      }
      message.setId(id + "+");
      synchronized (acknowledge) {
        acknowledge.put(id, entry);
      }
      if (entry.ackTimeoutTask != null) {
        long timeout = entry.ackTimeoutTask.ack.getTimeout();
        backgroundTimer.schedule(entry.ackTimeoutTask, timeout);
      }
    }
  }

  /**
   * Instantiates a new IOConnection.
   * 
   * @param url
   *            the URL
   * @param socket
   *            the socket
   */
  private IOConnection(String url, SocketIO socket) {
    try {
      this.url = new URL(url);
      this.urlStr = url;
      this.queryString = socket.getQueryString();
    } catch (MalformedURLException e) {
      throw new RuntimeException(e);
    }
    firstSocket = socket;
    headers = socket.getHeaders();
    sockets.put(socket.getNamespace(), socket);
    new ConnectThread().start();
  }

  /**
   * Cleanup. IOConnection is not usable after this calling this.
   */
  private synchronized void cleanup() {
    setState(STATE_INVALID);
    if (transport != null)
      transport.disconnect();
    sockets.clear();
    synchronized (connections) {
      List<IOConnection> con = connections.get(urlStr);
      if (con != null && con.size() > 1)
        con.remove(this);
      else
        connections.remove(urlStr);
    }
    logger.info("Cleanup");
    backgroundTimer.cancel();
    backgroundTimer = new Timer("backgroundTimer");
  }

  /**
   * Populates an error to the connected {@link IOCallback}s and shuts down.
   * 
   * @param e
   *            an exception
   */
  private void error(SocketIOException e) {
    for (SocketIO socket : sockets.values()) {
      socket.getCallback().onError(e);
    }
    cleanup();
  }

  /**
   * Sends a plain message to the {@link IOTransport}.
   * 
   * @param text
   *            the Text to be send.
   */
  private synchronized void sendPlain(String text) {
    if (getState() == STATE_READY)
      try {
        logger.info("> " + text);
        transport.send(text);
      } catch (Exception e) {
        logger.info("IOEx: saving");
        outputBuffer.add(text);
      }
    else {
      outputBuffer.add(text);
    }
  }

  /**
   * Invalidates an {@link IOTransport}, used for forced reconnecting.
   */
  private void invalidateTransport() {
    if (transport != null)
      transport.invalidate();
    transport = null;
  }

  /**
   * Reset timeout.
   */
  private synchronized void resetTimeout() {
    if (heartbeatTimeoutTask != null) {
      heartbeatTimeoutTask.cancel();
    }
    if (getState() != STATE_INVALID) {
      heartbeatTimeoutTask = new HearbeatTimeoutTask();
      backgroundTimer.schedule(heartbeatTimeoutTask, closingTimeout
          + heartbeatTimeout);
    }
  }

  /**
   * finds the corresponding callback object to an incoming message. Returns a
   * dummy callback if no corresponding callback can be found
   * 
   * @param message
   *            the message
   * @return the iO callback
   * @throws SocketIOException
   */
  private IOCallback findCallback(IOMessage message) throws SocketIOException {
    if ("".equals(message.getEndpoint()))
      return this;
    SocketIO socket = sockets.get(message.getEndpoint());
    if (socket == null) {
      throw new SocketIOException("Cannot find socket for '"
          + message.getEndpoint() + "'");
    }
    return socket.getCallback();
  }

  /**
   * Transport connected.
   * 
   * {@link IOTransport} calls this when a connection is established.
   */
  public synchronized void transportConnected() {
    if (reconnectTask != null) {
      reconnectTask.cancel();
      reconnectTask = null;
      onReconnect();
    }
    resetTimeout();
    this.keepAliveInQueue = false;
  }

  /**
   * Transport disconnected.
   * 
   * {@link IOTransport} calls this when a connection has been shut down.
   */
  public void transportDisconnected() {
    this.lastException = null;
    setState(STATE_INTERRUPTED);
    reconnect();
  }

  /**
   * Transport error.
   * 
   * @param error
   *            the error {@link IOTransport} calls this, when an exception
   *            has occurred and the transport is not usable anymore.
   */
  public void transportError(Exception error) {
    this.lastException = error;
    setState(STATE_INTERRUPTED);
    reconnect();
  }

  /**
   * {@link IOTransport} should call this function if it does not support
   * framing. If it does, transportMessage should be used
   * 
   * @param text
   *            the text
   */
  public void transportData(String text) {
    if (!text.startsWith(FRAME_DELIMITER)) {
      transportMessage(text);
      return;
    }

    Iterator<String> fragments = Arrays.asList(text.split(FRAME_DELIMITER))
        .listIterator(1);
    while (fragments.hasNext()) {
      int length = Integer.parseInt(fragments.next());
      String string = fragments.next();
      // Potential BUG: it is not defined if length is in bytes or
      // characters. Assuming characters.

      if (length != string.length()) {
        error(new SocketIOException("Garbage from server: " + text));
        return;
      }

      transportMessage(string);
    }
  }

  /**
   * Transport message. {@link IOTransport} calls this, when a message has
   * been received.
   * 
   * @param text
   *            the text
   */
  public void transportMessage(String text) {
    logger.info("< " + text);
    IOMessage message;
    try {
      message = new IOMessage(text);
    } catch (Exception e) {
      error(new SocketIOException("Garbage from server: " + text, e));
      return;
    }
    resetTimeout();
    switch (message.getType()) {
    case IOMessage.TYPE_DISCONNECT:
      try {
        findCallback(message).onDisconnect();
      } catch (Exception e) {
        error(new SocketIOException(
            "Exception was thrown in onDisconnect()", e));
      }
      break;
    case IOMessage.TYPE_CONNECT:
      try {
        if (firstSocket != null && "".equals(message.getEndpoint())) {
          setState(STATE_READY);
          if (firstSocket.getNamespace().equals("")) {
            firstSocket.getCallback().onConnect();
          } else {
            IOMessage connect = new IOMessage(
                IOMessage.TYPE_CONNECT,
                firstSocket.getNamespace(), "");
            sendPlain(connect.toString());
          }
          // should flush after connecting to namespace
          flushBuffer();
        } else {
          findCallback(message).onConnect();
        }
        firstSocket = null;
      } catch (Exception e) {
        error(new SocketIOException(
            "Exception was thrown in onConnect()", e));
      }
      break;
    case IOMessage.TYPE_HEARTBEAT:
      sendPlain("2::");
      break;
    case IOMessage.TYPE_MESSAGE:
      try {
        findCallback(message).onMessage(message.getData(),
            remoteAcknowledge(message));
      } catch (Exception e) {
        error(new SocketIOException(
            "Exception was thrown in onMessage(String).\n"
                + "Message was: " + message.toString(), e));
      }
      break;
    case IOMessage.TYPE_JSON_MESSAGE:
      try {
        JsonElement obj = null;
        String data = message.getData();
        if (data.trim().equals("null") == false)
          obj = new JsonParser().parse(data);
        try {
          findCallback(message).onMessage(obj,
              remoteAcknowledge(message));
        } catch (Exception e) {
          error(new SocketIOException(
              "Exception was thrown in onMessage(JSONObject).\n"
                  + "Message was: " + message.toString(), e));
        }
      } catch (JsonParseException e) {
        logger.warning("Malformated JSON received");
      }
      break;
    case IOMessage.TYPE_EVENT:
      try {
        JsonObject event = new JsonParser().parse(message.getData()).getAsJsonObject();
        JsonElement[] argsArray;
        if (event.has("args")) {
          JsonArray args = event.getAsJsonArray("args");
          argsArray = new JsonElement[args.size()];
          for (int i = 0; i < args.size(); i++) {
            if (args.get(i) != null)
              argsArray[i] = args.get(i);
          }
        } else
          argsArray = new JsonElement[0];
        String eventName = event.get("name").getAsString();
        try {
          findCallback(message).on(eventName,
              remoteAcknowledge(message), argsArray);
        } catch (Exception e) {
          error(new SocketIOException(
              "Exception was thrown in on(String, JSONObject[]).\n"
                  + "Message was: " + message.toString(), e));
        }
      } catch (JsonParseException e) {
        logger.warning("Malformated JSON received");
      }
      break;

    case IOMessage.TYPE_ACK:
      String[] data = message.getData().split("\\+", 2);
      if (data.length == 2) {
        try {
          int id = Integer.parseInt(data[0]);
          AckEntry ackEntry = null;
          synchronized (acknowledge) {
            ackEntry = acknowledge.get(id);
            if (ackEntry != null) {
              ackEntry.ackTimeoutTask.cancel();
              acknowledge.remove(id);
            }
          }
          if (ackEntry != null) {
            JsonArray array = new JsonParser().parse(data[1]).getAsJsonArray();
            JsonElement[] args = new JsonElement[array.size()];
            for (int i = 0; i < args.length; i++) {
              args[i] = array.get(i);
            }
            ackEntry.ack.ack(args);
          }
          else
            logger.warning("Received unknown or timeouted ack packet");
          
        } catch (NumberFormatException e) {
          logger.warning("Received malformated Acknowledge! This is potentially filling up the acknowledges!");
        } catch (JsonParseException e) {
          logger.warning("Received malformated Acknowledge data!");
        }
      } else if (data.length == 1) {
        sendPlain("6:::" + data[0]);
      }
      break;
    case IOMessage.TYPE_ERROR:
      try {
        findCallback(message).onError(
            new SocketIOException(message.getData()));
      } catch (SocketIOException e) {
        error(e);
      }
      if (message.getData().endsWith("+0")) {
        // We are advised to disconnect
        cleanup();
      }
      break;
    case IOMessage.TYPE_NOOP:
      break;
    default:
      logger.warning("Unkown type received" + message.getType());
      break;
    }
  }

  /**
   * Flushes the buffer data.
   */
  private synchronized void flushBuffer() {
    if (transport.canSendBulk()) {
      ConcurrentLinkedQueue<String> outputBuffer = this.outputBuffer;
      this.outputBuffer = new ConcurrentLinkedQueue<String>();
      try {
        // DEBUG
        String[] texts = outputBuffer.toArray(new String[outputBuffer
            .size()]);
        logger.info("Bulk start:");
        for (String text : texts) {
          logger.info("> " + text);
        }
        logger.info("Bulk end");
        // DEBUG END
        transport.sendBulk(texts);
      } catch (IOException e) {
        this.outputBuffer = outputBuffer;
      }
    } else {
      String text;
      while ((text = outputBuffer.poll()) != null)
        sendPlain(text);
    }
  }

  /**
   * forces a reconnect. This had become useful on some android devices which
   * do not shut down TCP-connections when switching from HSDPA to Wifi
   */
  public synchronized void reconnect() {
    if (getState() != STATE_INVALID) {
      invalidateTransport();
      setState(STATE_INTERRUPTED);
      if (reconnectTask != null) {
        reconnectTask.cancel();
      }
      reconnectTask = new ReconnectTask();
      backgroundTimer.schedule(reconnectTask, 1000);
    }
  }

  /**
   * Returns the session id. This should be called from a {@link IOTransport}
   * 
   * @return the session id to connect to the right Session.
   */
  public String getSessionId() {
    return sessionId;
  }

  /**
   * sends a String message from {@link SocketIO} to the {@link IOTransport}.
   * 
   * @param socket
   *            the socket
   * @param ack
   *            acknowledge package which can be called from the server
   * @param text
   *            the text
   */
  public void send(SocketIO socket, IOAcknowledge ack, String text) {
    IOMessage message = new IOMessage(IOMessage.TYPE_MESSAGE,
        socket.getNamespace(), text);
    synthesizeAck(message, ack);
    sendPlain(message.toString());
  }

  /**
   * sends a JSON message from {@link SocketIO} to the {@link IOTransport}.
   * 
   * @param socket
   *            the socket
   * @param ack
   *            acknowledge package which can be called from the server
   * @param json
   *            the json
   */
  public void send(SocketIO socket, IOAcknowledge ack, JsonElement json) {
    IOMessage message = new IOMessage(IOMessage.TYPE_JSON_MESSAGE, socket.getNamespace(), json.toString());
    synthesizeAck(message, ack);
    sendPlain(message.toString());
  }

  /**
   * emits an event from {@link SocketIO} to the {@link IOTransport}.
   * 
   * @param socket
   *            the socket
   * @param event
   *            the event
   * @param ack
   *            acknowledge package which can be called from the server
   * @param args
   *            the arguments to be send
   */
  public void emit(SocketIO socket, String event, IOAcknowledge ack, Object... args) {
    try {
      JsonArray jarray = new JsonArray();

      for (Object arg : args) {
        if (arg instanceof JsonElement) {
          jarray.add((JsonElement)arg);
        } else if (arg instanceof String) {
          jarray.add(new JsonParser().parse((String)arg));
        } else {
          error(new SocketIOException("a non-json message received: " + arg.toString()));
        }
      }

      JsonObject jobj = new JsonObject();
      jobj.add("name", new JsonPrimitive(event));
      jobj.add("args", jarray);

      IOMessage message = new IOMessage(IOMessage.TYPE_EVENT, socket.getNamespace(), jobj.toString());
      synthesizeAck(message, ack);
      sendPlain(message.toString());

    } catch (JsonParseException e) {
      error(new SocketIOException("Error while emitting an event. Make sure you only try to send arguments, which can be serialized into JSON."));
    }

  }

  /**
   * Checks if IOConnection is currently connected.
   * 
   * @return true, if is connected
   */
  public boolean isConnected() {
    return getState() == STATE_READY;
  }

  /**
   * Gets the current state of this IOConnection.
   * 
   * @return current state
   */
  private synchronized int getState() {
    return state;
  }

  /**
   * Sets the current state of this IOConnection.
   * 
   * @param state
   *            the new state
   */
  private synchronized void setState(int state) {
    if (getState() != STATE_INVALID) {
      this.state = state;
    }
  }

  /**
   * gets the currently used transport.
   * 
   * @return currently used transport
   */
  public IOTransport getTransport() {
    return transport;
  }

  @Override
  public void onDisconnect() {
      for (SocketIO socket : sockets.values())
          socket.getCallback().onDisconnect();
  }

  @Override
  public void onConnect() {
      for (SocketIO socket : sockets.values())
          socket.getCallback().onConnect();
  }

  @Override
  public void onReconnect() {
      for (SocketIO socket : sockets.values())
          socket.getCallback().onReconnect();
  }

  @Override
  public void onMessage(String data, IOAcknowledge ack) {
    for (SocketIO socket : sockets.values())
      socket.getCallback().onMessage(data, ack);
  }

  @Override
  public void onMessage(JsonElement json, IOAcknowledge ack) {
    for (SocketIO socket : sockets.values())
      socket.getCallback().onMessage(json, ack);
  }

  @Override
  public void on(String event, IOAcknowledge ack, JsonElement... args) {
    for (SocketIO socket : sockets.values())
      socket.getCallback().on(event, ack, args);
  }

  @Override
  public void onError(SocketIOException socketIOException) {
    for (SocketIO socket : sockets.values())
      socket.getCallback().onError(socketIOException);
  }
}