assorted-commits Mailing List for Assorted projects (Page 37)

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Revision: 1002
          http://assorted.svn.sourceforge.net/assorted/?rev=1002&view=rev
Author:   yangzhang
Date:     2008-10-07 22:35:53 +0000 (Tue, 07 Oct 2008)

Log Message:
-----------
fixed buggy writing; added unbounded writes; made the example into an echo server

Modified Paths:
--------------
    java-reactor/trunk/src/reactor/ReactorTest.java
    java-reactor/trunk/src/reactor/Session.java

Modified: java-reactor/trunk/src/reactor/ReactorTest.java
===================================================================

--- java-reactor/trunk/src/reactor/ReactorTest.java	2008-10-07 22:17:07 UTC (rev 1001)
+++ java-reactor/trunk/src/reactor/ReactorTest.java	2008-10-07 22:35:53 UTC (rev 1002)
@@ -129,8 +129,9 @@
 					// CharsetDecoder decoder = charset.newDecoder();
 					// CharBuffer charBuffer = decoder.decode(buf);
 					// System.out.println("read: " + charBuffer.toString());
-					System.out.println("read: '"
+					System.out.println("echoing: '"
 							+ Charset.forName("us-ascii").decode(buf) + "'");
+					session.write(buf);
 				} catch (Exception ex) {
 					throw new RuntimeException(ex);
 				}
@@ -146,12 +147,20 @@
 							public void handleConnect(Session session) {
 								System.out.println("connected");
 								try {
-									session.write(ByteBuffer.wrap("asdf"
+									session.write(ByteBuffer.wrap("hello"
 											.getBytes()));
 								} catch (Exception ex) {
 									throw new RuntimeException(ex);
 								}
 							}
+
+							@Override
+							public void handleRead(Session session,
+									InetSocketAddress src, ByteBuffer buf) {
+								System.out.println("got back: '"
+										+ Charset.forName("us-ascii").decode(
+												buf) + "'");
+							}
 						});
 			}
 		}, 1, TimeUnit.SECONDS);

Modified: java-reactor/trunk/src/reactor/Session.java
===================================================================
--- java-reactor/trunk/src/reactor/Session.java	2008-10-07 22:17:07 UTC (rev 1001)
+++ java-reactor/trunk/src/reactor/Session.java	2008-10-07 22:35:53 UTC (rev 1002)
@@ -11,6 +11,7 @@
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
+import java.nio.channels.WritableByteChannel;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -200,9 +201,10 @@
 			}
 		}
 	}
-	
+
 	/**
-	 * Handle the event where the socket has just established an outgoing connection.
+	 * Handle the event where the socket has just established an outgoing
+	 * connection.
 	 */
 	void handleConnect(SelectionKey key) throws IOException {
 		boolean result = ((SocketChannel) channel).finishConnect();
@@ -225,14 +227,9 @@
 		// Write until there's not more data ...
 		while (!pendingWrites.isEmpty()) {
 			ByteBuffer buf = (ByteBuffer) pendingWrites.get(0);
-			switch (type) {
-			case STREAM:
-				// XXX
-				break;
-			case DATAGRAM:
-				((DatagramChannel) channel).write(buf);
-				break;
-			}
+			int initPos = buf.position();
+			int bytes = ((WritableByteChannel) channel).write(buf);
+			assert buf.remaining() == buf.limit() - bytes && buf.position() == initPos + bytes;
 			if (buf.remaining() > 0) {
 				// ... or the socket's buffer fills up
 				break;
@@ -245,30 +242,32 @@
 			// interested
 			// in writing on this socket. Switch back to waiting
 			// for
-			// data.
+			// data (remove OP_WRITE).
 			key.interestOps(SelectionKey.OP_READ);
 		}
 	}
 
+	// TODO: use static (fixed) writeBuf (so that it can't change from under our
+	// feet)?
 	/**
 	 * Write data to this stream socket. This does not enqueue anything onto the
 	 * internal pendingWrites buffer if the socket's buffer is full, but instead
 	 * triggers an assertion failure.
 	 * 
-	 * @param writeBuf
+	 * @param buf
 	 *            The data to be written.
 	 */
-	public void write(ByteBuffer writeBuf) throws Exception {
-		switch (type) {
-		case STREAM:
-			int bytes = ((SocketChannel) channel).write(writeBuf);
-			// TODO: don't trigger an assertion failure!
-			assert bytes == writeBuf.limit();
-			break;
-		case DATAGRAM:
-			throw new Exception("cannot write() on a UDP session!");
-		default:
-			throw new AssertionError("unhandled session type");
+	public void write(ByteBuffer buf) throws Exception {
+		buf.rewind();
+		int initPos = buf.position();
+		int bytes = ((WritableByteChannel) channel).write(buf);
+		System.out.println("writing " + bytes);
+		assert buf.remaining() == buf.limit() - bytes && buf.position() == initPos + bytes;
+		if (buf.remaining() > 0) {
+			pendingWrites.add(buf);
+			// We're now interested in when the socket is ready for writing.
+			this.channel.keyFor(selector).interestOps(
+					SelectionKey.OP_READ | SelectionKey.OP_WRITE);
 		}
 	}
 


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.




2007	Jan	Feb	Mar	Apr	May	Jun	Jul	Aug	Sep	Oct	Nov (9)	Dec (12)
2008	Jan (86)	Feb (265)	Mar (96)	Apr (47)	May (136)	Jun (28)	Jul (57)	Aug (42)	Sep (20)	Oct (67)	Nov (37)	Dec (34)
2009	Jan (39)	Feb (85)	Mar (96)	Apr (24)	May (82)	Jun (13)	Jul (10)	Aug (8)	Sep (2)	Oct (20)	Nov (31)	Dec (17)
2010	Jan (16)	Feb (11)	Mar (17)	Apr (53)	May (31)	Jun (13)	Jul (3)	Aug (6)	Sep (11)	Oct (4)	Nov (17)	Dec (17)
2011	Jan (3)	Feb (19)	Mar (5)	Apr (17)	May (3)	Jun (4)	Jul (14)	Aug (3)	Sep (2)	Oct (1)	Nov (3)	Dec (2)
2012	Jan (3)	Feb (7)	Mar (1)	Apr	May (1)	Jun	Jul (4)	Aug (5)	Sep (2)	Oct (3)	Nov	Dec
2013	Jan	Feb	Mar (9)	Apr (5)	May	Jun (2)	Jul (1)	Aug (10)	Sep (1)	Oct (2)	Nov	Dec
2014	Jan (1)	Feb (3)	Mar (3)	Apr (1)	May (4)	Jun	Jul	Aug	Sep (2)	Oct	Nov	Dec
2015	Jan	Feb	Mar	Apr	May	Jun	Jul	Aug	Sep	Oct (1)	Nov	Dec
2016	Jan (1)	Feb	Mar (2)	Apr	May	Jun	Jul	Aug	Sep (1)	Oct	Nov	Dec
2017	Jan	Feb	Mar (1)	Apr	May (5)	Jun (1)	Jul	Aug	Sep	Oct	Nov	Dec (2)
2018	Jan	Feb	Mar	Apr	May	Jun (1)	Jul	Aug	Sep	Oct	Nov	Dec