1 files changed, 135 insertions, 0 deletions
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/nio/NioHandler.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/nio/NioHandler.java
new file mode 100644
index 0000000000..84e66c25bd
--- /dev/null
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/nio/NioHandler.java
@@ -0,0 +1,135 @@
+package org.apache.qpid.transport.network.nio;
+/*
+ * 
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * 
+ */
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.qpid.transport.Connection;
+import org.apache.qpid.transport.ConnectionDelegate;
+import org.apache.qpid.transport.Receiver;
+import org.apache.qpid.transport.network.Assembler;
+import org.apache.qpid.transport.network.Disassembler;
+import org.apache.qpid.transport.network.InputHandler;
+
+public class NioHandler implements Runnable
+{
+    private Receiver<ByteBuffer> _receiver;
+    private SocketChannel _ch;
+    private ByteBuffer _readBuf;
+    private static Map<Long,NioSender> _handlers = new ConcurrentHashMap<Long,NioSender>();
+
+    private NioHandler(){}
+
+    public static final Connection connect(String host, int port,
+            ConnectionDelegate delegate)
+    {
+        NioHandler handler = new NioHandler();
+        return handler.connectInternal(host,port,delegate);
+    }
+
+    private Connection connectInternal(String host, int port,
+            ConnectionDelegate delegate)
+    {
+        try
+        {
+            SocketAddress address = new InetSocketAddress(host,port);
+            _ch = SocketChannel.open();
+            _ch.socket().setReuseAddress(true);
+            _ch.configureBlocking(true);
+            _ch.socket().setTcpNoDelay(true);
+            if (address != null)
+            {
+                _ch.socket().connect(address);
+            }
+            while (_ch.isConnectionPending())
+            {
+
+            }
+
+        }
+        catch (SocketException e)
+        {
+
+            e.printStackTrace();
+        }
+        catch (IOException e)
+        {
+            e.printStackTrace();
+        }
+
+        NioSender sender = new NioSender(_ch);
+        Connection con = new Connection();
+        con.setSender(new Disassembler(sender, 64*1024 - 1));
+        con.setConnectionDelegate(delegate);
+
+        _handlers.put(con.getConnectionId(),sender);
+
+        _receiver = new InputHandler(new Assembler(con), InputHandler.State.FRAME_HDR);
+
+        Thread t = new Thread(this);
+        t.start();
+
+        return con;
+    }
+
+    public void run()
+    {
+        _readBuf = ByteBuffer.allocate(512);
+        long read = 0;
+        while(_ch.isConnected() && _ch.isOpen())
+        {
+            try
+            {
+                read = _ch.read(_readBuf);
+                if (read > 0)
+                {
+                    _readBuf.flip();
+                    ByteBuffer b = ByteBuffer.allocate(_readBuf.remaining());
+                    b.put(_readBuf);
+                    b.flip();
+                    _readBuf.clear();
+                    _receiver.received(b);
+                }
+            }
+            catch(Exception e)
+            {
+                e.printStackTrace();
+            }
+        }
+
+        //throw new EOFException("The underlying socket/channel has closed");
+    }
+
+    public static void startBatchingFrames(int connectionId)
+    {
+        NioSender sender = _handlers.get(connectionId);
+        sender.setStartBatching();
+    }
+
+
+}