Java & C# port

1 files changed, 79 insertions, 66 deletions

diff --git a/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java b/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
index 8d2d2be737c..b64c56b8d99 100644
--- a/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
+++ b/java/src/IceInternal/CommunicatorBatchOutgoingAsync.java
@@ -9,7 +9,7 @@
 
 package IceInternal;
 
-public class CommunicatorBatchOutgoingAsync extends BatchOutgoingAsync
+public class CommunicatorBatchOutgoingAsync extends Ice.AsyncResult
 {
     public CommunicatorBatchOutgoingAsync(Ice.Communicator communicator, Instance instance, String operation,
                                           CallbackBase callback)
@@ -27,100 +27,113 @@ public class CommunicatorBatchOutgoingAsync extends BatchOutgoingAsync
         // Assume all connections are flushed synchronously.
         //
         _sentSynchronously = true;
+
+        //
+        // Attach observer
+        //
+        _observer = ObserverHelper.get(instance, operation);
     }
 
-    public void flushConnection(Ice.Connection con)
+    public void flushConnection(Ice.ConnectionI con)
     {
-        synchronized(_monitor)
+        class BatchOutgoingAsyncI extends BatchOutgoingAsync
         {
-            ++_useCount;
-        }
-        con.begin_flushBatchRequests(_cb);
-    }
+            public 
+            BatchOutgoingAsyncI()
+            {
+                super(CommunicatorBatchOutgoingAsync.this._communicator, 
+                      CommunicatorBatchOutgoingAsync.this._instance, 
+                      CommunicatorBatchOutgoingAsync.this._operation, 
+                      null);
+            }
 
-    public void ready()
-    {
-        check(null, null, true);
-    }
+            public boolean 
+            __sent(Ice.ConnectionI con)
+            {
+                if(_remoteObserver != null)
+                {
+                    _remoteObserver.detach();
+                    _remoteObserver = null;
+                }
+                check(false);
+                return false;
+            }
 
-    private void completed(Ice.AsyncResult r)
-    {
-        Ice.Connection con = r.getConnection();
-        assert(con != null);
+            public void 
+            __finished(Ice.LocalException ex, boolean sent)
+            {
+                if(_remoteObserver != null)
+                {
+                    _remoteObserver.detach();
+                    _remoteObserver = null;
+                }
+                check(false);
+            }
 
-        try
+            public void
+            __attachRemoteObserver(Ice.ConnectionInfo info, Ice.Endpoint endpt)
+            {
+                if(CommunicatorBatchOutgoingAsync.this._observer != null)
+                {
+                    _remoteObserver = CommunicatorBatchOutgoingAsync.this._observer.getRemoteObserver(info, endpt);
+                    if(_remoteObserver != null)
+                    {
+                        _remoteObserver.attach();
+                    }
+                }
+            }
+        };
+
+        synchronized(_monitor)
         {
-            con.end_flushBatchRequests(r);
-            assert(false); // completed() should only be called when an exception occurs.
+            ++_useCount;
         }
-        catch(Ice.LocalException ex)
+        
+        int status = con.flushAsyncBatchRequests(new BatchOutgoingAsyncI());
+        if((status & AsyncStatus.Sent) > 0)
         {
-            check(r, ex, false);
+            _sentSynchronously = false;
         }
     }
 
-    private void sent(Ice.AsyncResult r)
+    public void ready()
     {
-        check(r, null, r.sentSynchronously());
+        check(true);
     }
 
-    private void check(Ice.AsyncResult r, Ice.LocalException ex, boolean userThread)
+    private void check(boolean userThread)
     {
-        boolean done = false;
-
         synchronized(_monitor)
         {
             assert(_useCount > 0);
-            --_useCount;
-
-            //
-            // We report that the communicator flush request was sent synchronously
-            // if all of the connection flush requests are sent synchronously.
-            //
-            if((r != null && !r.sentSynchronously()) || ex != null)
+            if(--_useCount > 0)
             {
-                _sentSynchronously = false;
+                return;
             }
-
-            if(_useCount == 0)
+            
+            if(_observer != null)
             {
-                done = true;
-                _state |= Done | OK | Sent;
-                _monitor.notifyAll();
+                _observer.detach();
+                _observer = null;
             }
-        }
 
-        if(done)
-        {
-            //
-            // sentSynchronously_ is immutable here.
-            //
-            if(!_sentSynchronously && userThread)
-            {
-                __sentAsync();
-            }
-            else
-            {
-                assert(_sentSynchronously == userThread); // sentSynchronously && !userThread is impossible.
-                __sent();
-            }
+            _state |= Done | OK | Sent;
+            _monitor.notifyAll();
         }
-    }
-
-    private int _useCount;
 
-    private Ice.Callback _cb = new Ice.Callback()
-    {
-        @Override
-        public void completed(Ice.AsyncResult r)
+        //
+        // sentSynchronously_ is immutable here.
+        //
+        if(!_sentSynchronously && userThread)
         {
-            CommunicatorBatchOutgoingAsync.this.completed(r);
+            __sentAsync();
         }
-
-        @Override
-        public void sent(Ice.AsyncResult r)
+        else
         {
-            CommunicatorBatchOutgoingAsync.this.sent(r);
+            assert(_sentSynchronously == userThread); // sentSynchronously && !userThread is impossible.
+            __sentInternal();
         }
-    };
+    }
+
+    private int _useCount;
 }