cvs commit: spice/sandbox/mesnet/src/java/org/realityforge/mesnet DataChunk.java SessionImpl.java TcpTransport.java MesnetSelectorEventHandler.java MessageUtils.java Session.java
Peter Donald <[email protected]> Sun, 09 Nov 2003 21:54:18 -0800
| Newsgroups | gmane.comp.java.spice.cvs |
|---|---|
| Message-ID | <[email protected]> |
donaldp 03/11/09 21:54:18
Modified: sandbox/mesnet/src/java/org/realityforge/mesnet
MesnetSelectorEventHandler.java MessageUtils.java
Session.java
Added: sandbox/mesnet/src/java/org/realityforge/mesnet
DataChunk.java SessionImpl.java TcpTransport.java
Log:
Checkin junk code and fix later
Revision Changes Path
1.2 +179 -35 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/MesnetSelectorEventHandler.java
Index: MesnetSelectorEventHandler.java
===================================================================
RCS file: /cvsroot/spice/spice/sandbox/mesnet/src/java/org/realityforge/mesnet/MesnetSelectorEventHandler.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- MesnetSelectorEventHandler.java 5 Nov 2003 04:08:36 -0000 1.1
+++ MesnetSelectorEventHandler.java 10 Nov 2003 05:54:18 -0000 1.2
@@ -2,47 +2,210 @@
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.BufferOverflowException;
import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import org.realityforge.sca.selector.SelectorEventHandler;
+import org.realityforge.sca.selector.SelectorManager;
/**
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
- * @version $Revision: 1.1 $ $Date: 2003/11/05 04:08:36 $
+ * @version $Revision: 1.2 $ $Date: 2003/11/10 05:54:18 $
*/
public class MesnetSelectorEventHandler
implements SelectorEventHandler
{
+ private static final byte[] MAGIC = new byte[]
+ {
+ 'm', 'a', 'g', 'i', 'c', '0', '1'
+ };
+ private static final int MAX_CONTROL_SIZE = MAGIC.length + 8 + 8;
+
+ private final ByteBuffer _message =
+ ByteBuffer.allocateDirect( MAX_CONTROL_SIZE );
+
+ private final SelectorManager _selectorManager;
+ private final SessionManager _sessionManager;
+
+ public MesnetSelectorEventHandler( final SelectorManager selectorManager,
+ final SessionManager sessionManager )
+ {
+ _selectorManager = selectorManager;
+ _sessionManager = sessionManager;
+ }
+
public void handleSelectorEvent( final SelectionKey key,
final Object userData )
{
- final Session session = (Session)userData;
- handleSessionEvent( key, session );
+ if( key.isAcceptable() )
+ {
+ performAccept( userData, key );
+ }
+ else
+ {
+ final TcpTransport transport = (TcpTransport)userData;
+ final Session session = transport.getSession();
+ if( null == session )
+ {
+ handleSessionlessEvent( transport );
+ }
+ else
+ {
+ handleSessionEvent( transport );
+ }
+ }
}
- public void handleSessionEvent( final SelectionKey key,
- final Session session )
+ void handleSessionlessEvent( final TcpTransport transport )
{
+ if( transport.getKey().isReadable() )
+ {
+ final SocketChannel channel = transport.getChannel();
+ try
+ {
+ final int count = channel.socket().getInputStream().available();
+ if( count >= MAX_CONTROL_SIZE )
+ {
+ _message.clear();
+ channel.read( _message );
+ for( int i = 0; i < MAGIC.length; i++ )
+ {
+ final byte data = _message.get();
+ if( data != MAGIC[ i ] )
+ {
+ //TODO: Notify about bad magic number
+ sendControlMessage( channel,
+ MessageUtils.MESSAGE_BAD_MAGIC );
+ transport.close();
+ return;
+ }
+ }
+ final long id = _message.getLong();
+ final long auth = _message.getLong();
+ if( id == -1 )
+ {
+ final Session session = _sessionManager.newSession();
+ session.setStatus( Session.STATUS_CONNECTED );
+
+ _message.putShort( (short)(9 * -1) ); //Size of message & fact it is control
+ _message.put( MessageUtils.MESSAGE_CONNECTED ); //message
+ _message.putLong( id );
+ _message.putLong( auth );
+ writeMessage( _message, channel );
+ }
+ else
+ {
+ final Session session = _sessionManager.findSession( id );
+ if( null == session )
+ {
+ sendControlMessage( channel,
+ MessageUtils.MESSAGE_BAD_SESSION );
+ transport.close();
+ return;
+ }
+
+ final Long sessionAuth = (Long)
+ session.getProperty( MessageUtils.AUTH_KEY );
+ if( null != sessionAuth &&
+ auth != sessionAuth.longValue() )
+ {
+ sendControlMessage( channel,
+ MessageUtils.MESSAGE_BAD_AUTH );
+ transport.close();
+ return;
+ }
+ else
+ {
+ session.setStatus( Session.STATUS_ESTABLISHED );
+ transport.setSession( session );
+
+ sendControlMessage( channel,
+ MessageUtils.MESSAGE_ESTABLISHED );
+ }
+ }
+ }
+ }
+ catch( final BufferOverflowException boe )
+ {
+ //TODO: Notify about buffer overflow
+ transport.close();
+ }
+ catch( final IOException ioe )
+ {
+ //TODO: Notify about esablishing connection
+ transport.close();
+ }
+ }
+ }
+
+ private void sendControlMessage( final SocketChannel channel,
+ final byte message )
+ throws IOException
+ {
+ _message.clear();
+ _message.putShort( (short)(1 * -1) );
+ _message.put( message );
+ writeMessage( _message, channel );
+ }
+
+ private void writeMessage( final ByteBuffer message,
+ final SocketChannel channel )
+ throws IOException
+ {
+ final int available = message.remaining();
+ final int writeCount = channel.write( message );
+ if( available != writeCount )
+ {
+ throw new BufferOverflowException();
+ }
+ }
+
+ void performAccept( final Object userData,
+ final SelectionKey key )
+ {
+ final ServerSocketChannel channel = (ServerSocketChannel)key.channel();
+ try
+ {
+ final SocketChannel socketChannel = channel.accept();
+ _selectorManager.registerChannel( socketChannel,
+ SelectionKey.OP_READ |
+ SelectionKey.OP_WRITE,
+ this,
+ userData );
+ }
+ catch( final IOException e )
+ {
+ //TODO: Note that problem with accept
+ }
+ }
+
+ public void handleSessionEvent( final TcpTransport transport )
+ {
+ final SelectionKey key = transport.getKey();
if( key.isReadable() )
{
- performRead( session );
+ performRead( transport );
}
if( key.isValid() && key.isWritable() )
{
- performWrite( session );
+ performWrite( transport );
}
}
- void performRead( final Session session )
+ void performRead( final TcpTransport transport )
{
try
{
- final ByteBuffer buffer = session.getReadBuffer();
- final int count = session.getChannel().read( buffer );
+ final DataChannel dataChannel =
+ transport.getSession().getDataChannel();
+ final ByteBuffer buffer = dataChannel.getReadBuffer();
+ final int count = transport.getChannel().read( buffer );
if( -1 == count )
{
- endConnection( session );
+ transport.close();
return;
}
//TODO: Notify Monitor that data received
@@ -67,14 +230,14 @@
}
catch( final IOException e )
{
- endConnection( session );
- return;
+ transport.close();
}
}
- void performWrite( final Session session )
+ void performWrite( final TcpTransport transport )
{
- final LinkedList queue = session.getWriteQueue();
+ final ByteBuffer writeBuffer =
+ transport.getSession().getDataChannel().getWriteBuffer();
synchronized( queue )
{
if( queue.size() > 0 )
@@ -99,28 +262,9 @@
catch( final IOException e )
{
queue.addFirst( data );
- endConnection( session );
+ transport.close();
}
}
}
}
-
- void endConnection( final Session session )
- {
- //TODO: Notify ending connection
- session.getKey().cancel();
- try
- {
- session.getChannel().close();
- }
- catch( IOException e )
- {
- // TODO: Note error closing channel
- }
- session.setKey( null );
- session.setChannel( null );
- //TODO: Attempt to resestablish connection with client
- //or wait till they connect?
- }
-
}
1.2 +2 -1 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/MessageUtils.java
Index: MessageUtils.java
===================================================================
RCS file: /cvsroot/spice/spice/sandbox/mesnet/src/java/org/realityforge/mesnet/MessageUtils.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- MessageUtils.java 10 Nov 2003 04:13:52 -0000 1.1
+++ MessageUtils.java 10 Nov 2003 05:54:18 -0000 1.2
@@ -4,7 +4,7 @@
/**
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
- * @version $Revision: 1.1 $ $Date: 2003/11/10 04:13:52 $
+ * @version $Revision: 1.2 $ $Date: 2003/11/10 05:54:18 $
*/
public class MessageUtils
{
@@ -12,6 +12,7 @@
public static final byte MESSAGE_ESTABLISHED = 1;
public static final byte MESSAGE_BAD_SESSION = 2;
public static final byte MESSAGE_BAD_AUTH = 3;
+ public static final byte MESSAGE_BAD_MAGIC = 4;
public static final String AUTH_KEY = "msg.auth";
}
1.3 +21 -98 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/Session.java
Index: Session.java
===================================================================
RCS file: /cvsroot/spice/spice/sandbox/mesnet/src/java/org/realityforge/mesnet/Session.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- Session.java 5 Nov 2003 23:15:28 -0000 1.2
+++ Session.java 10 Nov 2003 05:54:18 -0000 1.3
@@ -1,105 +1,28 @@
package org.realityforge.mesnet;
-import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-import java.util.LinkedList;
-
/**
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
- * @version $Revision: 1.2 $ $Date: 2003/11/05 23:15:28 $
+ * @version $Revision: 1.3 $ $Date: 2003/11/10 05:54:18 $
*/
-public class Session
+public interface Session
{
- private final LinkedList _readQueue = new LinkedList();
- private final LinkedList _writeQueue = new LinkedList();
- private final int _sessionID;
- private final int _sessionAuth;
- private final ByteBuffer _readBuffer;
- private final ByteBuffer _writeBuffer;
- private SocketChannel _channel;
- private SelectionKey _key;
- private long _lastReadTime;
- private long _lastWriteTime;
-
- public Session( final int sessionID,
- final int sessionAuth,
- final int maxReadSize,
- final int maxWriteSize )
- {
- _sessionID = sessionID;
- _sessionAuth = sessionAuth;
- _readBuffer = ByteBuffer.allocateDirect( maxReadSize );
- _writeBuffer = ByteBuffer.allocateDirect( maxWriteSize );
- }
-
- public SocketChannel getChannel()
- {
- return _channel;
- }
-
- public void setChannel( final SocketChannel channel )
- {
- _channel = channel;
- }
-
- public SelectionKey getKey()
- {
- return _key;
- }
-
- public void setKey( final SelectionKey key )
- {
- _key = key;
- }
-
- public long getLastReadTime()
- {
- return _lastReadTime;
- }
-
- public void setLastReadTime( final long lastReadTime )
- {
- _lastReadTime = lastReadTime;
- }
-
- public long getLastWriteTime()
- {
- return _lastWriteTime;
- }
-
- public void setLastWriteTime( final long lastWriteTime )
- {
- _lastWriteTime = lastWriteTime;
- }
-
- public int getSessionID()
- {
- return _sessionID;
- }
-
- public int getSessionAuth()
- {
- return _sessionAuth;
- }
-
- public ByteBuffer getReadBuffer()
- {
- return _readBuffer;
- }
-
- public ByteBuffer getWriteBuffer()
- {
- return _writeBuffer;
- }
-
- public LinkedList getWriteQueue()
- {
- return _writeQueue;
- }
-
- public LinkedList getReadQueue()
- {
- return _readQueue;
- }
+ int STATUS_NOT_CONNECTED = 0;
+ int STATUS_CONNECTED = 1;
+ int STATUS_ESTABLISHED = 2;
+ int STATUS_LOST = 3;
+ int STATUS_DISCONNECTED = 4;
+
+ long getSessionID();
+
+ int getStatus();
+
+ long getTimeOfLastStatusChange();
+
+ void setStatus( int status );
+
+ Object getProperty( String key );
+
+ void setProperty( String key, Object value );
+
+ DataChannel getDataChannel();
}
1.1 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/DataChunk.java
Index: DataChunk.java
===================================================================
package org.realityforge.mesnet;
import java.nio.ByteBuffer;
/**
*
*
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
* @version $Revision: 1.1 $ $Date: 2003/11/10 05:54:18 $
*/
public class DataChunk
{
private final short _sequence;
private final ByteBuffer _data;
public DataChunk( final short sequence,
final ByteBuffer data )
{
_sequence = sequence;
_data = data;
}
public short getSequence()
{
return _sequence;
}
public ByteBuffer getData()
{
return _data;
}
}
1.1 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/SessionImpl.java
Index: SessionImpl.java
===================================================================
package org.realityforge.mesnet;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
/**
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
* @version $Revision: 1.1 $ $Date: 2003/11/10 05:54:18 $
*/
public class SessionImpl
implements Session
{
private final long _sessionID;
private int _status;
private long _lastReadTime;
private long _lastWriteTime;
private long _timeOfLastStatusChange;
private DataChannel _dataChannel;
private Map _properties = new HashMap();
public SessionImpl( final long sessionID,
final long sessionAuth,
final int maxReadSize,
final int maxWriteSize )
{
_sessionID = sessionID;
setProperty( MessageUtils.AUTH_KEY, new Long( sessionAuth ) );
setStatus( STATUS_NOT_CONNECTED );
final ByteBuffer readBuffer =
ByteBuffer.allocateDirect( maxReadSize );
final ByteBuffer writeBuffer =
ByteBuffer.allocateDirect( maxWriteSize );
_dataChannel =
new DefaultDataChannel( this, readBuffer, writeBuffer );
}
public long getSessionID()
{
return _sessionID;
}
public int getStatus()
{
return _status;
}
public void setStatus( final int status )
{
_status = status;
_timeOfLastStatusChange = System.currentTimeMillis();
}
public long getTimeOfLastStatusChange()
{
return _timeOfLastStatusChange;
}
public void setProperty( final String key, final Object value )
{
_properties.put( key, value );
}
public Object getProperty( final String key )
{
return _properties.get( key );
}
public DataChannel getDataChannel()
{
return _dataChannel;
}
public long getLastReadTime()
{
return _lastReadTime;
}
public void setLastReadTime( final long lastReadTime )
{
_lastReadTime = lastReadTime;
}
public long getLastWriteTime()
{
return _lastWriteTime;
}
public void setLastWriteTime( final long lastWriteTime )
{
_lastWriteTime = lastWriteTime;
}
public void queueData()
{
throw new IllegalStateException( "Not implemented yet!" );
}
}
1.1 spice/sandbox/mesnet/src/java/org/realityforge/mesnet/TcpTransport.java
Index: TcpTransport.java
===================================================================
package org.realityforge.mesnet;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.channels.SelectableChannel;
import java.io.IOException;
/**
* An underlying transport layer that uses TCP/IP.
*
* @author <a href="mailto:peter at realityforge.org">Peter Donald</a>
* @version $Revision: 1.1 $ $Date: 2003/11/10 05:54:18 $
*/
public class TcpTransport
{
/**
* The key used to register channel in selector.
*/
private final SelectionKey _key;
/**
* The associated channel.
*/
private final SocketChannel _channel;
/**
* The associated session if any.
*/
private Session _session;
/**
* Create transport.
*
* @param key the key
*/
public TcpTransport( final SelectionKey key )
{
_key = key;
_channel = (SocketChannel)key.channel();
}
/**
* Return the SelectionKey.
*
* @return the SelectionKey.
*/
public SelectionKey getKey()
{
return _key;
}
/**
* Get underlying channel for transport.
*
* @return the transport
*/
public SocketChannel getChannel()
{
return _channel;
}
/**
* Get session associated with transport.
*
* @return the session
*/
public Session getSession()
{
return _session;
}
/**
* Associate a session with transport.
*
* @param session the session
*/
public void setSession( final Session session )
{
_session = session;
}
/**
* Close the channel and disconnect the key.
*/
public void close()
{
final SelectableChannel channel = _key.channel();
_key.attach( null );
_key.cancel();
try
{
channel.close();
}
catch( final IOException ioe )
{
//Ignore
}
}
}
-------------------------------------------------------
This SF.Net email sponsored by: ApacheCon 2003,
16-19 November in Las Vegas. Learn firsthand the latest
developments in Apache, PHP, Perl, XML, Java, MySQL,
WebDAV, and more! http://www.apachecon.com/