Client.java: workaround for NPE + close underling connection. BasicInteropTest.java: added a connection close and a flush so that the message will definitely be received prior to the JVM exiting.

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

2 files changed, 29 insertions, 9 deletions

diff --git a/java/client/src/main/java/org/apache/qpidity/nclient/Client.java b/java/client/src/main/java/org/apache/qpidity/nclient/Client.java
index 14c6d0732a..42b482442a 100644
--- a/java/client/src/main/java/org/apache/qpidity/nclient/Client.java
+++ b/java/client/src/main/java/org/apache/qpidity/nclient/Client.java
@@ -51,11 +51,16 @@ public class Client implements org.apache.qpidity.nclient.Connection
 
             @Override public void connectionClose(Channel context, ConnectionClose connectionClose)
             {
-                _exceptionListner.onException(
-                        new QpidException("Server closed the connection: Reason " +
-                                           connectionClose.getReplyText(),
-                                           ErrorCode.get(connectionClose.getReplyCode()),
-                                           null));
+                // XXX: replaced reference to _exceptionListner with
+                // throw new RuntimeException because
+                // _exceptionListner may be null. In general this
+                // needs to be reworked because not every connection
+                // close is an exception!
+                throw new RuntimeException
+                    (new QpidException("Server closed the connection: Reason " +
+                                       connectionClose.getReplyText(),
+                                       ErrorCode.get(connectionClose.getReplyCode()),
+                                       null));
             }
         };
 
@@ -102,7 +107,7 @@ public class Client implements org.apache.qpidity.nclient.Connection
     {
         Channel ch = _conn.getChannel(0);
         ch.connectionClose(0, "client is closing", 0, 0);
-        //need to close the connection underneath as well
+        _conn.close();
     }
 
     public Session createSession(long expiryInSeconds)
diff --git a/java/client/src/main/java/org/apache/qpidity/nclient/interop/BasicInteropTest.java b/java/client/src/main/java/org/apache/qpidity/nclient/interop/BasicInteropTest.java
index a5bb12e116..b3d81cd2c1 100644
--- a/java/client/src/main/java/org/apache/qpidity/nclient/interop/BasicInteropTest.java
+++ b/java/client/src/main/java/org/apache/qpidity/nclient/interop/BasicInteropTest.java
@@ -25,6 +25,11 @@ public class BasicInteropTest implements ExceptionListener
         this.host = host;
     }
 
+    public void close() throws QpidException
+    {
+        conn.close();
+    }
+
     public void testCreateConnection(){
         System.out.println("------- Creating connection--------");
         conn = Client.createConnection();
@@ -46,7 +51,7 @@ public class BasicInteropTest implements ExceptionListener
 
     public void testExchange(){
         System.out.println("------- Creating an exchange --------");
-        session.exchangeDeclare("test", "amq.direct", "", null);
+        session.exchangeDeclare("test", "direct", "", null);
         session.sync();
         System.out.println("------- Exchange created --------");
     }
@@ -73,7 +78,8 @@ public class BasicInteropTest implements ExceptionListener
         System.out.println("------- Message sent --------");
     }
 
-    public void testSubscribe(){
+    public void testSubscribe()
+    {
         System.out.println("------- Sending a subscribe --------");
         session.messageSubscribe("testQueue", "myDest",
                                  Session.TRANSFER_CONFIRM_MODE_REQUIRED,
@@ -97,6 +103,13 @@ public class BasicInteropTest implements ExceptionListener
         session.messageFlowMode("myDest", Session.MESSAGE_FLOW_MODE_WINDOW);
         System.out.println("------- Setting Credit --------");
         session.messageFlow("myDest", Session.MESSAGE_FLOW_UNIT_MESSAGE, 1);
+        session.messageFlow("myDest", Session.MESSAGE_FLOW_UNIT_BYTE, -1);
+    }
+
+    public void testMessageFlush()
+    {
+        session.messageFlush("myDest");
+        session.sync();
     }
 
     public void onException(QpidException e)
@@ -107,7 +120,7 @@ public class BasicInteropTest implements ExceptionListener
         System.out.println("------- /Broker Notified an error --------");
     }
 
-    public static void main(String[] args)
+    public static void main(String[] args) throws QpidException
     {
         String host = "0.0.0.0";
         if (args.length>0)
@@ -122,5 +135,7 @@ public class BasicInteropTest implements ExceptionListener
         t.testQueue();
         t.testSubscribe();
         t.testSendMessage();
+        t.testMessageFlush();
+        t.close();
     }
 }