changed DtxSession to return Futures, moved MessageListener to util and added URL support to the client

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

10 files changed, 51 insertions, 40 deletions

diff --git a/java/client/src/main/java/org/apache/qpidity/client/Client.java b/java/client/src/main/java/org/apache/qpidity/client/Client.java
index 5af685071d..55ca885e43 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/Client.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/Client.java
@@ -1,11 +1,11 @@
 package org.apache.qpidity.client;
 
-import java.net.URL;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.apache.qpidity.BrokerDetails;
 import org.apache.qpidity.Channel;
 import org.apache.qpidity.Connection;
 import org.apache.qpidity.ConnectionClose;
@@ -14,9 +14,9 @@ import org.apache.qpidity.ErrorCode;
 import org.apache.qpidity.MinaHandler;
 import org.apache.qpidity.QpidException;
 import org.apache.qpidity.SessionDelegate;
-import org.apache.qpidity.url.QpidURL;
 import org.apache.qpidity.client.impl.ClientSession;
 import org.apache.qpidity.client.impl.ClientSessionDelegate;
+import org.apache.qpidity.url.QpidURL;
 
 
 public class Client implements org.apache.qpidity.client.Connection
@@ -86,7 +86,13 @@ public class Client implements org.apache.qpidity.client.Connection
      */
     public void connect(QpidURL url) throws QpidException
     {
-        throw new UnsupportedOperationException(); 
+        // temp impl to tests
+        BrokerDetails details = url.getAllBrokerDetails().get(0);
+        connect(details.getHost(),
+                details.getPort(),
+                details.getVirtualHost(),
+                details.getUserName(),
+                details.getPassword());
     }
     
     public void close() throws QpidException
diff --git a/java/client/src/main/java/org/apache/qpidity/client/DtxSession.java b/java/client/src/main/java/org/apache/qpidity/client/DtxSession.java
index 920e5cea80..21532f7d46 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/DtxSession.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/DtxSession.java
@@ -18,8 +18,14 @@
  */
 package org.apache.qpidity.client;
 
-import javax.transaction.xa.Xid;
-
+import org.apache.qpidity.DtxCoordinationCommitResult;
+import org.apache.qpidity.DtxCoordinationGetTimeoutResult;
+import org.apache.qpidity.DtxCoordinationPrepareResult;
+import org.apache.qpidity.DtxCoordinationRecoverResult;
+import org.apache.qpidity.DtxCoordinationRollbackResult;
+import org.apache.qpidity.DtxDemarcationEndResult;
+import org.apache.qpidity.DtxDemarcationStartResult;
+import org.apache.qpidity.Future;
 import org.apache.qpidity.Option;
 
 /**
@@ -40,7 +46,7 @@ public interface DtxSession extends Session
      * @param xid     Specifies the xid of the transaction branch to be started.
      * @param options Possible options are: {@link Option#JOIN} and {@link Option#RESUME}.
      */
-    public void dtxDemarcationStart(Xid xid, Option... options);
+    public Future<DtxDemarcationStartResult> dtxDemarcationStart(String xid, Option... options);
 
     /**
      * This method is called when the work done on behalf a transaction branch finishes or needs to
@@ -57,7 +63,7 @@ public interface DtxSession extends Session
      * @param xid     Specifies the xid of the transaction branch to be ended.
      * @param options Available options are: {@link Option#FAIL} and {@link Option#SUSPEND}.
      */
-    public void dtxDemarcationEnd(Xid xid, Option... options);
+    public Future<DtxDemarcationEndResult> dtxDemarcationEnd(String xid, Option... options);
 
     /**
      * Commit the work done on behalf a transaction branch. This method commits the work associated
@@ -70,14 +76,14 @@ public interface DtxSession extends Session
      * @param xid     Specifies the xid of the transaction branch to be committed.
      * @param options Available option is: {@link Option#ONE_PHASE}
      */
-    public void dtxCoordinationCommit(Xid xid, Option... options);
+    public Future<DtxCoordinationCommitResult> dtxCoordinationCommit(String xid, Option... options);
 
     /**
      * This method is called to forget about a heuristically completed transaction branch.
      *
      * @param xid Specifies the xid of the transaction branch to be forgotten.
      */
-    public void dtxCoordinationForget(Xid xid);
+    public void dtxCoordinationForget(String xid);
 
     /**
      * This method obtains the current transaction timeout value in seconds. If set-timeout was not
@@ -87,7 +93,7 @@ public interface DtxSession extends Session
      * @param xid Specifies the xid of the transaction branch for getting the timeout.
      * @return The current transaction timeout value in seconds.
      */
-    public long dtxCoordinationGetTimeout(Xid xid);
+    public Future<DtxCoordinationGetTimeoutResult> dtxCoordinationGetTimeout(String xid);
 
     /**
      * This method prepares for commitment any message produced or consumed on behalf of xid.
@@ -103,7 +109,7 @@ public interface DtxSession extends Session
      *         <p/>
      *         xa-rbtimeout: The work represented by this transaction branch took too long.
      */
-    public short dtxCoordinationPrepare(Xid xid);
+    public Future<DtxCoordinationPrepareResult> dtxCoordinationPrepare(String xid);
 
     /**
      * This method is called to obtain a list of transaction branches that are in a prepared or
@@ -111,7 +117,7 @@ public interface DtxSession extends Session
      *
      * @return a array of xids to be recovered.
      */
-    public Xid[] dtxCoordinationRecover();
+    public Future<DtxCoordinationRecoverResult> dtxCoordinationRecover();
 
     /**
      * This method rolls back the work associated with xid. Any produced messages are discarded and
@@ -119,7 +125,7 @@ public interface DtxSession extends Session
      *
      * @param xid Specifies the xid of the transaction branch that can be rolled back.
      */
-    public void dtxCoordinationRollback(Xid xid);
+    public Future<DtxCoordinationRollbackResult> dtxCoordinationRollback(String xid);
 
     /**
      * Sets the specified transaction branch timeout value in seconds.
@@ -127,5 +133,5 @@ public interface DtxSession extends Session
      * @param xid     Specifies the xid of the transaction branch for setting the timeout.
      * @param timeout The transaction timeout value in seconds.
      */
-    public void dtxCoordinationSetTimeout(Xid xid, long timeout);
+    public void dtxCoordinationSetTimeout(String xid, long timeout);
 }
diff --git a/java/client/src/main/java/org/apache/qpidity/client/impl/ClientSession.java b/java/client/src/main/java/org/apache/qpidity/client/impl/ClientSession.java
index 9247925073..6c2d4d477f 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/impl/ClientSession.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/impl/ClientSession.java
@@ -3,8 +3,6 @@ package org.apache.qpidity.client.impl;
 import java.io.EOFException;
 import java.io.IOException;
 import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
 import java.util.Map;
 
 import org.apache.qpidity.Option;
@@ -14,7 +12,6 @@ import org.apache.qpidity.RangeSet;
 import org.apache.qpidity.api.Message;
 import org.apache.qpidity.client.ExceptionListener;
 import org.apache.qpidity.client.MessagePartListener;
-import org.apache.qpidity.client.Session;
 
 /**
  * Implements a Qpid Sesion. 
diff --git a/java/client/src/main/java/org/apache/qpidity/client/impl/DemoClient.java b/java/client/src/main/java/org/apache/qpidity/client/impl/DemoClient.java
index e2962b4f22..1f8e9610d1 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/impl/DemoClient.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/impl/DemoClient.java
@@ -7,8 +7,8 @@ import org.apache.qpidity.api.Message;
 import org.apache.qpidity.client.Client;
 import org.apache.qpidity.client.Connection;
 import org.apache.qpidity.client.ExceptionListener;
-import org.apache.qpidity.client.MessageListener;
 import org.apache.qpidity.client.Session;
+import org.apache.qpidity.client.util.MessageListener;
 import org.apache.qpidity.client.util.MessagePartListenerAdapter;
 
 public class DemoClient
diff --git a/java/client/src/main/java/org/apache/qpidity/client/impl/LargeMsgDemoClient.java b/java/client/src/main/java/org/apache/qpidity/client/impl/LargeMsgDemoClient.java
index 38a7b36403..acf5c283b2 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/impl/LargeMsgDemoClient.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/impl/LargeMsgDemoClient.java
@@ -9,9 +9,9 @@ import org.apache.qpidity.api.Message;
 import org.apache.qpidity.client.Client;
 import org.apache.qpidity.client.Connection;
 import org.apache.qpidity.client.ExceptionListener;
-import org.apache.qpidity.client.MessageListener;
 import org.apache.qpidity.client.Session;
 import org.apache.qpidity.client.util.FileMessage;
+import org.apache.qpidity.client.util.MessageListener;
 import org.apache.qpidity.client.util.MessagePartListenerAdapter;
 
 public class LargeMsgDemoClient
diff --git a/java/client/src/main/java/org/apache/qpidity/client/MessageListener.java b/java/client/src/main/java/org/apache/qpidity/client/util/MessageListener.java
index eab9d70f00..bf157c91d7 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/MessageListener.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/util/MessageListener.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.qpidity.client;
+package org.apache.qpidity.client.util;
 
 import org.apache.qpidity.api.Message;
 
diff --git a/java/client/src/main/java/org/apache/qpidity/client/util/MessagePartListenerAdapter.java b/java/client/src/main/java/org/apache/qpidity/client/util/MessagePartListenerAdapter.java
index c4b8ae3f8b..e4f19ea6c3 100644
--- a/java/client/src/main/java/org/apache/qpidity/client/util/MessagePartListenerAdapter.java
+++ b/java/client/src/main/java/org/apache/qpidity/client/util/MessagePartListenerAdapter.java
@@ -6,7 +6,6 @@ import java.nio.ByteBuffer;
 import org.apache.qpidity.DeliveryProperties;
 import org.apache.qpidity.MessageProperties;
 import org.apache.qpidity.Struct;
-import org.apache.qpidity.client.MessageListener;
 import org.apache.qpidity.client.MessagePartListener;
 
 /** 
diff --git a/java/client/src/main/java/org/apache/qpidity/jms/QpidBrowserListener.java b/java/client/src/main/java/org/apache/qpidity/jms/QpidBrowserListener.java
index 05b97a1154..ecff09a9aa 100644
--- a/java/client/src/main/java/org/apache/qpidity/jms/QpidBrowserListener.java
+++ b/java/client/src/main/java/org/apache/qpidity/jms/QpidBrowserListener.java
@@ -20,7 +20,7 @@ package org.apache.qpidity.jms;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.qpidity.api.Message;
-import org.apache.qpidity.client.MessageListener;
+import org.apache.qpidity.client.util.MessageListener;
 
 /**
  * This listener idspatches messaes to its browser.
diff --git a/java/client/src/main/java/org/apache/qpidity/jms/QpidMessageListener.java b/java/client/src/main/java/org/apache/qpidity/jms/QpidMessageListener.java
index 082d38ce21..ada5d048e1 100644
--- a/java/client/src/main/java/org/apache/qpidity/jms/QpidMessageListener.java
+++ b/java/client/src/main/java/org/apache/qpidity/jms/QpidMessageListener.java
@@ -20,7 +20,7 @@ package org.apache.qpidity.jms;
 import org.apache.qpidity.jms.message.QpidMessage;
 import org.apache.qpidity.jms.message.MessageFactory;
 import org.apache.qpidity.api.Message;
-import org.apache.qpidity.client.MessageListener;
+import org.apache.qpidity.client.util.MessageListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
diff --git a/java/client/src/main/java/org/apache/qpidity/jms/XAResourceImpl.java b/java/client/src/main/java/org/apache/qpidity/jms/XAResourceImpl.java
index efcb4430c8..fc298b01cb 100644
--- a/java/client/src/main/java/org/apache/qpidity/jms/XAResourceImpl.java
+++ b/java/client/src/main/java/org/apache/qpidity/jms/XAResourceImpl.java
@@ -17,14 +17,13 @@
  */
 package org.apache.qpidity.jms;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.qpidity.Option;
-import org.apache.qpidity.QpidException;
-
+import javax.transaction.xa.XAException;
 import javax.transaction.xa.XAResource;
 import javax.transaction.xa.Xid;
-import javax.transaction.xa.XAException;
+
+import org.apache.qpidity.Option;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This is an implementation of javax.jms.XAResource.
@@ -72,7 +71,7 @@ public class XAResourceImpl implements XAResource
         {
             _logger.debug("commit ", xid);
         }
-        _xaSession.getQpidSession().dtxCoordinationCommit(xid, b ? Option.ONE_PHASE : Option.NO_OPTION);
+        _xaSession.getQpidSession().dtxCoordinationCommit(new String(xid.getGlobalTransactionId()), b ? Option.ONE_PHASE : Option.NO_OPTION);
     }
 
     /**
@@ -99,14 +98,14 @@ public class XAResourceImpl implements XAResource
         }
         xid = null;
         _xaSession.getQpidSession()
-                .dtxDemarcationEnd(xid, flag == XAResource.TMFAIL ? Option.FAIL : Option.NO_OPTION,
+                .dtxDemarcationEnd(new String(xid.getGlobalTransactionId()), flag == XAResource.TMFAIL ? Option.FAIL : Option.NO_OPTION,
                                    flag == XAResource.TMSUSPEND ? Option.SUSPEND : Option.NO_OPTION);
     }
 
     /**
      * Tells the resource manager to forget about a heuristically completed transaction branch.
      *
-     * @param xid A global transaction identifier
+     * @param new String(xid.getGlobalTransactionId() A global transaction identifier
      * @throws XAException An error has occurred. Possible exception values are XAER_RMERR, XAER_RMFAIL,
      *                     XAER_NOTA, XAER_INVAL, or XAER_PROTO.
      */
@@ -116,7 +115,7 @@ public class XAResourceImpl implements XAResource
         {
             _logger.debug("forget ", xid);
         }
-        _xaSession.getQpidSession().dtxCoordinationForget(xid);
+        _xaSession.getQpidSession().dtxCoordinationForget(new String(xid.getGlobalTransactionId()));
     }
 
     /**
@@ -133,7 +132,8 @@ public class XAResourceImpl implements XAResource
         int result = 0;
         if (_xid != null)
         {
-            result = (int) _xaSession.getQpidSession().dtxCoordinationGetTimeout(_xid);
+            result = 0; 
+            _xaSession.getQpidSession().dtxCoordinationGetTimeout(new String(_xid.getGlobalTransactionId()));
         }
         return result;
     }
@@ -169,8 +169,9 @@ public class XAResourceImpl implements XAResource
             _logger.debug("prepare ", xid);
         }
         int result;
-        result = _xaSession.getQpidSession()
-        .dtxCoordinationPrepare(xid);
+        result = 0;
+        _xaSession.getQpidSession()
+        .dtxCoordinationPrepare(new String(xid.getGlobalTransactionId()));
         
         if (result == XAException.XA_RDONLY)
         {
@@ -198,8 +199,10 @@ public class XAResourceImpl implements XAResource
     public Xid[] recover(int flag) throws XAException
     {
 //      the flag is ignored 
-        return _xaSession.getQpidSession()
+        
+        _xaSession.getQpidSession()
                 .dtxCoordinationRecover();
+        return null;
     }
 
     /**
@@ -212,7 +215,7 @@ public class XAResourceImpl implements XAResource
     {
 //      the flag is ignored
         _xaSession.getQpidSession()
-                .dtxCoordinationRollback(xid);
+                .dtxCoordinationRollback(new String(xid.getGlobalTransactionId()));
     }
 
     /**
@@ -231,7 +234,7 @@ public class XAResourceImpl implements XAResource
         if (_xid != null)
         {
             _xaSession.getQpidSession()
-            .dtxCoordinationSetTimeout(_xid, timeout);
+            .dtxCoordinationSetTimeout(new String(_xid.getGlobalTransactionId()), timeout);
             result = true;
         }
         return result;
@@ -259,7 +262,7 @@ public class XAResourceImpl implements XAResource
         }
         _xid = xid;
         _xaSession.getQpidSession()
-        .dtxDemarcationStart(xid, flag == XAResource.TMJOIN ? Option.JOIN : Option.NO_OPTION,
+        .dtxDemarcationStart(new String(xid.getGlobalTransactionId()), flag == XAResource.TMJOIN ? Option.JOIN : Option.NO_OPTION,
                              flag == XAResource.TMRESUME ? Option.RESUME : Option.NO_OPTION);
     }
 }