outgoing.abort

8 files changed, 96 insertions, 21 deletions

diff --git a/java/src/Ice/ConnectionI.java b/java/src/Ice/ConnectionI.java
index 3983c3a183e..5d985429f03 100644
--- a/java/src/Ice/ConnectionI.java
+++ b/java/src/Ice/ConnectionI.java
@@ -644,27 +644,43 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
 	_batchStream.swap(os);
 
 	//
-	// _batchStream now belongs to the caller, until
-	// finishBatchRequest() is called.
+	// The batch stream now belongs to the caller, until
+	// finishBatchRequest() or abortBatchRequest() is called.
 	//
     }
 
     public synchronized void
     finishBatchRequest(IceInternal.BasicStream os)
     {
-        if(_exception != null)
-        {
-            throw _exception;
-        }
+	//
+	// Get the batch stream back and increment the number of
+	// requests in the batch.
+        //
+	_batchStream.swap(os);
+	++_batchRequestNum;
 
-        assert(_state > StateNotValidated);
-	assert(_state < StateClosing);
+	//
+	// Notify about the batch stream not being in use anymore.
+	//
+	assert(_batchStreamInUse);
+        _batchStreamInUse = false;
+	notifyAll();
+    }
 
-        _batchStream.swap(os); // Get the batch stream back.
-	++_batchRequestNum; // Increment the number of requests in the batch.
+    public synchronized void
+    abortBatchRequest()
+    {
+	//
+	// Destroy and reset the batch stream and batch count. We
+	// cannot safe old requests in the batch stream, as they might
+	// be corrupted due to incomplete marshaling.
+	//
+	_batchStream.destroy();
+	_batchStream = new IceInternal.BasicStream(_instance);
+	_batchRequestNum = 0;
 
 	//
-	// Give the ConnectionI back.
+	// Notify about the batch stream not being in use anymore.
 	//
 	assert(_batchStreamInUse);
         _batchStreamInUse = false;
@@ -1565,6 +1581,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
 		    registerWithPool();
 		    unregisterWithPool();
 		}
+
 		break;
             }
         }
diff --git a/java/src/IceInternal/Outgoing.java b/java/src/IceInternal/Outgoing.java
index ea5b54f89c2..c294c0702b0 100644
--- a/java/src/IceInternal/Outgoing.java
+++ b/java/src/IceInternal/Outgoing.java
@@ -12,7 +12,8 @@ package IceInternal;
 public final class Outgoing
 {
     public
-    Outgoing(Ice.ConnectionI connection, Reference ref, String operation, Ice.OperationMode mode, java.util.Map context)
+    Outgoing(Ice.ConnectionI connection, Reference ref, String operation, Ice.OperationMode mode,
+	     java.util.Map context)
     {
         _connection = connection;
         _reference = ref;
@@ -25,8 +26,8 @@ public final class Outgoing
 
     //
     // Do NOT use a finalizer, this would cause a severe performance
-    // penalty! We must make sure that destroy() is called instead,
-    // to reclaim resources.
+    // penalty! We must make sure that destroy() is called instead, to
+    // reclaim resources.
     //
     public void
     destroy()
@@ -38,7 +39,6 @@ public final class Outgoing
 	assert(_os != null);
         _os.destroy();
 	_os = null;
-
     }
 
     //
@@ -65,6 +65,8 @@ public final class Outgoing
     invoke()
         throws NonRepeatable
     {
+	assert(_state == StateUnsent);
+
         _os.endWriteEncaps();
 
         switch(_reference.mode)
@@ -193,12 +195,13 @@ public final class Outgoing
 		//
 		// For oneway and datagram requests, the connection
 		// object never calls back on this object. Therefore
-		// we don't need to lock the mutex, keep track of
-		// state, or save exceptions. We simply let all
-		// exceptions from sending propagate to the caller,
-		// because such exceptions can be retried without
-		// violating "at-most-once".
+		// we don't need to lock the mutex or save
+		// exceptions. We simply let all exceptions from
+		// sending propagate to the caller, because such
+		// exceptions can be retried without violating
+		// "at-most-once".
 		//
+		_state = StateInProgress;
 		_connection.sendRequest(_os, null);
                 break;
             }
@@ -211,6 +214,7 @@ public final class Outgoing
 		// regular oneways and datagrams (see comment above)
 		// apply.
 		//
+		_state = StateInProgress;
                 _connection.finishBatchRequest(_os);
                 break;
             }
@@ -219,6 +223,33 @@ public final class Outgoing
         return true;
     }
 
+    public void
+    abort(Ice.LocalException ex)
+        throws NonRepeatable
+    {
+	assert(_state == StateUnsent);
+
+	//
+	// If we didn't finish a batch oneway or datagram request, we
+	// must notify the connection about that we give up ownership
+	// of the batch stream.
+	//
+	if(_reference.mode == Reference.ModeBatchOneway || _reference.mode == Reference.ModeBatchDatagram)
+	{
+	    _connection.abortBatchRequest();
+
+	    //
+	    // If we abort a batch requests, we cannot retry, because
+	    // not only the batch request that caused the problem will
+	    // be aborted, but all other requests in the batch as
+	    // well.
+	    //
+	    throw new NonRepeatable(ex);
+	}
+
+	throw ex;
+    }
+
     public synchronized void
     finished(BasicStream is)
     {
diff --git a/java/test/Ice/operations/AllTests.java b/java/test/Ice/operations/AllTests.java
index 496b377b447..a53b1c88aad 100644
--- a/java/test/Ice/operations/AllTests.java
+++ b/java/test/Ice/operations/AllTests.java
@@ -53,6 +53,12 @@ public class AllTests
 	    TwowaysAMI.twowaysAMI(cl);
 	    TwowaysAMI.twowaysAMI(derived);
 	    System.out.println("ok");
+
+	    System.out.print("testing batch oneway operations... ");
+	    System.out.flush();
+	    BatchOneways.batchOneways(cl);
+	    BatchOneways.batchOneways(derived);
+	    System.out.println("ok");
 	}
 
         return cl;
diff --git a/java/test/Ice/operations/Client.java b/java/test/Ice/operations/Client.java
index 2f3fd8a2e77..ccd8f4d1675 100644
--- a/java/test/Ice/operations/Client.java
+++ b/java/test/Ice/operations/Client.java
@@ -47,6 +47,13 @@ public class Client
 	    properties.setProperty("Ice.ThreadPool.Client.Size", "2");
 	    properties.setProperty("Ice.ThreadPool.Client.SizeWarn", "0");
 
+	    //
+	    // We must set MessageSizeMax to an explicit values,
+	    // because we run tests to check whether
+	    // Ice.MemoryLimitException is raised as expected.
+	    //
+	    properties.setProperty("Ice.MessageSizeMax", "100");
+
             communicator = Ice.Util.initialize(argsH);
             status = run(argsH.value, communicator);
         }
diff --git a/java/test/Ice/operations/MyDerivedClassI.java b/java/test/Ice/operations/MyDerivedClassI.java
index f1ca02f6b97..ae5f1172181 100644
--- a/java/test/Ice/operations/MyDerivedClassI.java
+++ b/java/test/Ice/operations/MyDerivedClassI.java
@@ -293,6 +293,11 @@ public final class MyDerivedClassI extends Test.MyDerivedClass
         return r;
     }
 
+    public void
+    opByteSOneway(byte[] s, Ice.Current current)
+    {
+    }
+
     public java.util.Map
     opContext(Ice.Current current)
     {
diff --git a/java/test/Ice/operations/Test.ice b/java/test/Ice/operations/Test.ice
index a819ad845a5..c57cb879968 100644
--- a/java/test/Ice/operations/Test.ice
+++ b/java/test/Ice/operations/Test.ice
@@ -145,8 +145,9 @@ dictionary<string, MyEnum> StringMyEnumD;
 
     IntS opIntS(IntS s);
 
-    Ice::Context opContext();
+    void opByteSOneway(ByteS s);
 
+    Ice::Context opContext();
 };
 
 ["ami"] class MyDerivedClass extends MyClass
diff --git a/java/test/Ice/operationsAMD/MyDerivedClassI.java b/java/test/Ice/operationsAMD/MyDerivedClassI.java
index 36f3bd1192b..68a6732f75e 100644
--- a/java/test/Ice/operationsAMD/MyDerivedClassI.java
+++ b/java/test/Ice/operationsAMD/MyDerivedClassI.java
@@ -331,6 +331,12 @@ public final class MyDerivedClassI extends Test.MyDerivedClass
     }
 
     public void
+    opByteSOneway_async(Test.AMD_MyClass_opByteSOneway cb, byte[] s, Ice.Current current)
+    {
+        cb.ice_response();
+    }
+
+    public void
     opContext_async(Test.AMD_MyClass_opContext cb, Ice.Current current)
     {
         cb.ice_response(current.ctx);
diff --git a/java/test/Ice/operationsAMD/TestAMD.ice b/java/test/Ice/operationsAMD/TestAMD.ice
index e6f7f4f205d..650fce73cd8 100644
--- a/java/test/Ice/operationsAMD/TestAMD.ice
+++ b/java/test/Ice/operationsAMD/TestAMD.ice
@@ -143,6 +143,8 @@ dictionary<string, MyEnum> StringMyEnumD;
 
     IntS opIntS(IntS s);
 
+    void opByteSOneway(ByteS s);
+
     StringStringD opContext();
 
 };