ICE-5661 some changes to class structure for OutgoingAsync

11 files changed, 118 insertions, 48 deletions

diff --git a/java/src/Ice/CommunicatorI.java b/java/src/Ice/CommunicatorI.java
index 121d82d7f2b..ed517e4ae9f 100644
--- a/java/src/Ice/CommunicatorI.java
+++ b/java/src/Ice/CommunicatorI.java
@@ -283,8 +283,8 @@ public final class CommunicatorI implements Communicator
     public void
     end_flushBatchRequests(AsyncResult r)
     {
-        IceInternal.AsyncResultI ri = (IceInternal.AsyncResultI)r;
-        IceInternal.AsyncResultI.check(ri, this, __flushBatchRequests_name);
+        IceInternal.OutgoingAsyncBase ri = (IceInternal.OutgoingAsyncBase)r;
+        IceInternal.OutgoingAsyncBase.check(ri, this, __flushBatchRequests_name);
         ri.__wait();
     }
 
diff --git a/java/src/Ice/ConnectionI.java b/java/src/Ice/ConnectionI.java
index cb3383bfb56..e3c555f6174 100644
--- a/java/src/Ice/ConnectionI.java
+++ b/java/src/Ice/ConnectionI.java
@@ -651,8 +651,8 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
     @Override
     public void end_flushBatchRequests(AsyncResult ir)
     {
-        IceInternal.AsyncResultI r = (IceInternal.AsyncResultI) ir;
-        IceInternal.AsyncResultI.check(r, this, __flushBatchRequests_name);
+        IceInternal.OutgoingAsyncBase r = (IceInternal.OutgoingAsyncBase) ir;
+        IceInternal.OutgoingAsyncBase.check(r, this, __flushBatchRequests_name);
         r.__wait();
     }
 
diff --git a/java/src/Ice/ObjectPrxHelperBase.java b/java/src/Ice/ObjectPrxHelperBase.java
index 72d3d99f150..9387bb4a6ba 100644
--- a/java/src/Ice/ObjectPrxHelperBase.java
+++ b/java/src/Ice/ObjectPrxHelperBase.java
@@ -305,8 +305,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public final boolean
     end_ice_isA(AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI)__iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_isA_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase)__iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_isA_name);
         try
         {
             if(!__result.__wait())
@@ -797,8 +797,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public final String[]
     end_ice_ids(AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI) __iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_ids_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase) __iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_ids_name);
         try
         {
             if(!__result.__wait())
@@ -1073,8 +1073,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public final String
     end_ice_id(AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI) __iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_id_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase) __iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_id_name);
         try
         {
             if(!__result.__wait())
@@ -1468,8 +1468,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public final boolean
     end_ice_invoke(ByteSeqHolder outParams, AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI) __iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_invoke_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase) __iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_invoke_name);
         try
         {
             boolean ok = __result.__wait();
@@ -2488,8 +2488,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public Ice.Connection
     end_ice_getConnection(AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI)__iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_getConnection_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase)__iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_getConnection_name);
         __result.__wait();
         return ice_getCachedConnection();
     }
@@ -2654,8 +2654,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public void
     end_ice_flushBatchRequests(AsyncResult __iresult)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI)__iresult;
-        IceInternal.AsyncResultI.check(__result, this, __ice_flushBatchRequests_name);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase)__iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, __ice_flushBatchRequests_name);
         __result.__wait();
     }
 
@@ -2780,8 +2780,8 @@ public class ObjectPrxHelperBase implements ObjectPrx, java.io.Serializable
     public final void
     __end(AsyncResult __iresult, String operation)
     {
-        IceInternal.AsyncResultI __result = (IceInternal.AsyncResultI)__iresult;
-        IceInternal.AsyncResultI.check(__result, this, operation);
+        IceInternal.OutgoingAsyncBase __result = (IceInternal.OutgoingAsyncBase)__iresult;
+        IceInternal.OutgoingAsyncBase.check(__result, this, operation);
         try
         {
             boolean ok = __result.__wait();
diff --git a/java/src/IceInternal/BatchOutgoingAsync.java b/java/src/IceInternal/BatchOutgoingAsync.java
index d5953310639..e78879afa6a 100644
--- a/java/src/IceInternal/BatchOutgoingAsync.java
+++ b/java/src/IceInternal/BatchOutgoingAsync.java
@@ -9,7 +9,7 @@
 
 package IceInternal;
 
-public class BatchOutgoingAsync extends AsyncResultI implements OutgoingAsyncMessageCallback
+public class BatchOutgoingAsync extends OutgoingAsyncBase implements OutgoingAsyncMessageCallback
 {
     BatchOutgoingAsync(Ice.Communicator communicator, Instance instance, String operation, CallbackBase callback)
     {
@@ -96,6 +96,13 @@ public class BatchOutgoingAsync extends AsyncResultI implements OutgoingAsyncMes
 
     @Override
     public void
+    processRetry(boolean destroyed)
+    {
+        // Does not implement retry
+    }
+
+    @Override
+    public void
     dispatchInvocationCancel(final Ice.LocalException ex, ThreadPool threadPool, Ice.Connection connection)
     {
         threadPool.dispatch(new DispatchWorkItem(connection)
diff --git a/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java b/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
index 6ebd6fb3a3e..d5c6189064a 100644
--- a/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
+++ b/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
@@ -9,7 +9,7 @@
 
 package IceInternal;
 
-public class CommunicatorBatchOutgoingAsync extends IceInternal.AsyncResultI
+public class CommunicatorBatchOutgoingAsync extends OutgoingAsyncBase
 {
     public CommunicatorBatchOutgoingAsync(Ice.Communicator communicator, Instance instance, String operation,
                                           CallbackBase callback)
diff --git a/java/src/IceInternal/GetConnectionOutgoingAsync.java b/java/src/IceInternal/GetConnectionOutgoingAsync.java
index 95878b90709..75ef2d9b60f 100644
--- a/java/src/IceInternal/GetConnectionOutgoingAsync.java
+++ b/java/src/IceInternal/GetConnectionOutgoingAsync.java
@@ -9,11 +9,13 @@
 
 package IceInternal;
 
-public class GetConnectionOutgoingAsync extends IceInternal.OutgoingAsync
+public class GetConnectionOutgoingAsync extends OutgoingAsyncBase implements OutgoingAsyncMessageCallback
 {
-    public GetConnectionOutgoingAsync(Ice.ObjectPrxHelperBase prx, String operation, CallbackBase callback)
+    public GetConnectionOutgoingAsync(Ice.ObjectPrxHelperBase prx, String operation, CallbackBase cb)
     {
-        super(prx, operation, callback);
+        super(prx.ice_getCommunicator(), ((Ice.ObjectPrxHelperBase) prx).__reference().getInstance(), operation, cb);
+        _proxy = (Ice.ObjectPrxHelperBase) prx;
+        _cnt = 0;
         _observer = ObserverHelper.get(prx, operation);
     }
 
@@ -39,6 +41,12 @@ public class GetConnectionOutgoingAsync extends IceInternal.OutgoingAsync
     }
 
     @Override
+    public Ice.ObjectPrx getProxy()
+    {
+        return _proxy;
+    }
+
+    @Override
     public int send(Ice.ConnectionI conection, boolean compress, boolean response)
         throws RetryException
     {
@@ -84,6 +92,39 @@ public class GetConnectionOutgoingAsync extends IceInternal.OutgoingAsync
         }
     }
 
+    @Override
+    public void processRetry(boolean destroyed)
+    {
+        if(destroyed)
+        {
+            invokeExceptionAsync(new Ice.CommunicatorDestroyedException());
+        }
+        else
+        {
+            try
+            {
+                __invoke();
+            }
+            catch(Ice.LocalException ex)
+            {
+                invokeExceptionAsync(ex);
+            }
+        }
+    }
+
+    @Override
+    public void dispatchInvocationCancel(final Ice.LocalException ex, ThreadPool threadPool, Ice.Connection connection)
+    {
+        threadPool.dispatch(new DispatchWorkItem(connection)
+        {
+            @Override
+            public void run()
+            {
+                GetConnectionOutgoingAsync.this.finished(ex);
+            }
+        });
+    }
+
     private void handleException(Ice.Exception exc)
     {
        try
@@ -105,4 +146,8 @@ public class GetConnectionOutgoingAsync extends IceInternal.OutgoingAsync
             throw ex;
         }
     }
+
+    private Ice.ObjectPrxHelperBase _proxy;
+    private RequestHandler _handler = null;
+    private int _cnt;
 }
diff --git a/java/src/IceInternal/OutgoingAsync.java b/java/src/IceInternal/OutgoingAsync.java
index 992b9ca6265..94e3a8ae2b4 100644
--- a/java/src/IceInternal/OutgoingAsync.java
+++ b/java/src/IceInternal/OutgoingAsync.java
@@ -9,7 +9,7 @@
 
 package IceInternal;
 
-public class OutgoingAsync extends AsyncResultI implements OutgoingAsyncMessageCallback
+public class OutgoingAsync extends OutgoingAsyncBase implements OutgoingAsyncMessageCallback
 {
     public OutgoingAsync(Ice.ObjectPrx prx, String operation, CallbackBase cb)
     {
@@ -245,6 +245,26 @@ public class OutgoingAsync extends AsyncResultI implements OutgoingAsyncMessageC
     }
 
     @Override
+    public void processRetry(boolean destroyed)
+    {
+        if(destroyed)
+        {
+            invokeExceptionAsync(new Ice.CommunicatorDestroyedException());
+        }
+        else
+        {
+            try
+            {
+                invoke(false);
+            }
+            catch(Ice.LocalException ex)
+            {
+                invokeExceptionAsync(ex);
+            }
+        }
+    }
+
+    @Override
     public void dispatchInvocationCancel(final Ice.LocalException ex, ThreadPool threadPool, Ice.Connection connection)
     {
         threadPool.dispatch(new DispatchWorkItem(connection)
@@ -654,9 +674,9 @@ public class OutgoingAsync extends AsyncResultI implements OutgoingAsyncMessageC
     }
 
     protected Ice.ObjectPrxHelperBase _proxy;
-    protected RequestHandler _handler;
-    protected int _cnt;
 
+    private RequestHandler _handler;
+    private int _cnt;
     private Ice.EncodingVersion _encoding;
     private Ice.OperationMode _mode;
     private boolean _sent;
diff --git a/java/src/IceInternal/AsyncResultI.java b/java/src/IceInternal/OutgoingAsyncBase.java
index e2ca0b7c466..8be0b3072bd 100644
--- a/java/src/IceInternal/AsyncResultI.java
+++ b/java/src/IceInternal/OutgoingAsyncBase.java
@@ -21,10 +21,10 @@ import Ice.UserException;
  * With this object, an application can obtain several attributes of the
  * invocation and discover its outcome.
  **/
-public class AsyncResultI implements Ice.AsyncResult
+public class OutgoingAsyncBase implements Ice.AsyncResult
 {
-    protected AsyncResultI(Communicator communicator, IceInternal.Instance instance, String op,
-                          IceInternal.CallbackBase del)
+    protected OutgoingAsyncBase(Communicator communicator, IceInternal.Instance instance, String op,
+                                IceInternal.CallbackBase del)
     {
         _communicator = communicator;
         _instance = instance;
@@ -35,9 +35,9 @@ public class AsyncResultI implements Ice.AsyncResult
         _exception = null;
         _callback = del;
     }
-    
-    protected AsyncResultI(Communicator communicator, Instance instance, String op, CallbackBase del, BasicStream is,
-            BasicStream os)
+
+    protected OutgoingAsyncBase(Communicator communicator, Instance instance, String op, CallbackBase del,
+                                BasicStream is, BasicStream os)
     {
         _communicator = communicator;
         _instance = instance;
@@ -455,7 +455,7 @@ public class AsyncResultI implements Ice.AsyncResult
     public void cacheMessageBuffers()
     {
     }
-    
+
     public final void invokeCompleted()
     {
         //
@@ -493,7 +493,7 @@ public class AsyncResultI implements Ice.AsyncResult
                     Thread.currentThread().setContextClassLoader(null);
                 }
             }
-            
+
             cacheMessageBuffers();
         }
 
@@ -520,7 +520,7 @@ public class AsyncResultI implements Ice.AsyncResult
                             new Ice.InvocationTimeoutException());
         }
     }
-    
+
     private static void check(AsyncResult r, String operation)
     {
         if(r == null)
@@ -548,7 +548,7 @@ public class AsyncResultI implements Ice.AsyncResult
         String s = "error raised by AMI callback:\n" + IceInternal.Ex.toString(error);
         _instance.initializationData().logger.error(s);
     }
-    
+
     protected Communicator _communicator;
     protected IceInternal.Instance _instance;
     protected String _operation;
diff --git a/java/src/IceInternal/OutgoingAsyncMessageCallback.java b/java/src/IceInternal/OutgoingAsyncMessageCallback.java
index 95c5fd2bb45..9c069918933 100644
--- a/java/src/IceInternal/OutgoingAsyncMessageCallback.java
+++ b/java/src/IceInternal/OutgoingAsyncMessageCallback.java
@@ -44,6 +44,11 @@ public interface OutgoingAsyncMessageCallback
     void finished(Ice.Exception ex);
 
     //
+    // Called by the retry queue to process retry.
+    //
+    void processRetry(boolean destroyed);
+
+    //
     // Helper to dispatch the cancellation exception.
     //
     void dispatchInvocationCancel(Ice.LocalException ex, ThreadPool threadPool, Ice.Connection connection);
diff --git a/java/src/IceInternal/RetryQueue.java b/java/src/IceInternal/RetryQueue.java
index 6a6de427dfe..023992f31f9 100644
--- a/java/src/IceInternal/RetryQueue.java
+++ b/java/src/IceInternal/RetryQueue.java
@@ -17,7 +17,7 @@ public class RetryQueue
     }
 
     synchronized public void
-    add(OutgoingAsync outAsync, int interval)
+    add(OutgoingAsyncMessageCallback outAsync, int interval)
     {
         RetryTask task = new RetryTask(this, outAsync);
         task.setFuture(_instance.timer().schedule(task, interval, java.util.concurrent.TimeUnit.MILLISECONDS));
diff --git a/java/src/IceInternal/RetryTask.java b/java/src/IceInternal/RetryTask.java
index e4f3a6c26fe..2c991f65819 100644
--- a/java/src/IceInternal/RetryTask.java
+++ b/java/src/IceInternal/RetryTask.java
@@ -11,7 +11,7 @@ package IceInternal;
 
 class RetryTask implements Runnable
 {
-    RetryTask(RetryQueue queue, OutgoingAsync outAsync)
+    RetryTask(RetryQueue queue, OutgoingAsyncMessageCallback outAsync)
     {
         _queue = queue;
         _outAsync = outAsync;
@@ -23,14 +23,7 @@ class RetryTask implements Runnable
     {
         if(_queue.remove(this))
         {
-            try
-            {
-                _outAsync.invoke(false);
-            }
-            catch(Ice.LocalException ex)
-            {
-                _outAsync.invokeExceptionAsync(ex);
-            }
+            _outAsync.processRetry(false);
         }
     }
 
@@ -38,7 +31,7 @@ class RetryTask implements Runnable
     destroy()
     {
         _future.cancel(false);
-        _outAsync.invokeExceptionAsync(new Ice.CommunicatorDestroyedException());
+        _outAsync.processRetry(true);
     }
 
     public void setFuture(java.util.concurrent.Future<?> future)
@@ -47,6 +40,6 @@ class RetryTask implements Runnable
     }
 
     private final RetryQueue _queue;
-    private final OutgoingAsync _outAsync;
+    private final OutgoingAsyncMessageCallback _outAsync;
     private java.util.concurrent.Future<?> _future;
 }