This is related to QPID-1102.

I have fixed the error handling and revised the while loop in IoSender based on the comments received during the code review


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@676938 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 27 insertions, 28 deletions

diff --git a/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoHandler.java b/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoHandler.java
index 568526d9db..82fd03857a 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoHandler.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoHandler.java
@@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.qpidity.transport.Connection;
 import org.apache.qpidity.transport.ConnectionDelegate;
 import org.apache.qpidity.transport.Receiver;
+import org.apache.qpidity.transport.TransportException;
 import org.apache.qpidity.transport.network.Assembler;
 import org.apache.qpidity.transport.network.Disassembler;
 import org.apache.qpidity.transport.network.InputHandler;
@@ -94,19 +95,14 @@ public class IoHandler implements Runnable
             {
                 _socket.connect(new InetSocketAddress(address, port));
             }
-            while (!_socket.isConnected())
-            {
-
-            }
-
         }
         catch (SocketException e)
         {
-            throw new RuntimeException("Error connecting to broker",e);
+            throw new TransportException("Error connecting to broker",e);
         }
         catch (IOException e)
         {
-            throw new RuntimeException("Error connecting to broker",e);
+            throw new TransportException("Error connecting to broker",e);
         }
 
         IoSender sender = new IoSender(_socket);
@@ -133,28 +129,21 @@ public class IoHandler implements Runnable
         {
             InputStream in = _socket.getInputStream();
             int read = 0;
-            while(_socket.isConnected())
+            while(read != -1)
             {
-                try
-                {
-                    read = in.read(_readBuf);
-                    if (read > 0)
-                    {
-                        ByteBuffer b = ByteBuffer.allocate(read);
-                        b.put(_readBuf,0,read);
-                        b.flip();
-                        _receiver.received(b);
-                    }
-                }
-                catch(Exception e)
+                read = in.read(_readBuf);
+                if (read > 0)
                 {
-                    throw new RuntimeException("Error reading from socket input stream",e);
+                    ByteBuffer b = ByteBuffer.allocate(read);
+                    b.put(_readBuf,0,read);
+                    b.flip();
+                    _receiver.received(b);
                 }
             }
         }
         catch (IOException e)
         {
-            throw new RuntimeException("Error getting input stream from the socket",e);
+            _receiver.exception(new Exception("Error getting input stream from the socket",e));
         }
         finally
         {
diff --git a/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoSender.java b/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoSender.java
index d50442be5c..511d5d5b9e 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoSender.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/network/io/IoSender.java
@@ -23,6 +23,7 @@ import java.io.OutputStream;
 import java.net.Socket;
 
 import org.apache.qpidity.transport.Sender;
+import org.apache.qpidity.transport.TransportException;
 
 public class IoSender implements Sender<java.nio.ByteBuffer>
 {
@@ -39,7 +40,7 @@ public class IoSender implements Sender<java.nio.ByteBuffer>
         }
         catch(IOException e)
         {
-            throw new RuntimeException("Error getting output stream for socket",e);
+            throw new TransportException("Error getting output stream for socket",e);
         }
     }
 
@@ -66,8 +67,18 @@ public class IoSender implements Sender<java.nio.ByteBuffer>
      */
     private void write(java.nio.ByteBuffer buf)
     {
-        byte[] array = new byte[buf.remaining()];
-        buf.get(array);
+        byte[] array = null;
+
+        if (buf.hasArray())
+        {
+            array = buf.array();
+        }
+        else
+        {
+            array = new byte[buf.remaining()];
+            buf.get(array);
+        }
+
         if( _socket.isConnected())
         {
             synchronized (lock)
@@ -78,14 +89,13 @@ public class IoSender implements Sender<java.nio.ByteBuffer>
                 }
                 catch(Exception e)
                 {
-                    e.fillInStackTrace();
-                    throw new RuntimeException("Error trying to write to the socket",e);
+                    throw new TransportException("Error trying to write to the socket",e);
                 }
             }
         }
         else
         {
-            throw new RuntimeException("Trying to write on a closed socket");
+            throw new TransportException("Trying to write on a closed socket");
         }
     }