Fixed non-thread safe AMD dispatch, fixes #448 (#449)

The caching of the output stream which was added back to solve #414 made the
dispatch of AMD requests non thread-safe if a dispatch interceptor was installed
and if the interceptor retried the dispatch. Multiple continuation could run
concurrently and eventually re-use the cached output stream to marshal multiple
responses.

3 files changed, 31 insertions, 1 deletions

diff --git a/java-compat/test/src/main/java/test/Ice/interceptor/AMDInterceptorI.java b/java-compat/test/src/main/java/test/Ice/interceptor/AMDInterceptorI.java
index a6a855b0595..db208419d44 100644
--- a/java-compat/test/src/main/java/test/Ice/interceptor/AMDInterceptorI.java
+++ b/java-compat/test/src/main/java/test/Ice/interceptor/AMDInterceptorI.java
@@ -70,6 +70,15 @@ class AMDInterceptorI extends InterceptorI implements Ice.DispatchInterceptorAsy
 
             request.getCurrent().ctx.put("retry", "no");
         }
+        else if(current.ctx.get("retry") != null && current.ctx.get("retry").equals("yes"))
+        {
+            //
+            // Retry the dispatch to ensure that abandoning the result of the dispatch
+            // works fine and is thread-safe
+            //
+            _servant.ice_dispatch(request);
+            _servant.ice_dispatch(request);
+        }
 
         _lastStatus = _servant.ice_dispatch(request, this);
 
diff --git a/java-compat/test/src/main/java/test/Ice/interceptor/Client.java b/java-compat/test/src/main/java/test/Ice/interceptor/Client.java
index 0661f9e2117..2c95ccfdc21 100644
--- a/java-compat/test/src/main/java/test/Ice/interceptor/Client.java
+++ b/java-compat/test/src/main/java/test/Ice/interceptor/Client.java
@@ -119,6 +119,16 @@ public class Client extends test.TestHelper
         test(prx.amdAddWithRetry(33, 12) == 45);
         test(interceptor.getLastOperation().equals("amdAddWithRetry"));
         test(!interceptor.getLastStatus());
+        {
+            java.util.Map<String, String> ctx = new java.util.HashMap<>();
+            ctx.put("retry", "yes");
+            for(int i = 0; i < 10; ++i)
+            {
+                test(prx.amdAdd(33, 12, ctx) == 45);
+                test(interceptor.getLastOperation().equals("amdAdd"));
+                test(!interceptor.getLastStatus());
+            }
+        }
         out.println("ok");
         out.print("testing user exception... ");
         out.flush();
diff --git a/java-compat/test/src/main/java/test/Ice/interceptor/MyObjectI.java b/java-compat/test/src/main/java/test/Ice/interceptor/MyObjectI.java
index 1f60cb0adb8..cebd1882ccd 100644
--- a/java-compat/test/src/main/java/test/Ice/interceptor/MyObjectI.java
+++ b/java-compat/test/src/main/java/test/Ice/interceptor/MyObjectI.java
@@ -65,6 +65,7 @@ class MyObjectI extends _MyObjectDisp
     public void
     amdAdd_async(final AMD_MyObject_amdAdd cb, final int x, final int y, Ice.Current current)
     {
+        final boolean retry = current.ctx.get("retry") != null && current.ctx.get("retry").equals("yes");
         Thread thread = new Thread()
             {
                 @Override
@@ -78,7 +79,17 @@ class MyObjectI extends _MyObjectDisp
                     catch(InterruptedException e)
                     {
                     }
-                    cb.ice_response(x + y);
+                    try
+                    {
+                        cb.ice_response(x + y);
+                    }
+                    catch(Ice.ResponseSentException ex)
+                    {
+                        if(!retry)
+                        {
+                            throw ex;
+                        }
+                    }
                 }
             };