2 files changed, 68 insertions, 20 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)
     {