QPID-6350: [Java Common] Change IoSender to half close once writing is done.

Change IoSender half close once it knows it has finished writing data down the pipe.  This is done
to prevent the peer from seeing sporadic socket exception 'connection resets' if it happens that the other
side closes the socket first.  If half close is not supported (windows platform or SSLSocket), this change
has no effect.

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

4 files changed, 27 insertions, 6 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index 890aeda11b..e63949cc69 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -537,10 +537,7 @@ public class Connection extends ConnectionInvoker
         connectionLost.set(true);
         synchronized (lock)
         {
-            if(log.isDebugEnabled())
-            {
-                log.debug("exception: %s state : %s", e.getMessage(), state);
-            }
+            log.error(e, "exception: %s", e.getMessage());
 
             switch (state)
             {
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
index e06782c58a..25222e5285 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
@@ -26,12 +26,15 @@ import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.net.ssl.SSLSocket;
+
 import org.apache.qpid.thread.Threading;
 import org.apache.qpid.transport.Sender;
 import org.apache.qpid.transport.SenderClosedException;
 import org.apache.qpid.transport.SenderException;
 import org.apache.qpid.transport.TransportException;
 import org.apache.qpid.transport.util.Logger;
+import org.apache.qpid.util.SystemUtils;
 
 
 public final class IoSender implements Runnable, Sender<ByteBuffer>
@@ -58,6 +61,12 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
     private final Thread senderThread;
     private IoReceiver _receiver;
     private final String _remoteSocketAddress;
+    private static final boolean shutdownBroken;
+
+    static
+    {
+        shutdownBroken = SystemUtils.isWindows();
+    }
 
     private volatile Throwable exception = null;
 
@@ -314,6 +323,18 @@ public final class IoSender implements Runnable, Sender<ByteBuffer>
                 }
             }
         }
+
+        if (!shutdownBroken && !(socket instanceof SSLSocket))
+        {
+            try
+            {
+                socket.shutdownOutput();
+            }
+            catch (IOException e)
+            {
+                //pass
+            }
+        }
     }
 
     public void setIdleTimeout(int i)
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
index 1bbf166d82..8e1395aa83 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
@@ -187,7 +187,10 @@ public class SSLReceiver implements Receiver<ByteBuffer>
             }
             catch(SSLException e)
             {
-                log.error(e, "Error caught in SSLReceiver");
+                if (log.isDebugEnabled())
+                {
+                    log.debug(e, "Error caught in SSLReceiver");
+                }
                 _sslStatus.setSslErrorFlag();
                 synchronized(_sslStatus.getSslLock())
                 {
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
index 7c61136b42..53bd7e49b7 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
@@ -141,7 +141,7 @@ public class SSLSender implements Sender<ByteBuffer>
 
     public void send(ByteBuffer appData)
     {
-        if (closed.get())
+        if (closed.get() && !_sslStatus.getSslErrorFlag())
         {
             throw new SenderException("SSL Sender is closed");
         }