From 6e209ea2aa7ba5e50038d800ca3272f81e1bb70b Mon Sep 17 00:00:00 2001
From: Ulf Wiger <ulf@feuerlabs.com>
Date: Sat, 15 Jun 2013 12:05:00 +0200
Subject: [PATCH] shorthand, stats & stateful for 'approve' queues

---
 src/jobs_server.erl | 25 +++++++++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

diff --git a/src/jobs_server.erl b/src/jobs_server.erl
index b2ebb1b..e114f4e 100644
--- a/src/jobs_server.erl
+++ b/src/jobs_server.erl
@@ -123,6 +123,9 @@ run(Type, Fun) when is_function(Fun, 1) ->
 
 -spec done(reg_obj()) -> ok.
 %%
+done({undefined, _}) ->
+    %% no monitoring on this job (e.g. from a {action, approve} queue)
+    ok;
 done(Opaque) ->
     gen_server:cast(?MODULE, {done, self(), del_info(Opaque)}).
 
@@ -408,6 +411,10 @@ normalize_options(Opts) ->
 		[{type, {producer, F}}];
 	   (passive) ->
 		[{type, {passive, fifo}}];
+	   (A) when A==approve; A==reject ->
+		[{type, {action, approve}}];
+	   ({action, A}) when A==approve; A==reject ->
+		[{type, {action, approve}}];
 	   ({K, V}) ->
 		[{K, V}]
 	end, Opts)).
@@ -618,8 +625,22 @@ i_handle_call({ask, Type}, From, #st{queues = Qs} = S) ->
     TS = timestamp(),
     {Qname, S1} = select_queue(Type, TS, S),
     case get_queue(Qname, Qs) of
-        #queue{type = #action{a = approve}} -> approve(From, []), {noreply, S1};
-	#queue{type = #action{a = reject}} ->  reject(From)     , {noreply, S1};
+        #queue{type = #action{a = approve}, stateful = undefined,
+	       approved = Approved} = Q ->
+	    S2 = S1#st{queues = lists:keyreplace(
+				  Qname, #queue.name, Qs,
+				  Q#queue{approved = Approved + 1})},
+	    approve(From, {undefined, []}), {noreply, S2};
+        #queue{type = #action{a = approve}, stateful = Stf,
+	       approved = Approved} = Q ->
+	    {Val, Stf1} = update_stateful(Stf, [], S1#st.info_f),
+	    S2 = S1#st{queues = lists:keyreplace(
+				  Qname, #queue.name, Qs,
+				  Q#queue{stateful = Stf1,
+					  approved = Approved + 1})},
+	    approve(From, {undefined, [{info, Val}]}), {noreply, S2};
+	#queue{type = #action{a = reject}} ->
+	    reject(From), {noreply, S1};
 	#queue{type = #producer{}} ->
 	    {reply, badarg, S1};
         #queue{} = Q ->