added release-pending-sends

2 files changed, 22 insertions, 5 deletions

diff --git a/src/clj/clojure/core.clj b/src/clj/clojure/core.clj
index 4cb7fd66..c6e81bad 100644
--- a/src/clj/clojure/core.clj
+++ b/src/clj/clojure/core.clj
@@ -1074,6 +1074,15 @@
   [#^clojure.lang.Agent a f & args]
     (. a (dispatch f args true)))
 
+(defn release-pending-sends 
+  "Normally, actions sent directly or indirectly during another action
+  are held until the action completes (changes the agent's
+  state). This function can be used to dispatch any pending sent
+  actions immediately. This has no impact on actions sent during a
+  transaction, which are still held until commit. If no action is
+  occurring, does nothing. Returns the number of actions dispatched."  
+  [] (clojure.lang.Agent/releasePendingSends))
+
 (defn add-watch
   "Experimental.
   Adds a watcher to an agent. Whenever the agent runs an action, any
diff --git a/src/jvm/clojure/lang/Agent.java b/src/jvm/clojure/lang/Agent.java
index 55b97df8..e4500174 100644
--- a/src/jvm/clojure/lang/Agent.java
+++ b/src/jvm/clojure/lang/Agent.java
@@ -84,11 +84,7 @@ static class Action implements Runnable{
 
 			if(!hadError)
 				{
-				for(ISeq s = nested.get().seq(); s != null; s = s.rest())
-					{
-					Action a = (Action) s.first();
-					a.agent.enqueue(a);
-					}
+				releasePendingSends();
 				}
 
 			boolean popped = false;
@@ -232,4 +228,16 @@ public Agent removeWatch(Object watcher) throws Exception{
 	return this;
 }
 
+static public int releasePendingSends(){
+	IPersistentVector sends = nested.get();
+	if(sends == null)
+		return 0;
+	for(int i=0;i<sends.count();i++)
+		{
+		Action a = (Action) sends.valAt(i);
+		a.agent.enqueue(a);
+		}
+	nested.set(PersistentVector.EMPTY);
+	return sends.count();
+}
 }