[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;
               }
       }
}


-- 
Mvh/Anri