Re: [xSocket-develop] xSocket (multiplexed) and serialized objects

SourceForge Headquarters 1320 Columbia Street Suite 310 San Diego, CA 92101 +1 (858) 422-6466

I don't think I'm really using multiplexing, but I did start with the multiplexed server 2.0b1 code.
I don't think about indexing the openConnections in the Handler because I failed to mention they are IConnectionScoped

However, I have a singleton that does basically the same thing as you have illustrated.

Psuedocode of my serverside network architecture:

ServerHandler serverProtocolHandler = new ServerHandler();
ServerHandler.addMyEventListener(Provider.getSingletonController()); // It has implemented IMyListener
...
ConnectionUtils.start(server);
---

ServerHandler implements
            IConnectHandler, IDisconnectHandler,
            IDataHandler, IIdleTimeoutHandler,
            IConnectionTimeoutHandler, IConnectionScoped

            boolean authorized = false;
            INonBlockingConnection connection; // For disconnect method. Not copied in this.copy().

onConnection
            connection.setIdleTimeoutMillis(30 * 60 * 1000);

onData
            this.connection = connection;
            try{
                        byte[] readBytes = connection.readBytesByDelimiter(Constants.NETWORK_DELIMITER,length);
                        ByteArrayInputStream bis = new ByteArrayInputStream(readBytes);
                        ObjectInputStream incomingMyEventStream = new ObjectInputStream(bis);
                        incomingMyEvent = incomingMyEventStream.readObject();

                        if(this.authorized)
                        {
                                    if(incomingMyEvent.getType() > 0) // Double check object is good by using a method
                                                fireEvent((castBasedOnSwitch)incomingMyEvent)
                        }
                        else // Must be initial message. Better be a login.
                        {
                                    ((LoginMyEvent)incomingMyEvent).setConnection(connection);
                                    ((LoginMyEvent)incomingMyEvent).setConnectionHandler(this);
                                    fireEvent(incomingMyEvent);
                        }
            catch(Exception e)
            {
                        this.disconnect(); // Contains try{}catch{} for disconnect
            }

---

SingletonController

HashMap eventCommandMap;

void handle(MyEvent event)
            eventCommandMap.get(Integer.valueOf(event.getType())).execute(event);

---

LoginCommand

void execute(MyEvent event)
            // Database lookup or whatever you want to store accounts in.
            int id = lookup(  ((LoginMyEvent)event).getUserName(),          ((LoginMyEvent)event).getPassword()  );

            // however you want to check it, > -1 or != 0 or whatever. Also can create persistent serverside object in the validate(int).
            if(validate(id))
            ){
                        // For indexing serverside persistent object->connection together. See addValidConnection(LoginMyEvent)
                        ((LoginMyEvent)event).setId(id);
                        ((LoginMyEvent)event).getConnectionHandler().setAuthorized(true);

                        // ApplicationModel is application state data and accessors only.
                        Provider.getSingletonController().getApplicationModel().addValidConnection((LoginMyEvent)event);
            }
            else
            {
                        ((LoginMyEvent)event).getConnectionHandler().disconnect();
            }

---

ApplicationModel:

Set<Integer,INonBlockingConnection> validConnections;

void addValidConnection(LoginMyEvent loginMyEvent)
            synchronized(validConnections)
                        validConnections.put(Integer.valueOf(loginMyEvent.getId()),loginMyEvent.getConnection());

---

The client uses the standard NIO client implementation found at http://rox-xmlrpc.sourceforge.net/niotut/

Both client and Server have GUIs that an event-driven MVC with the singleton controller and an extra presentation layer for where SWING falls down.

I'm not sure if this is entirely clear but I've tried to describe the process as succinctly as possible.

Now I can create a severside persistent object and always lookup its connection for message sending whenever the server needs to send one.
If the client sends a message the handler will validate and throw it to my application controller for proper handling.

I'm not sure if there are synchronization issues I have yet to discover, from this, but in my simple unit tests it looks like it's working ^^

________________________________
From: Gregor Roth [mailto:gre...@go...]
Sent: Thursday, May 29, 2008 11:13 PM
To: xSo...@li...
Cc: Jeff Crane
Subject: Re: [xSocket-develop] xSocket (multiplexed) and serialized objects

Hi Jeff,

by implementing the onConnect and onDisconnect method you are able to manage the open connections. See enclosed example. Does this answer your question?

Gregor

private static final class Test implements IDataHandler, IConnectHandler, IDisconnectHandler {

    private final HashMap<String, INonBlockingConnection> openConnections = new HashMap<String,INonBlockingConnection>();

    public boolean onConnect(INonBlockingConnection connection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
        connection.setIdleTimeoutMillis(30 * 60 * 1000); // if not data is be received within 30 min, the connection will be closed

        synchronized (openConnections) {
            openConnections.put(connection.getId(), connection);
        }
        return true;
    }

    public boolean onDisconnect(INonBlockingConnection connection) throws IOException {
        synchronized (openConnections) {
            openConnections.remove(connection.getId());
        }
        return true;
    }

    public boolean onData(INonBlockingConnection connection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {

         ...
        data = connection.read...
        ...
        send(...);

        return true;
    }

    // will be called concurrently
    private void send(Object message) {

        HashMap<String, INonBlockingConnection> openConnectionsCopy = null;
        synchronized (openConnections) {
            openConnectionsCopy = (HashMap<String, INonBlockingConnection>) openConnections.clone();
        }

        for (INonBlockingConnection connection : openConnectionsCopy.values()) {
            try {
                synchronized (connection) {  // connection methods are not threadsafe
                    connection.write...
                    ...
                }
            } catch (IOException ioe) {
                synchronized (openConnections) {
                    openConnections.remove(connection.getId());
                }
            }
        }
    }
}

Hi Jeff,

by implementing the onConnect and onDisconnect method you are able to manage the open connections. See enclosed example. Does this answers your question?

Gregor

    private static final class Test implements IDataHandler, IConnectHandler, IDisconnectHandler {

        private final HashMap<String, INonBlockingConnection> openConnections = new HashMap<String, INonBlockingConnection>();

        public boolean onConnect(INonBlockingConnection connection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
            connection.setIdleTimeoutMillis(30 * 60 * 1000); // if not data is be received within 30 min, the connection will be closed

            synchronized (openConnections) {
                openConnections.put(connection.getId(), connection);
            }
            return true;
        }

        public boolean onDisconnect(INonBlockingConnection connection) throws IOException {
            synchronized (openConnections) {
                openConnections.remove(connection.getId());
            }
            return true;
        }

        public boolean onData(INonBlockingConnection connection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {

            ...
            data = connection.read...
            ...
            send(...);

            return true;
        }

        // will be called concurrently
        private void send(Object message) {

            HashMap<String, INonBlockingConnection> openConnectionsCopy = null;
            synchronized (openConnections) {
                openConnectionsCopy = (HashMap<String, INonBlockingConnection>) openConnections.clone();
            }

            for (INonBlockingConnection connection : openConnectionsCopy.values()) {
                try {
                    synchronized (connection) {  // connection methods are not threadsafe
                        connection.write...
                        ...
                    }
                } catch (IOException ioe) {
                    synchronized (openConnections) {
                        openConnections.remove(connection.getId());
                    }
                }
            }
        }
    }

2008/5/29 Jeff Crane <JC...@sp...<mailto:JC...@sp...>>:

I read and implemented the answer to my first question. ConnectionUtil held the function and usage.

Clientside:

client = new NonBlockingConnection(...);

ByteArrayOutputStream bos = new ByteArrayOutputStream();

ObjectOutputStream out = new ObjectOutputStream(bos);

out.writeObject(genericObj);

out.writeObject(DELIMITER);

client.write(bos.size());

client.write(bos.toByteArray());

Server Protocol Handler in onData(...)

int length = ConnectionUtils.validateSufficientDatasizeByIntLengthField(connection);

byte[] test = connection.readBytesByDelimiter(DELIMITER,length);

ByteArrayInputStream bis = new ByteArrayInputStream(test);

ObjectInputStream incomingObjstream = new ObjectInputStream(bis);

Object incomingObj = null;

try{

            incomingObj = incomingObjstream.readObject();

}

catch(ClassNotFoundException e){e.printStackTrace();

}

incomingObjstream.close();

2. What is the standard practice to index open connections for sending? I use my protocol handler to read the object, handle the object, throw an event. If the server wants to disconnect or send a message, how do I do a lookup? I could do a singleton HashMap or pass a reference in the event thrown (which doesn't help if it's interacting with another connection) or am I missing something in the xSocket interfaces?

-------------------------------------------------------------------------
This SF.net email is sponsored by: Microsoft
Defy all challenges. Microsoft(R) Visual Studio 2008.
http://clk.atdmt.com/MRT/go/vse0120000070mrt/direct/01/
_______________________________________________
xSocket-develop mailing list
xSo...@li...<mailto:xSo...@li...>
https://lists.sourceforge.net/lists/listinfo/xsocket-develop