[xSocket-develop] Client sometimes not receiving data from Server.

[Anri D. <anr...@gm...>]

Hi,

First off, I love xSocket so far. But I've run into a problem when
writing tests for my game server.

I've created a simple test using an echo server and 10
NonBlockingConnection clients.

All works well when the server's onData handler method returns fairly quickly.
But if I put something CPU intensive inside the onData method, to
simulate some sort of processing on the server (CPU bound, no I/O
atm), some of the clients never receive the server echo.

I'm using default settings (server handler is multi-threaded, and
flush mode sync) and I suspect Chapter 11 in the tutorial warns about
this, but I don't quite get it.

Could you please explain what I do wrong ? Thanks a bunch.

Source as follows:

import java.io.IOException;
import java.nio.BufferUnderflowException;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.ConcurrentHashMap;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xsocket.MaxReadSizeExceededException;
import org.xsocket.connection.IConnectHandler;
import org.xsocket.connection.IDataHandler;
import org.xsocket.connection.IDisconnectHandler;
import org.xsocket.connection.IHandler;
import org.xsocket.connection.INonBlockingConnection;
import org.xsocket.connection.IServer;
import org.xsocket.connection.NonBlockingConnection;
import org.xsocket.connection.Server;

public class TestXSocket
{
	public static Logger		log					= LoggerFactory.getLogger(TestXSocket.class);
	public static final String	PROTOCOL_DELIMITER	= "\0";

	@Test
	public void testClientServerCom()
	{
		IServer serv = null;

		try
		{
			log.info("Starting test server..");
			serv = new Server("127.0.0.1", 10101, new ServerHandler());
			serv.start();
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}

		ConcurrentHashMap<String, INonBlockingConnection> clients = new
ConcurrentHashMap<String, INonBlockingConnection>();

		for (int i = 1; i <= 10; i++)
		{
			try
			{
				INonBlockingConnection con = new
NonBlockingConnection(serv.getLocalAddress(),
						10101, new ClientHandler(i), true, 1000);
				
				con.write("Hey Dude #" + i + TestXSocket.PROTOCOL_DELIMITER);

				clients.put(con.getId(), con);
			}
			catch (IOException e)
			{
				e.printStackTrace();
				break;
			}
		}

		while (clients.size() > 0)
		{
			for (String conId : clients.keySet())
			{
				INonBlockingConnection con = clients.get(conId);
				ClientHandler handler = (ClientHandler) con.getHandler();

				if (handler.hasGotEcho())
				{
					log.info("Removing client {}", handler.getClientId());
					clients.remove(conId);
				}
				else
				{
					log.info("Client {} still waiting for echo..", handler.getClientId());
				}
			}

			try
			{
				Thread.sleep(2000);
			}
			catch (Exception e)
			{
				log.info("Wait interrupted!");
				break;
			}
		}

		try
		{
			serv.close();
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
	}

	private class ServerHandler implements IHandler, IConnectHandler, IDataHandler,
			IDisconnectHandler
	{

		@Override
		public boolean onConnect(INonBlockingConnection arg0) throws IOException,
				BufferUnderflowException, MaxReadSizeExceededException
		{
			log.info("Server Connected {}", arg0.getId());

			arg0.write("Welcome " + arg0.getId() + TestXSocket.PROTOCOL_DELIMITER);
			return true;
		}

		@Override
		public boolean onData(INonBlockingConnection arg0) throws IOException,
				BufferUnderflowException, ClosedChannelException,
MaxReadSizeExceededException
		{
			String read = arg0.readStringByDelimiter(TestXSocket.PROTOCOL_DELIMITER);
			log.info("Server Data Received {} : {}", arg0.getId(), read);

			StringBuffer sb = new StringBuffer();
			for (int i = 0; i < 1000000; i++)
			{
				sb.append(i);
			}
			sb.setLength(0);

			arg0.write("Echo " + read + TestXSocket.PROTOCOL_DELIMITER);

			log.info("Server Sent Echo : {}", read);

			return true;
		}

		@Override
		public boolean onDisconnect(INonBlockingConnection arg0) throws IOException
		{
			log.info("Server Disconnected {}", arg0.getId());

			arg0.write("Bye " + arg0.getId() + TestXSocket.PROTOCOL_DELIMITER);

			return true;
		}
	}

	private class ClientHandler implements IConnectHandler, IDataHandler,
IDisconnectHandler
	{
		private int		clientId	= 0;
		private boolean	gotEcho		= false;

		public ClientHandler(int clientId)
		{
			this.clientId = clientId;
		}

		public int getClientId()
		{
			return clientId;
		}

		public boolean hasGotEcho()
		{
			return gotEcho;
		}

		public void setGotEcho(boolean gotEcho)
		{
			this.gotEcho = gotEcho;
		}

		@Override
		public boolean onConnect(INonBlockingConnection arg0) throws IOException,
				BufferUnderflowException, MaxReadSizeExceededException
		{
			log.info("Client Connected {}", clientId);
			return false;
		}

		@Override
		public boolean onData(INonBlockingConnection arg0) throws IOException,
				BufferUnderflowException, ClosedChannelException,
MaxReadSizeExceededException
		{
			String read = arg0.readStringByDelimiter(TestXSocket.PROTOCOL_DELIMITER);

			log.info("Client Data Received {} : {}", clientId, read);

			if (read.contains("Echo "))
			{
				setGotEcho(true);
			}

			return false;
		}

		@Override
		public boolean onDisconnect(INonBlockingConnection arg0) throws IOException
		{
			log.info("Client Disconnected {}", clientId);
			return false;
		}
	}
}