modules/fdbworkers/src/flash/tools/debugger/concrete/DProtocol.java - flex-sdk - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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 flash.tools.debugger.concrete;

 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.Socket;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.util.EnumMap;

 import flash.tools.debugger.SessionManager;
 import flash.util.Trace;

 /**
  * Implements the lower portion of Flash Player debug protocol.  This class is able to
  * communicate with the Flash Player sending and receiving any and all messages neccessary
  * in order to continue a debug session with the Player.
  *
  * It does not understand the context of messages that it receives and merely provides
  * a channel for formatting and unformatting the messages.
  *
  * The messages are defined on the flash side in core/debugtags.h and handled in the
  * code under core/playerdebugger.cpp
  *
  * Messages that are received via this class are packaged in a DMessage and then
  * provided to any listeners if requested.   Filtering of incoming messages
  * at this level is not supported.
  */
 public class DProtocol implements Runnable
 {
 	public static final int DEBUG_PORT = 7935;

 	/* We connect to AIR in the case of AIR on Android over USB */
 	public static final int DEBUG_CONNECT_PORT = 7936;

 	private final BufferedInputStream	m_in;
 	private final BufferedOutputStream	m_out;
 	private final EnumMap<ListenerIndex, DProtocolNotifierIF> m_listeners; // WARNING: accessed from multiple threads
 	private long						m_msgRx;		// WARNING: accessed from multiple threads; use synchronized (this)
 	private long						m_msgTx;		// WARNING: accessed from multiple threads; use synchronized (this)
 	private volatile boolean			m_stopRx;		// WARNING: accessed from multiple threads
 	private volatile Thread				m_rxThread;		// WARNING: accessed from multiple threads
 	private volatile Exception m_disconnectCause;
 	private volatile Socket m_socket;
 	private boolean m_detectBrokenSocket;

 	public enum ListenerIndex
 	{
 		PlayerSession,

 		/**
 		 * The DMessageCounter must always be the LAST listener, so that the message has
 		 * been fully processed before we wake up any threads that were waiting until a
 		 * message comes in.
 		 */
 		MessageCounter
 	}

 	public DProtocol(BufferedInputStream in, BufferedOutputStream out)
 	{
 		m_in = in;
 		m_out = out;
 		m_listeners = new EnumMap<ListenerIndex, DProtocolNotifierIF>(ListenerIndex.class);
 		m_msgRx = 0;
 		m_msgTx = 0;
 		m_stopRx = false;
 		m_rxThread = null;
 		m_socket = null;
 		m_detectBrokenSocket = false;
 		// Create a message counter, which will listen to us for messages
 		addListener(ListenerIndex.MessageCounter, new DMessageCounter());
 	}

 	public DProtocol(BufferedInputStream in, BufferedOutputStream out,
 			Socket s, boolean detectBrokenSocket)
 	{
 		this(in, out);
 		m_socket = s;
 		m_detectBrokenSocket = detectBrokenSocket;
 	}


 	/**
 	 * Set the base socket options
 	 * @throws SocketException
 	 */
 	static void applyBaseSocketSettings(Socket s) throws SocketException
 	{
 		// For performance reasons, it is very important that we setTcpNoDelay(true),
 		// thus disabling Nagle's algorithm.  Google for TCP_NODELAY or Nagle
 		// for more information.
 		//
 		// In addition, we now use a BufferedOutputStream instead of an OutputStream.
 		//
 		// These changes result in a huge speedup on the Mac.
 		s.setTcpNoDelay(true);
 	}

 	static DProtocol createDProtocolFromSocket(Socket s, boolean detectBrokenSocket) throws IOException
 	{
 		BufferedInputStream in = new BufferedInputStream(s.getInputStream());
 		BufferedOutputStream out = new BufferedOutputStream(s.getOutputStream());

 		DProtocol dp = new DProtocol(in, out, s, detectBrokenSocket);
 		return dp;
 	}

 	/**
      * Build a DProtocol object from a the given socket connection.
      */
 	static DProtocol createFromSocket(Socket s) throws IOException
 	{
 		applyBaseSocketSettings(s);
 		return createDProtocolFromSocket(s, false);
 	}

 	/**
      * Build a DProtocol object from a the given socket connection
      * and applies socket specific settings set in SessionManager
      * like socket timeout.
      */
 	static DProtocol createFromSocket(Socket s, SessionManager sessionManager) throws IOException
 	{
 		applyBaseSocketSettings(s);
 		int socketTimeout = sessionManager.getPreference(SessionManager.PREF_SOCKET_TIMEOUT);
 		boolean checkSocket = false;
 		if (socketTimeout > 0)
 		{
 			s.setSoTimeout(socketTimeout);
 			checkSocket = true;
 		}
 		return createDProtocolFromSocket(s, checkSocket);
 	}

 	/**
 	 * Allow outside entities to listen for incoming DMessages.
 	 *
 	 * @param index
 	 *            the index of this listener. Listeners have a strictly defined
 	 *            order.
 	 * @param n
 	 *            the listener
 	 */
 	public boolean addListener(ListenerIndex index, DProtocolNotifierIF n)
 	{
 		synchronized (m_listeners)
 		{
 			m_listeners.put(index, n);
 		}
 		return true;
 	}

 	public long messagesReceived()		{ synchronized (this) { return m_msgRx; } }
 	public long messagesSent()			{ synchronized (this) { return m_msgTx; } }

 	/**
 	 * Entry point for our receive thread
 	 */
 	public void run()
 	{
 		try
 		{
 			m_stopRx = false;
 			listenForMessages();
 		}
 		catch(Exception ex)
 		{
 			m_disconnectCause = ex;
 			if (Trace.error &&
 				!(ex instanceof SocketException && ex.getMessage().equalsIgnoreCase("socket closed"))) // closed-socket is not an error //$NON-NLS-1$
 			{
 				ex.printStackTrace();
 			}
 		}

 		/* notify our listeners that we are no longer listening;  game over */
 		DProtocolNotifierIF[] listeners;
 		synchronized (m_listeners)
 		{
 			listeners = m_listeners.values().toArray(new DProtocolNotifierIF[m_listeners.size()]); // copy the list to avoid multithreading problems
 		}
 		for (int i=0; i<listeners.length; ++i)
 		{
 			DProtocolNotifierIF elem = listeners[i];
 			try
 			{
 				elem.disconnected();
 			}
 			catch(Exception exc) /* catch unchecked exceptions */
 			{
 				if (Trace.error)
 					exc.printStackTrace();
 			}
 		}

 		// final notice that this thread is dead!
 		m_rxThread = null;
 		m_socket = null;
 	}

 	/**
 	 * Create and start up a thread for our receiving messages.
 	 */
 	public boolean bind()
 	{
 		/* create a new thread object for us which just listens to incoming messages */
 		boolean worked = true;
 		if (m_rxThread == null)
 		{
 			getMessageCounter().clearInCounts();
 			getMessageCounter().clearOutCounts();

 			m_rxThread = new Thread(this, "DJAPI message listener"); //$NON-NLS-1$
 			m_rxThread.setDaemon(true);
 			m_rxThread.start();
 		}
 		else
 			worked = false;

 		return worked;
 	}

 	/**
 	 * Shutdown our receive thread
 	 */
 	public boolean unbind()
 	{
 		boolean worked = true;
 		if (m_rxThread == null)
 			worked = false;
 		else
 			m_stopRx = true;

 		return worked;
 	}

 	/**
 	 * Main rx loop which waits for commands and then issues them to anyone listening.
      */
 	void listenForMessages() throws IOException
 	{
 		DProtocolNotifierIF[] listeners = new DProtocolNotifierIF[0];

 		while(!m_stopRx)
 		{
 			/* read the data */
 			try
 			{
 				DMessage msg = rxMessage();

 				/* Now traverse our list of interested parties and let them deal with the message */
 				synchronized (m_listeners)
 				{
 					listeners = m_listeners.values().toArray(listeners); // copy the array to avoid multithreading problems
 				}
 				for (int i=0; i<listeners.length; ++i)
 				{
 					DProtocolNotifierIF elem = listeners[i];
 					try
 					{
 						elem.messageArrived(msg, this);
 					}
 					catch (Exception exc) /* catch unchecked exceptions */
 					{
 //						if (Trace.error)
 //						{
 							System.err.println("Error in listener parsing incoming message :"); //$NON-NLS-1$
 							System.err.println(msg.inToString(16));
 							exc.printStackTrace();
 //						}
 					}
 					msg.reset();  /* allow others to reparse the message */
 				}

 				/* now dispose with the message */
 				DMessageCache.free(msg);
 			}
 			catch(InterruptedIOException iio)
 			{
 				// this is a healthy exception that we simply ignore, since it means we haven't seen
 				// data for a while; is all.
 			}
 		}
 	}

 	/**
 	 * Transmit the message down the socket.
 	 *
 	 * This function is not synchronized; it is only called from one place, which is
 	 * PlayerSession.sendMessage().  That function is synchronized.
 	 */
 	void txMessage(DMessage message) throws IOException
 	{
 		int size = message.getSize();
 		int command = message.getType();

         //System.out.println("txMessage: " + DMessage.outTypeName(command) + " size=" + size);

         writeDWord(size);
 		writeDWord(command);
 		writeData(message.getData(), size);

 		m_out.flush();
 		synchronized (this) { m_msgTx++; }
 		getMessageCounter().messageSent(message);
 	}

 	class SendThread extends Thread {
 		public IOException exception = null;
 		public volatile boolean completed = false;

 		@Override
 		public void run() {
 			try {
 				DMessage dm = DMessageCache.alloc(4);
 				dm.setType(DMessage.OutSetSquelch);
 				dm.putDWord(1);
 				txMessage(dm);
 				DMessageCache.free(dm);
 				this.completed = true;
 			}
 			catch (IOException e) {
 				this.exception = e;
 			}
 		}
 	}
 	/**
      * Get the next message on the input stream, using the context contained within
      * the message itself to demark its end
      */
 	private DMessage rxMessage() throws IOException
 	{
         int size = -1;
 		int command = 0;

 		try
 		{
 			size = (int)readDWord();
 			command = (int)readDWord();
 		}
 		catch (SocketTimeoutException e)
 		{
 			if (!m_detectBrokenSocket)
 				throw e;
 			//schedule a simple message to be sent for
 			//heartbeat check
 			/**
 			 * Our logic kicks in after PREF_SOCKET_TIMEOUT
 			 * milliseconds to try and detect broken connection by writing
 			 * a squelch message to the player. If the write
 			 * succeeds, we assume everything is normal
 			 * (we don't wait for an ack). Otherwise, we save the error
 			 * that clients of FDB can use.
 			 *
 			 * On Mac, the write() blocks which is why it must
 			 * be done in a separate thread. The thread may take
 			 * upto five minutes to die even after interrupt().
 			 *
 			 * On Windows, the write() succeeds, but we later get
 			 * a recv abort.
 			 */
 			int oldBufferSize = -1;

 			if (m_socket != null) {
 				oldBufferSize = m_socket.getSendBufferSize();
 				m_socket.setSendBufferSize(1);
 			}

 			SendThread t = new SendThread();
 			t.start();
 			long waitBegin = System.currentTimeMillis();

 			while (true) {
 				try {
 					t.join(1000);
 					if (t.completed)
 						break;
 				} catch (InterruptedException e1) {
 					break;
 				}
 				long waitEnd = System.currentTimeMillis();
 				if (waitEnd - waitBegin > 10000)
 					break;
 			}
 			boolean success = true;
 			if (t.isAlive()) {
 				t.interrupt();
 				success = false;
 			}
 			if (oldBufferSize > 0) {
 				m_socket.setSendBufferSize(oldBufferSize);
 			}
 			if (!t.completed) {
 				success = false;
 			}
 			if (t.exception != null) {
 				throw t.exception;
 			}
 			if (success)
 				throw e;
 			else
 				throw new SocketException("Broken pipe"); //$NON-NLS-1$
 		}
         //System.out.println("rxMessage: " + DMessage.inTypeName(command) + " size=" + size);

 		if (size < 0)
 			throw new IOException("socket closed"); //$NON-NLS-1$

 //		if (DMessage.inTypeName(command).startsWith("InUnknown")) {
 //			System.out.println("Ignoring unknown message");
 //			size = 0;
 //		}

 		/**
 		 * Ask our message cache for a message
 		 */
 		DMessage message = DMessageCache.alloc(size);
 		byte[] messageContent = message.getData();
 		int offset = 0;

 		/* block until we get the entire message, which may come in pieces */
 		while (offset < size)
 			offset += m_in.read(messageContent, offset, size - offset);

 		/* now we have the data of the message, set its type and we are done */
 		message.setType(command);
 		synchronized (this) { m_msgRx++; }
 		return message;
 	}

 	void writeDWord(long dw) throws IOException
 	{
 		byte b0 = (byte)(dw & 0xff);
 		byte b1 = (byte)((dw >> 8) & 0xff);
 		byte b2 = (byte)((dw >> 16) & 0xff);
 		byte b3 = (byte)((dw >> 24) & 0xff);

 		m_out.write(b0);
 		m_out.write(b1);
 		m_out.write(b2);
 		m_out.write(b3);
 	}

 	void writeData(byte[] data, long size) throws IOException
 	{
 		if (size > 0)
 			m_out.write(data, 0, (int)size);
 	}


 	/**
 	 * Extract the next 4 bytes, which form a 32b integer, from the stream
 	 */
 	long readDWord() throws IOException
 	{
 		int b0 = m_in.read();
 		int b1 = m_in.read();
 		int b2 = m_in.read();
 		int b3 = m_in.read();

 		long value = ((b3 << 24) & 0xff000000) | ((b2 << 16) & 0xff0000) | ((b1 << 8) & 0xff00) | (b0 & 0xff);
 		return value;
 	}

 	public DMessageCounter getMessageCounter()
 	{
 		synchronized (m_listeners)
 		{
 			return (DMessageCounter) m_listeners.get(ListenerIndex.MessageCounter);
 		}
 	}

 	public Exception getDisconnectCause() {
 		return m_disconnectCause;
 	}

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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 flash.tools.debugger.concrete;

	import java.io.BufferedInputStream;
	import java.io.BufferedOutputStream;
	import java.io.IOException;
	import java.io.InterruptedIOException;
	import java.net.Socket;
	import java.net.SocketException;
	import java.net.SocketTimeoutException;
	import java.util.EnumMap;

	import flash.tools.debugger.SessionManager;
	import flash.util.Trace;

	/**
	* Implements the lower portion of Flash Player debug protocol. This class is able to
	* communicate with the Flash Player sending and receiving any and all messages neccessary
	* in order to continue a debug session with the Player.
	*
	* It does not understand the context of messages that it receives and merely provides
	* a channel for formatting and unformatting the messages.
	*
	* The messages are defined on the flash side in core/debugtags.h and handled in the
	* code under core/playerdebugger.cpp
	*
	* Messages that are received via this class are packaged in a DMessage and then
	* provided to any listeners if requested. Filtering of incoming messages
	* at this level is not supported.
	*/
	public class DProtocol implements Runnable
	{
	public static final int DEBUG_PORT = 7935;

	/* We connect to AIR in the case of AIR on Android over USB */
	public static final int DEBUG_CONNECT_PORT = 7936;

	private final BufferedInputStream m_in;
	private final BufferedOutputStream m_out;
	private final EnumMap<ListenerIndex, DProtocolNotifierIF> m_listeners; // WARNING: accessed from multiple threads
	private long m_msgRx; // WARNING: accessed from multiple threads; use synchronized (this)
	private long m_msgTx; // WARNING: accessed from multiple threads; use synchronized (this)
	private volatile boolean m_stopRx; // WARNING: accessed from multiple threads
	private volatile Thread m_rxThread; // WARNING: accessed from multiple threads
	private volatile Exception m_disconnectCause;
	private volatile Socket m_socket;
	private boolean m_detectBrokenSocket;

	public enum ListenerIndex
	{
	PlayerSession,

	/**
	* The DMessageCounter must always be the LAST listener, so that the message has
	* been fully processed before we wake up any threads that were waiting until a
	* message comes in.
	*/
	MessageCounter
	}

	public DProtocol(BufferedInputStream in, BufferedOutputStream out)
	{
	m_in = in;
	m_out = out;
	m_listeners = new EnumMap<ListenerIndex, DProtocolNotifierIF>(ListenerIndex.class);
	m_msgRx = 0;
	m_msgTx = 0;
	m_stopRx = false;
	m_rxThread = null;
	m_socket = null;
	m_detectBrokenSocket = false;
	// Create a message counter, which will listen to us for messages
	addListener(ListenerIndex.MessageCounter, new DMessageCounter());
	}

	public DProtocol(BufferedInputStream in, BufferedOutputStream out,
	Socket s, boolean detectBrokenSocket)
	{
	this(in, out);
	m_socket = s;
	m_detectBrokenSocket = detectBrokenSocket;
	}


	/**
	* Set the base socket options
	* @throws SocketException
	*/
	static void applyBaseSocketSettings(Socket s) throws SocketException
	{
	// For performance reasons, it is very important that we setTcpNoDelay(true),
	// thus disabling Nagle's algorithm. Google for TCP_NODELAY or Nagle
	// for more information.
	//
	// In addition, we now use a BufferedOutputStream instead of an OutputStream.
	//
	// These changes result in a huge speedup on the Mac.
	s.setTcpNoDelay(true);
	}

	static DProtocol createDProtocolFromSocket(Socket s, boolean detectBrokenSocket) throws IOException
	{
	BufferedInputStream in = new BufferedInputStream(s.getInputStream());
	BufferedOutputStream out = new BufferedOutputStream(s.getOutputStream());

	DProtocol dp = new DProtocol(in, out, s, detectBrokenSocket);
	return dp;
	}

	/**
	* Build a DProtocol object from a the given socket connection.
	*/
	static DProtocol createFromSocket(Socket s) throws IOException
	{
	applyBaseSocketSettings(s);
	return createDProtocolFromSocket(s, false);
	}

	/**
	* Build a DProtocol object from a the given socket connection
	* and applies socket specific settings set in SessionManager
	* like socket timeout.
	*/
	static DProtocol createFromSocket(Socket s, SessionManager sessionManager) throws IOException
	{
	applyBaseSocketSettings(s);
	int socketTimeout = sessionManager.getPreference(SessionManager.PREF_SOCKET_TIMEOUT);
	boolean checkSocket = false;
	if (socketTimeout > 0)
	{
	s.setSoTimeout(socketTimeout);
	checkSocket = true;
	}
	return createDProtocolFromSocket(s, checkSocket);
	}

	/**
	* Allow outside entities to listen for incoming DMessages.
	*
	* @param index
	* the index of this listener. Listeners have a strictly defined
	* order.
	* @param n
	* the listener
	*/
	public boolean addListener(ListenerIndex index, DProtocolNotifierIF n)
	{
	synchronized (m_listeners)
	{
	m_listeners.put(index, n);
	}
	return true;
	}

	public long messagesReceived() { synchronized (this) { return m_msgRx; } }
	public long messagesSent() { synchronized (this) { return m_msgTx; } }

	/**
	* Entry point for our receive thread
	*/
	public void run()
	{
	try
	{
	m_stopRx = false;
	listenForMessages();
	}
	catch(Exception ex)
	{
	m_disconnectCause = ex;
	if (Trace.error &&
	!(ex instanceof SocketException && ex.getMessage().equalsIgnoreCase("socket closed"))) // closed-socket is not an error //$NON-NLS-1$
	{
	ex.printStackTrace();
	}
	}

	/* notify our listeners that we are no longer listening; game over */
	DProtocolNotifierIF[] listeners;
	synchronized (m_listeners)
	{
	listeners = m_listeners.values().toArray(new DProtocolNotifierIF[m_listeners.size()]); // copy the list to avoid multithreading problems
	}
	for (int i=0; i<listeners.length; ++i)
	{
	DProtocolNotifierIF elem = listeners[i];
	try
	{
	elem.disconnected();
	}
	catch(Exception exc) /* catch unchecked exceptions */
	{
	if (Trace.error)
	exc.printStackTrace();
	}
	}

	// final notice that this thread is dead!
	m_rxThread = null;
	m_socket = null;
	}

	/**
	* Create and start up a thread for our receiving messages.
	*/
	public boolean bind()
	{
	/* create a new thread object for us which just listens to incoming messages */
	boolean worked = true;
	if (m_rxThread == null)
	{
	getMessageCounter().clearInCounts();
	getMessageCounter().clearOutCounts();

	m_rxThread = new Thread(this, "DJAPI message listener"); //$NON-NLS-1$
	m_rxThread.setDaemon(true);
	m_rxThread.start();
	}
	else
	worked = false;

	return worked;
	}

	/**
	* Shutdown our receive thread
	*/
	public boolean unbind()
	{
	boolean worked = true;
	if (m_rxThread == null)
	worked = false;
	else
	m_stopRx = true;

	return worked;
	}

	/**
	* Main rx loop which waits for commands and then issues them to anyone listening.
	*/
	void listenForMessages() throws IOException
	{
	DProtocolNotifierIF[] listeners = new DProtocolNotifierIF[0];

	while(!m_stopRx)
	{
	/* read the data */
	try
	{
	DMessage msg = rxMessage();

	/* Now traverse our list of interested parties and let them deal with the message */
	synchronized (m_listeners)
	{
	listeners = m_listeners.values().toArray(listeners); // copy the array to avoid multithreading problems
	}
	for (int i=0; i<listeners.length; ++i)
	{
	DProtocolNotifierIF elem = listeners[i];
	try
	{
	elem.messageArrived(msg, this);
	}
	catch (Exception exc) /* catch unchecked exceptions */
	{
	// if (Trace.error)
	// {
	System.err.println("Error in listener parsing incoming message :"); //$NON-NLS-1$
	System.err.println(msg.inToString(16));
	exc.printStackTrace();
	// }
	}
	msg.reset(); /* allow others to reparse the message */
	}

	/* now dispose with the message */
	DMessageCache.free(msg);
	}
	catch(InterruptedIOException iio)
	{
	// this is a healthy exception that we simply ignore, since it means we haven't seen
	// data for a while; is all.
	}
	}
	}

	/**
	* Transmit the message down the socket.
	*
	* This function is not synchronized; it is only called from one place, which is
	* PlayerSession.sendMessage(). That function is synchronized.
	*/
	void txMessage(DMessage message) throws IOException
	{
	int size = message.getSize();
	int command = message.getType();

	//System.out.println("txMessage: " + DMessage.outTypeName(command) + " size=" + size);

	writeDWord(size);
	writeDWord(command);
	writeData(message.getData(), size);

	m_out.flush();
	synchronized (this) { m_msgTx++; }
	getMessageCounter().messageSent(message);
	}

	class SendThread extends Thread {
	public IOException exception = null;
	public volatile boolean completed = false;

	@Override
	public void run() {
	try {
	DMessage dm = DMessageCache.alloc(4);
	dm.setType(DMessage.OutSetSquelch);
	dm.putDWord(1);
	txMessage(dm);
	DMessageCache.free(dm);
	this.completed = true;
	}
	catch (IOException e) {
	this.exception = e;
	}
	}
	}
	/**
	* Get the next message on the input stream, using the context contained within
	* the message itself to demark its end
	*/
	private DMessage rxMessage() throws IOException
	{
	int size = -1;
	int command = 0;

	try
	{
	size = (int)readDWord();
	command = (int)readDWord();
	}
	catch (SocketTimeoutException e)
	{
	if (!m_detectBrokenSocket)
	throw e;
	//schedule a simple message to be sent for
	//heartbeat check
	/**
	* Our logic kicks in after PREF_SOCKET_TIMEOUT
	* milliseconds to try and detect broken connection by writing
	* a squelch message to the player. If the write
	* succeeds, we assume everything is normal
	* (we don't wait for an ack). Otherwise, we save the error
	* that clients of FDB can use.
	*
	* On Mac, the write() blocks which is why it must
	* be done in a separate thread. The thread may take
	* upto five minutes to die even after interrupt().
	*
	* On Windows, the write() succeeds, but we later get
	* a recv abort.
	*/
	int oldBufferSize = -1;

	if (m_socket != null) {
	oldBufferSize = m_socket.getSendBufferSize();
	m_socket.setSendBufferSize(1);
	}

	SendThread t = new SendThread();
	t.start();
	long waitBegin = System.currentTimeMillis();

	while (true) {
	try {
	t.join(1000);
	if (t.completed)
	break;
	} catch (InterruptedException e1) {
	break;
	}
	long waitEnd = System.currentTimeMillis();
	if (waitEnd - waitBegin > 10000)
	break;
	}
	boolean success = true;
	if (t.isAlive()) {
	t.interrupt();
	success = false;
	}
	if (oldBufferSize > 0) {
	m_socket.setSendBufferSize(oldBufferSize);
	}
	if (!t.completed) {
	success = false;
	}
	if (t.exception != null) {
	throw t.exception;
	}
	if (success)
	throw e;
	else
	throw new SocketException("Broken pipe"); //$NON-NLS-1$
	}
	//System.out.println("rxMessage: " + DMessage.inTypeName(command) + " size=" + size);

	if (size < 0)
	throw new IOException("socket closed"); //$NON-NLS-1$

	// if (DMessage.inTypeName(command).startsWith("InUnknown")) {
	// System.out.println("Ignoring unknown message");
	// size = 0;
	// }

	/**
	* Ask our message cache for a message
	*/
	DMessage message = DMessageCache.alloc(size);
	byte[] messageContent = message.getData();
	int offset = 0;

	/* block until we get the entire message, which may come in pieces */
	while (offset < size)
	offset += m_in.read(messageContent, offset, size - offset);

	/* now we have the data of the message, set its type and we are done */
	message.setType(command);
	synchronized (this) { m_msgRx++; }
	return message;
	}

	void writeDWord(long dw) throws IOException
	{
	byte b0 = (byte)(dw & 0xff);
	byte b1 = (byte)((dw >> 8) & 0xff);
	byte b2 = (byte)((dw >> 16) & 0xff);
	byte b3 = (byte)((dw >> 24) & 0xff);

	m_out.write(b0);
	m_out.write(b1);
	m_out.write(b2);
	m_out.write(b3);
	}

	void writeData(byte[] data, long size) throws IOException
	{
	if (size > 0)
	m_out.write(data, 0, (int)size);
	}


	/**
	* Extract the next 4 bytes, which form a 32b integer, from the stream
	*/
	long readDWord() throws IOException
	{
	int b0 = m_in.read();
	int b1 = m_in.read();
	int b2 = m_in.read();
	int b3 = m_in.read();

	long value = ((b3 << 24) & 0xff000000) \| ((b2 << 16) & 0xff0000) \| ((b1 << 8) & 0xff00) \| (b0 & 0xff);
	return value;
	}

	public DMessageCounter getMessageCounter()
	{
	synchronized (m_listeners)
	{
	return (DMessageCounter) m_listeners.get(ListenerIndex.MessageCounter);
	}
	}

	public Exception getDisconnectCause() {
	return m_disconnectCause;
	}

	}