summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorFilipe David Borba Manana <fdmanana@apache.org>2010-08-21 20:27:22 +0000
committerFilipe David Borba Manana <fdmanana@apache.org>2010-08-21 20:27:22 +0000
commit1c9aacd6729b5b5fbfcc2f8d3473faa2ec9a7dd9 (patch)
treea795a24541f99174d96dae1747029ba4b671bddf /src
parent58e2e130afbe692e216a0943907b7c26b1d36e8b (diff)
Added some whitespaces to make the code easier to the eye.
git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@987826 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'src')
-rw-r--r--src/couchdb/couch_work_queue.erl59
1 files changed, 38 insertions, 21 deletions
diff --git a/src/couchdb/couch_work_queue.erl b/src/couchdb/couch_work_queue.erl
index 637858b0..07fab2cc 100644
--- a/src/couchdb/couch_work_queue.erl
+++ b/src/couchdb/couch_work_queue.erl
@@ -13,36 +13,46 @@
-module(couch_work_queue).
-behaviour(gen_server).
--export([new/1,queue/2,dequeue/1,dequeue/2,close/1]).
--export([init/1, terminate/2, handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
+% public API
+-export([new/1, queue/2, dequeue/1, dequeue/2, close/1]).
+
+% gen_server callbacks
+-export([init/1, terminate/2]).
+-export([handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
-record(q, {
- queue=queue:new(),
- blocked=[],
+ queue = queue:new(),
+ blocked = [],
max_size,
max_items,
- items=0,
- size=0,
- work_waiters=[],
- close_on_dequeue=false,
- multi_workers=false
+ items = 0,
+ size = 0,
+ work_waiters = [],
+ close_on_dequeue = false,
+ multi_workers = false
}).
+
new(Options) ->
gen_server:start_link(couch_work_queue, Options, []).
+
queue(Wq, Item) ->
gen_server:call(Wq, {queue, Item}, infinity).
+
dequeue(Wq) ->
dequeue(Wq, all).
+
dequeue(Wq, MaxItems) ->
- try gen_server:call(Wq, {dequeue, MaxItems}, infinity)
+ try
+ gen_server:call(Wq, {dequeue, MaxItems}, infinity)
catch
_:_ -> closed
end.
+
close(Wq) ->
gen_server:cast(Wq, close).
@@ -55,23 +65,26 @@ init(Options) ->
},
{ok, Q}.
+
terminate(_Reason, #q{work_waiters=Workers}) ->
lists:foreach(fun({W, _}) -> gen_server:reply(W, closed) end, Workers).
+
-handle_call({queue, Item}, From, #q{work_waiters=[]}=Q0) ->
- Q = Q0#q{size=Q0#q.size + byte_size(term_to_binary(Item)),
- items=Q0#q.items + 1,
- queue=queue:in(Item, Q0#q.queue)},
+handle_call({queue, Item}, From, #q{work_waiters = []} = Q0) ->
+ Q = Q0#q{size = Q0#q.size + byte_size(term_to_binary(Item)),
+ items = Q0#q.items + 1,
+ queue = queue:in(Item, Q0#q.queue)},
case (Q#q.size >= Q#q.max_size) orelse
(Q#q.items >= Q#q.max_items) of
true ->
- {noreply, Q#q{blocked=[From | Q#q.blocked]}};
+ {noreply, Q#q{blocked = [From | Q#q.blocked]}};
false ->
{reply, ok, Q}
end;
+
handle_call({queue, Item}, _From, #q{work_waiters = [{W, _Max} | Rest]} = Q) ->
gen_server:reply(W, {ok, [Item]}),
- {reply, ok, Q#q{work_waiters=Rest}};
+ {reply, ok, Q#q{work_waiters = Rest}};
handle_call({dequeue, Max}, From, Q) ->
#q{work_waiters = Workers, multi_workers = Multi, items = Count} = Q,
@@ -89,6 +102,7 @@ handle_call({dequeue, Max}, From, Q) ->
end
end.
+
deliver_queue_items(Max, Q) ->
#q{
queue = Queue,
@@ -112,23 +126,26 @@ deliver_queue_items(Max, Q) ->
end
end.
+
dequeue_items(0, Queue, Blocked, DequeuedAcc) ->
{lists:reverse(DequeuedAcc), Queue, Blocked};
+
dequeue_items(NumItems, Queue, Blocked, DequeuedAcc) ->
{{value, Item}, Queue2} = queue:out(Queue),
- case Blocked of
+ Blocked2 = case Blocked of
[] ->
Blocked2 = Blocked;
- [From|Blocked2] ->
+ [From | Blocked2] ->
gen_server:reply(From, ok)
end,
- dequeue_items(NumItems-1, Queue2, Blocked2, [Item | DequeuedAcc]).
+ dequeue_items(NumItems - 1, Queue2, Blocked2, [Item | DequeuedAcc]).
-handle_cast(close, #q{items=0}=Q) ->
+handle_cast(close, #q{items = 0} = Q) ->
{stop, normal, Q};
+
handle_cast(close, Q) ->
- {noreply, Q#q{close_on_dequeue=true}}.
+ {noreply, Q#q{close_on_dequeue = true}}.
code_change(_OldVsn, State, _Extra) ->