svn commit: r1027667 - /couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl

[fd...@apache.org]

Author: fdmanana
Date: Tue Oct 26 18:11:53 2010
New Revision: 1027667

URL: http://svn.apache.org/viewvc?rev=1027667&view=rev
Log:
Make max_size parameter optional.

Modified:
    couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl

Modified: couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl
URL: http://svn.apache.org/viewvc/couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl?rev=1027667&r1=1027666&r2=1027667&view=diff
==============================================================================
--- couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl (original)
+++ couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl Tue Oct 26 18:11:53 2010
@@ -59,8 +59,8 @@ close(Wq) ->
 
 init(Options) ->
     Q = #q{
-        max_size = couch_util:get_value(max_size, Options),
-        max_items = couch_util:get_value(max_items, Options),
+        max_size = couch_util:get_value(max_size, Options, nil),
+        max_items = couch_util:get_value(max_items, Options, nil),
         multi_workers = couch_util:get_value(multi_workers, Options, false)
     },
     {ok, Q}.
@@ -71,11 +71,11 @@ terminate(_Reason, #q{work_waiters=Worke
 
     
 handle_call({queue, Item}, From, #q{work_waiters = []} = Q0) ->
-    Q = Q0#q{size = Q0#q.size + byte_size(term_to_binary(Item)),
+    Q = Q0#q{size = increment_queue_size(Q0, Item),
                 items = Q0#q.items + 1,
                 queue = queue:in(Item, Q0#q.queue)},
     case (Q#q.size >= Q#q.max_size) orelse
-        (is_integer(Q#q.max_items) andalso (Q#q.items >= Q#q.max_items)) of
+            (Q#q.items >= Q#q.max_items) of
     true ->
         {noreply, Q#q{blocked = [From | Q#q.blocked]}};
     false ->
@@ -153,3 +153,10 @@ code_change(_OldVsn, State, _Extra) ->
 
 handle_info(X, Q) ->
     {stop, X, Q}.
+
+increment_queue_size(#q{max_size = nil, size = Size}, _Item) ->
+    Size;
+increment_queue_size(#q{size = Size}, Item) when is_binary(Item) ->
+    Size + byte_size(Item);
+increment_queue_size(#q{size = Size}, Item) ->
+    Size + byte_size(term_to_binary(Item)).