Changeset 460

Timestamp:

07/21/07 09:36:27 (1 year ago)

Author:

blackhedd

Message:

dev, datagram support

Files:

• version_0/java/src/com/rubyeventmachine/EmReactor.java (modified) (10 diffs)

version_0/java/src/com/rubyeventmachine/EmReactor.java

@@ -123 +123 @@
-                                                        sn.configureBlocking(false);
-                                                        String b = createBinding();
-Channel ec = new Eventable
+SocketChannel ec = new EventableSocket
-                                                        Connections.put(b, ec);
-                                                        eventCallback ((String)k.attachment(), EM_CONNECTION_ACCEPTED, ByteBuffer.wrap(b.getBytes()));
@@ -130 +130 @@
-
-                                        if (k.isReadable()) {
+                                                EventableChannel ec = (EventableChannel)k.attachment();
+                                                myReadBuffer.clear();
+                                                ec.readInboundData (myReadBuffer);
+                                                myReadBuffer.flip();
+                                                String b = ec.getBinding();
+                                                if (myReadBuffer.limit() > 0) {
+                                                        eventCallback (b, EM_CONNECTION_READ, myReadBuffer);
+                                                }
+                                                else {
+                                                        eventCallback (b, EM_CONNECTION_UNBOUND, EmptyByteBuffer);
+                                                        Connections.remove(b);
+                                                        k.channel().close();                                                    
+                                                }
+                                                /*
+                                                System.out.println ("READABLE");
-                                                SocketChannel sn = (SocketChannel) k.channel();
-                                                //ByteBuffer bb = ByteBuffer.allocate(16 * 1024);
@@ -152 +167 @@
-                                                        sn.close();
-                                                }
+                                                */
-                                        }
-                                
@@ -165 +181 @@
-                                        
-                                        if (k.isConnectable()) {
-Channel ec = (Eventable
+SocketChannel ec = (EventableSocket
-                                                if (ec.finishConnecting()) {
-                                                        eventCallback (ec.getBinding(), EM_CONNECTION_COMPLETED, EmptyByteBuffer);
@@ -190 +206 @@
-                mySelector.close();
-                mySelector = null;
+
+                // run down open connections and sockets.
+                Iterator<ServerSocketChannel> i = Acceptors.values().iterator();
+                while (i.hasNext()) {
+                        i.next().close();
+                }
+                
+                Iterator<EventableChannel> i2 = Connections.values().iterator();
+                while (i2.hasNext())
+                        i2.next().close();
-        }
-        
@@ -218 +244 @@
-                BindingIndex++;
-                String s = createBinding();
-                //System.out.println (new Date().getTime()+ milliseconds);
-                Timers.put(new Date().getTime() + milliseconds, s);
-                return s;
@@ -241 +266 @@
-        }
-        
+
+        /**
+         * 
+         * @param address
+         * @param port
+         * @return
+         * @throws IOException
+         */
+        public String openUdpSocket (String address, int port) throws IOException {
+                DatagramChannel dg = DatagramChannel.open();
+                dg.configureBlocking(false);
+                dg.socket().bind( new InetSocketAddress(address,port));
+                String b = createBinding();
+                EventableChannel ec = new EventableDatagramChannel (dg, b, mySelector);
+                dg.register(mySelector, SelectionKey.OP_READ, ec);
+                Connections.put(b, ec);
+                return b;
+        }
+        
-        public void sendData (String sig, ByteBuffer bb) throws IOException {
-                (Connections.get(sig)).scheduleOutboundData( bb );
@@ -249 +293 @@
-        }
-        
+        /**
+         * 
+         * @param sig
+         * @param data
+         * @param length
+         * @param recipAddress
+         * @param recipPort
+         */
+        public void sendDatagram (String sig, String data, int length, String recipAddress, int recipPort) {
+                sendDatagram (sig, ByteBuffer.wrap(data.getBytes()), recipAddress, recipPort);
+        }
+        
+        /**
+         * 
+         * @param sig
+         * @param bb
+         * @param recipAddress
+         * @param recipPort
+         */
+        public void sendDatagram (String sig, ByteBuffer bb, String recipAddress, int recipPort) {
+                (Connections.get(sig)).scheduleOutboundDatagram( bb, recipAddress, recipPort);
+        }
+
+        
+        /**
+         * 
+         * @param address
+         * @param port
+         * @return
+         * @throws ClosedChannelException
+         */
-        public String connectTcpServer (String address, int port) throws ClosedChannelException {
-                String b = createBinding();
@@ -255 +330 @@
-                        SocketChannel sc = SocketChannel.open();
-                        sc.configureBlocking(false);
-Channel ec = new Eventable
+SocketChannel ec = new EventableSocket
-
-                        if (sc.connect (new InetSocketAddress (address, port))) {
@@ -285 +360 @@
-
-        public void closeConnection (String sig, boolean afterWriting) throws ClosedChannelException {
-                //System.out.println ("???"+Connections.get(sig));
-                Connections.get(sig).scheduleClose (afterWriting);
-        }