You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2014/02/05 00:44:18 UTC
[05/44] Remove src/couch
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ed98610c/src/couch/src/couch_util.erl
----------------------------------------------------------------------
diff --git a/src/couch/src/couch_util.erl b/src/couch/src/couch_util.erl
deleted file mode 100644
index d09211a..0000000
--- a/src/couch/src/couch_util.erl
+++ /dev/null
@@ -1,500 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-% http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
-
--module(couch_util).
-
--export([priv_dir/0, normpath/1]).
--export([should_flush/0, should_flush/1, to_existing_atom/1]).
--export([rand32/0, implode/2, collate/2, collate/3]).
--export([abs_pathname/1,abs_pathname/2, trim/1]).
--export([encodeBase64Url/1, decodeBase64Url/1]).
--export([validate_utf8/1, to_hex/1, parse_term/1, dict_find/3]).
--export([get_nested_json_value/2, json_user_ctx/1]).
--export([proplist_apply_field/2, json_apply_field/2]).
--export([to_binary/1, to_integer/1, to_list/1, url_encode/1]).
--export([verify/2,simple_call/2,shutdown_sync/1]).
--export([get_value/2, get_value/3]).
--export([md5/1, md5_init/0, md5_update/2, md5_final/1]).
--export([reorder_results/2]).
--export([url_strip_password/1]).
--export([encode_doc_id/1]).
--export([with_db/2]).
--export([rfc1123_date/0, rfc1123_date/1]).
--export([integer_to_boolean/1, boolean_to_integer/1]).
-
--include_lib("couch/include/couch_db.hrl").
-
-% arbitrarily chosen amount of memory to use before flushing to disk
--define(FLUSH_MAX_MEM, 10000000).
-
-priv_dir() ->
- case code:priv_dir(couch) of
- {error, bad_name} ->
- % small hack, in dev mode "app" is couchdb. Fixing requires
- % renaming src/couch to src/couch. Not really worth the hassle.
- % -Damien
- code:priv_dir(couchdb);
- Dir -> Dir
- end.
-
-% Normalize a pathname by removing .. and . components.
-normpath(Path) ->
- normparts(filename:split(Path), []).
-
-normparts([], Acc) ->
- filename:join(lists:reverse(Acc));
-normparts([".." | RestParts], [_Drop | RestAcc]) ->
- normparts(RestParts, RestAcc);
-normparts(["." | RestParts], Acc) ->
- normparts(RestParts, Acc);
-normparts([Part | RestParts], Acc) ->
- normparts(RestParts, [Part | Acc]).
-
-% works like list_to_existing_atom, except can be list or binary and it
-% gives you the original value instead of an error if no existing atom.
-to_existing_atom(V) when is_list(V) ->
- try list_to_existing_atom(V) catch _:_ -> V end;
-to_existing_atom(V) when is_binary(V) ->
- try list_to_existing_atom(?b2l(V)) catch _:_ -> V end;
-to_existing_atom(V) when is_atom(V) ->
- V.
-
-shutdown_sync(Pid) when not is_pid(Pid)->
- ok;
-shutdown_sync(Pid) ->
- MRef = erlang:monitor(process, Pid),
- try
- catch unlink(Pid),
- catch exit(Pid, shutdown),
- receive
- {'DOWN', MRef, _, _, _} ->
- ok
- end
- after
- erlang:demonitor(MRef, [flush])
- end.
-
-
-simple_call(Pid, Message) ->
- MRef = erlang:monitor(process, Pid),
- try
- Pid ! {self(), Message},
- receive
- {Pid, Result} ->
- Result;
- {'DOWN', MRef, _, _, Reason} ->
- exit(Reason)
- end
- after
- erlang:demonitor(MRef, [flush])
- end.
-
-validate_utf8(Data) when is_list(Data) ->
- validate_utf8(?l2b(Data));
-validate_utf8(Bin) when is_binary(Bin) ->
- validate_utf8_fast(Bin, 0).
-
-validate_utf8_fast(B, O) ->
- case B of
- <<_:O/binary>> ->
- true;
- <<_:O/binary, C1, _/binary>> when
- C1 < 128 ->
- validate_utf8_fast(B, 1 + O);
- <<_:O/binary, C1, C2, _/binary>> when
- C1 >= 194, C1 =< 223,
- C2 >= 128, C2 =< 191 ->
- validate_utf8_fast(B, 2 + O);
- <<_:O/binary, C1, C2, C3, _/binary>> when
- C1 >= 224, C1 =< 239,
- C2 >= 128, C2 =< 191,
- C3 >= 128, C3 =< 191 ->
- validate_utf8_fast(B, 3 + O);
- <<_:O/binary, C1, C2, C3, C4, _/binary>> when
- C1 >= 240, C1 =< 244,
- C2 >= 128, C2 =< 191,
- C3 >= 128, C3 =< 191,
- C4 >= 128, C4 =< 191 ->
- validate_utf8_fast(B, 4 + O);
- _ ->
- false
- end.
-
-to_hex([]) ->
- [];
-to_hex(Bin) when is_binary(Bin) ->
- to_hex(binary_to_list(Bin));
-to_hex([H|T]) ->
- [to_digit(H div 16), to_digit(H rem 16) | to_hex(T)].
-
-to_digit(N) when N < 10 -> $0 + N;
-to_digit(N) -> $a + N-10.
-
-
-parse_term(Bin) when is_binary(Bin) ->
- parse_term(binary_to_list(Bin));
-parse_term(List) ->
- {ok, Tokens, _} = erl_scan:string(List ++ "."),
- erl_parse:parse_term(Tokens).
-
-get_value(Key, List) ->
- get_value(Key, List, undefined).
-
-get_value(Key, List, Default) ->
- case lists:keysearch(Key, 1, List) of
- {value, {Key,Value}} ->
- Value;
- false ->
- Default
- end.
-
-get_nested_json_value({Props}, [Key|Keys]) ->
- case couch_util:get_value(Key, Props, nil) of
- nil -> throw({not_found, <<"missing json key: ", Key/binary>>});
- Value -> get_nested_json_value(Value, Keys)
- end;
-get_nested_json_value(Value, []) ->
- Value;
-get_nested_json_value(_NotJSONObj, _) ->
- throw({not_found, json_mismatch}).
-
-proplist_apply_field(H, L) ->
- {R} = json_apply_field(H, {L}),
- R.
-
-json_apply_field(H, {L}) ->
- json_apply_field(H, L, []).
-json_apply_field({Key, NewValue}, [{Key, _OldVal} | Headers], Acc) ->
- json_apply_field({Key, NewValue}, Headers, Acc);
-json_apply_field({Key, NewValue}, [{OtherKey, OtherVal} | Headers], Acc) ->
- json_apply_field({Key, NewValue}, Headers, [{OtherKey, OtherVal} | Acc]);
-json_apply_field({Key, NewValue}, [], Acc) ->
- {[{Key, NewValue}|Acc]}.
-
-json_user_ctx(#db{name=ShardName, user_ctx=Ctx}) ->
- {[{<<"db">>, mem3:dbname(ShardName)},
- {<<"name">>,Ctx#user_ctx.name},
- {<<"roles">>,Ctx#user_ctx.roles}]}.
-
-
-% returns a random integer
-rand32() ->
- crypto:rand_uniform(0, 16#100000000).
-
-% given a pathname "../foo/bar/" it gives back the fully qualified
-% absolute pathname.
-abs_pathname(" " ++ Filename) ->
- % strip leading whitspace
- abs_pathname(Filename);
-abs_pathname([$/ |_]=Filename) ->
- Filename;
-abs_pathname(Filename) ->
- {ok, Cwd} = file:get_cwd(),
- {Filename2, Args} = separate_cmd_args(Filename, ""),
- abs_pathname(Filename2, Cwd) ++ Args.
-
-abs_pathname(Filename, Dir) ->
- Name = filename:absname(Filename, Dir ++ "/"),
- OutFilename = filename:join(fix_path_list(filename:split(Name), [])),
- % If the filename is a dir (last char slash, put back end slash
- case string:right(Filename,1) of
- "/" ->
- OutFilename ++ "/";
- "\\" ->
- OutFilename ++ "/";
- _Else->
- OutFilename
- end.
-
-% if this as an executable with arguments, seperate out the arguments
-% ""./foo\ bar.sh -baz=blah" -> {"./foo\ bar.sh", " -baz=blah"}
-separate_cmd_args("", CmdAcc) ->
- {lists:reverse(CmdAcc), ""};
-separate_cmd_args("\\ " ++ Rest, CmdAcc) -> % handle skipped value
- separate_cmd_args(Rest, " \\" ++ CmdAcc);
-separate_cmd_args(" " ++ Rest, CmdAcc) ->
- {lists:reverse(CmdAcc), " " ++ Rest};
-separate_cmd_args([Char|Rest], CmdAcc) ->
- separate_cmd_args(Rest, [Char | CmdAcc]).
-
-% Is a character whitespace?
-is_whitespace($\s) -> true;
-is_whitespace($\t) -> true;
-is_whitespace($\n) -> true;
-is_whitespace($\r) -> true;
-is_whitespace(_Else) -> false.
-
-
-% removes leading and trailing whitespace from a string
-trim(String) ->
- String2 = lists:dropwhile(fun is_whitespace/1, String),
- lists:reverse(lists:dropwhile(fun is_whitespace/1, lists:reverse(String2))).
-
-% takes a heirarchical list of dirs and removes the dots ".", double dots
-% ".." and the corresponding parent dirs.
-fix_path_list([], Acc) ->
- lists:reverse(Acc);
-fix_path_list([".."|Rest], [_PrevAcc|RestAcc]) ->
- fix_path_list(Rest, RestAcc);
-fix_path_list(["."|Rest], Acc) ->
- fix_path_list(Rest, Acc);
-fix_path_list([Dir | Rest], Acc) ->
- fix_path_list(Rest, [Dir | Acc]).
-
-
-implode(List, Sep) ->
- implode(List, Sep, []).
-
-implode([], _Sep, Acc) ->
- lists:flatten(lists:reverse(Acc));
-implode([H], Sep, Acc) ->
- implode([], Sep, [H|Acc]);
-implode([H|T], Sep, Acc) ->
- implode(T, Sep, [Sep,H|Acc]).
-
-
-drv_port() ->
- case get(couch_drv_port) of
- undefined ->
- Port = open_port({spawn, "couch_icu_driver"}, []),
- put(couch_drv_port, Port),
- Port;
- Port ->
- Port
- end.
-
-collate(A, B) ->
- collate(A, B, []).
-
-collate(A, B, Options) when is_binary(A), is_binary(B) ->
- Operation =
- case lists:member(nocase, Options) of
- true -> 1; % Case insensitive
- false -> 0 % Case sensitive
- end,
- SizeA = byte_size(A),
- SizeB = byte_size(B),
- Bin = <<SizeA:32/native, A/binary, SizeB:32/native, B/binary>>,
- [Result] = erlang:port_control(drv_port(), Operation, Bin),
- % Result is 0 for lt, 1 for eq and 2 for gt. Subtract 1 to return the
- % expected typical -1, 0, 1
- Result - 1.
-
-should_flush() ->
- should_flush(?FLUSH_MAX_MEM).
-
-should_flush(MemThreshHold) ->
- {memory, ProcMem} = process_info(self(), memory),
- BinMem = lists:foldl(fun({_Id, Size, _NRefs}, Acc) -> Size+Acc end,
- 0, element(2,process_info(self(), binary))),
- if ProcMem+BinMem > 2*MemThreshHold ->
- garbage_collect(),
- {memory, ProcMem2} = process_info(self(), memory),
- BinMem2 = lists:foldl(fun({_Id, Size, _NRefs}, Acc) -> Size+Acc end,
- 0, element(2,process_info(self(), binary))),
- ProcMem2+BinMem2 > MemThreshHold;
- true -> false end.
-
-encodeBase64Url(Url) ->
- Url1 = re:replace(base64:encode(Url), ["=+", $$], ""),
- Url2 = re:replace(Url1, "/", "_", [global]),
- re:replace(Url2, "\\+", "-", [global, {return, binary}]).
-
-decodeBase64Url(Url64) ->
- Url1 = re:replace(Url64, "-", "+", [global]),
- Url2 = re:replace(Url1, "_", "/", [global]),
- Padding = lists:duplicate((4 - iolist_size(Url2) rem 4) rem 4, $=),
- base64:decode(iolist_to_binary([Url2, Padding])).
-
-dict_find(Key, Dict, DefaultValue) ->
- case dict:find(Key, Dict) of
- {ok, Value} ->
- Value;
- error ->
- DefaultValue
- end.
-
-to_binary(V) when is_binary(V) ->
- V;
-to_binary(V) when is_list(V) ->
- try
- list_to_binary(V)
- catch
- _:_ ->
- list_to_binary(io_lib:format("~p", [V]))
- end;
-to_binary(V) when is_atom(V) ->
- list_to_binary(atom_to_list(V));
-to_binary(V) ->
- list_to_binary(io_lib:format("~p", [V])).
-
-to_integer(V) when is_integer(V) ->
- V;
-to_integer(V) when is_list(V) ->
- erlang:list_to_integer(V);
-to_integer(V) when is_binary(V) ->
- erlang:list_to_integer(binary_to_list(V)).
-
-to_list(V) when is_list(V) ->
- V;
-to_list(V) when is_binary(V) ->
- binary_to_list(V);
-to_list(V) when is_atom(V) ->
- atom_to_list(V);
-to_list(V) ->
- lists:flatten(io_lib:format("~p", [V])).
-
-url_encode(Bin) when is_binary(Bin) ->
- url_encode(binary_to_list(Bin));
-url_encode([H|T]) ->
- if
- H >= $a, $z >= H ->
- [H|url_encode(T)];
- H >= $A, $Z >= H ->
- [H|url_encode(T)];
- H >= $0, $9 >= H ->
- [H|url_encode(T)];
- H == $_; H == $.; H == $-; H == $: ->
- [H|url_encode(T)];
- true ->
- case lists:flatten(io_lib:format("~.16.0B", [H])) of
- [X, Y] ->
- [$%, X, Y | url_encode(T)];
- [X] ->
- [$%, $0, X | url_encode(T)]
- end
- end;
-url_encode([]) ->
- [].
-
-verify([X|RestX], [Y|RestY], Result) ->
- verify(RestX, RestY, (X bxor Y) bor Result);
-verify([], [], Result) ->
- Result == 0.
-
-verify(<<X/binary>>, <<Y/binary>>) ->
- verify(?b2l(X), ?b2l(Y));
-verify(X, Y) when is_list(X) and is_list(Y) ->
- case length(X) == length(Y) of
- true ->
- verify(X, Y, 0);
- false ->
- false
- end;
-verify(_X, _Y) -> false.
-
--spec md5(Data::(iolist() | binary())) -> Digest::binary().
-md5(Data) ->
- try crypto:md5(Data) catch error:_ -> erlang:md5(Data) end.
-
--spec md5_init() -> Context::binary().
-md5_init() ->
- try crypto:md5_init() catch error:_ -> erlang:md5_init() end.
-
--spec md5_update(Context::binary(), Data::(iolist() | binary())) ->
- NewContext::binary().
-md5_update(Ctx, D) ->
- try crypto:md5_update(Ctx,D) catch error:_ -> erlang:md5_update(Ctx,D) end.
-
--spec md5_final(Context::binary()) -> Digest::binary().
-md5_final(Ctx) ->
- try crypto:md5_final(Ctx) catch error:_ -> erlang:md5_final(Ctx) end.
-
-% linear search is faster for small lists, length() is 0.5 ms for 100k list
-reorder_results(Keys, SortedResults) when length(Keys) < 100 ->
- [couch_util:get_value(Key, SortedResults) || Key <- Keys];
-reorder_results(Keys, SortedResults) ->
- KeyDict = dict:from_list(SortedResults),
- [dict:fetch(Key, KeyDict) || Key <- Keys].
-
-url_strip_password(Url) ->
- re:replace(Url,
- "http(s)?://([^:]+):[^@]+@(.*)$",
- "http\\1://\\2:*****@\\3",
- [{return, list}]).
-
-encode_doc_id(#doc{id = Id}) ->
- encode_doc_id(Id);
-encode_doc_id(Id) when is_list(Id) ->
- encode_doc_id(?l2b(Id));
-encode_doc_id(<<"_design/", Rest/binary>>) ->
- "_design/" ++ url_encode(Rest);
-encode_doc_id(<<"_local/", Rest/binary>>) ->
- "_local/" ++ url_encode(Rest);
-encode_doc_id(Id) ->
- url_encode(Id).
-
-
-with_db(Db, Fun) when is_record(Db, db) ->
- Fun(Db);
-with_db(DbName, Fun) ->
- case couch_db:open_int(DbName, [{user_ctx, #user_ctx{roles=[<<"_admin">>]}}]) of
- {ok, Db} ->
- try
- Fun(Db)
- after
- catch couch_db:close(Db)
- end;
- Else ->
- throw(Else)
- end.
-
-rfc1123_date() ->
- {{YYYY,MM,DD},{Hour,Min,Sec}} = calendar:universal_time(),
- DayNumber = calendar:day_of_the_week({YYYY,MM,DD}),
- lists:flatten(
- io_lib:format("~s, ~2.2.0w ~3.s ~4.4.0w ~2.2.0w:~2.2.0w:~2.2.0w GMT",
- [day(DayNumber),DD,month(MM),YYYY,Hour,Min,Sec])).
-
-rfc1123_date(undefined) ->
- undefined;
-rfc1123_date(UniversalTime) ->
- {{YYYY,MM,DD},{Hour,Min,Sec}} = UniversalTime,
- DayNumber = calendar:day_of_the_week({YYYY,MM,DD}),
- lists:flatten(
- io_lib:format("~s, ~2.2.0w ~3.s ~4.4.0w ~2.2.0w:~2.2.0w:~2.2.0w GMT",
- [day(DayNumber),DD,month(MM),YYYY,Hour,Min,Sec])).
-
-%% day
-
-day(1) -> "Mon";
-day(2) -> "Tue";
-day(3) -> "Wed";
-day(4) -> "Thu";
-day(5) -> "Fri";
-day(6) -> "Sat";
-day(7) -> "Sun".
-
-%% month
-
-month(1) -> "Jan";
-month(2) -> "Feb";
-month(3) -> "Mar";
-month(4) -> "Apr";
-month(5) -> "May";
-month(6) -> "Jun";
-month(7) -> "Jul";
-month(8) -> "Aug";
-month(9) -> "Sep";
-month(10) -> "Oct";
-month(11) -> "Nov";
-month(12) -> "Dec".
-
-integer_to_boolean(1) ->
- true;
-integer_to_boolean(0) ->
- false.
-
-boolean_to_integer(true) ->
- 1;
-boolean_to_integer(false) ->
- 0.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ed98610c/src/couch/src/couch_uuids.erl
----------------------------------------------------------------------
diff --git a/src/couch/src/couch_uuids.erl b/src/couch/src/couch_uuids.erl
deleted file mode 100644
index 3065938..0000000
--- a/src/couch/src/couch_uuids.erl
+++ /dev/null
@@ -1,116 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-% http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
--module(couch_uuids).
--include_lib("couch/include/couch_db.hrl").
-
--behaviour(gen_server).
--behaviour(config_listener).
-
--export([start/0, stop/0]).
--export([new/0, random/0, utc_random/0]).
-
--export([init/1, terminate/2, code_change/3]).
--export([handle_call/3, handle_cast/2, handle_info/2]).
-
-% config_listener api
--export([handle_config_change/5]).
-
-start() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-stop() ->
- gen_server:cast(?MODULE, stop).
-
-new() ->
- gen_server:call(?MODULE, create).
-
-random() ->
- list_to_binary(couch_util:to_hex(crypto:rand_bytes(16))).
-
-utc_random() ->
- utc_suffix(couch_util:to_hex(crypto:rand_bytes(9))).
-
-utc_suffix(Suffix) ->
- Now = {_, _, Micro} = now(),
- Nowish = calendar:now_to_universal_time(Now),
- Nowsecs = calendar:datetime_to_gregorian_seconds(Nowish),
- Then = calendar:datetime_to_gregorian_seconds({{1970, 1, 1}, {0, 0, 0}}),
- Prefix = io_lib:format("~14.16.0b", [(Nowsecs - Then) * 1000000 + Micro]),
- list_to_binary(Prefix ++ Suffix).
-
-init([]) ->
- ok = config:listen_for_changes(?MODULE, nil),
- {ok, state()}.
-
-terminate(_Reason, _State) ->
- ok.
-
-handle_call(create, _From, random) ->
- {reply, random(), random};
-handle_call(create, _From, utc_random) ->
- {reply, utc_random(), utc_random};
-handle_call(create, _From, {utc_id, UtcIdSuffix}) ->
- {reply, utc_suffix(UtcIdSuffix), {utc_id, UtcIdSuffix}};
-handle_call(create, _From, {sequential, Pref, Seq}) ->
- Result = ?l2b(Pref ++ io_lib:format("~6.16.0b", [Seq])),
- case Seq >= 16#fff000 of
- true ->
- {reply, Result, {sequential, new_prefix(), inc()}};
- _ ->
- {reply, Result, {sequential, Pref, Seq + inc()}}
- end.
-
-handle_cast(change, _State) ->
- {noreply, state()};
-handle_cast(stop, State) ->
- {stop, normal, State};
-handle_cast(_Msg, State) ->
- {noreply, State}.
-
-handle_info({gen_event_EXIT, {config_listener, ?MODULE}, _Reason}, State) ->
- erlang:send_after(5000, self(), restart_config_listener),
- {noreply, State};
-handle_info(restart_config_listener, State) ->
- ok = config:listen_for_changes(?MODULE, nil),
- {noreply, State};
-handle_info(_Info, State) ->
- {noreply, State}.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-handle_config_change("uuids", _, _, _, _) ->
- {ok, gen_server:cast(?MODULE, change)};
-handle_config_change(_, _, _, _, _) ->
- {ok, nil}.
-
-new_prefix() ->
- couch_util:to_hex((crypto:rand_bytes(13))).
-
-inc() ->
- crypto:rand_uniform(1, 16#ffe).
-
-state() ->
- AlgoStr = config:get("uuids", "algorithm", "random"),
- case couch_util:to_existing_atom(AlgoStr) of
- random ->
- random;
- utc_random ->
- utc_random;
- utc_id ->
- UtcIdSuffix = config:get("uuids", "utc_id_suffix", ""),
- {utc_id, UtcIdSuffix};
- sequential ->
- {sequential, new_prefix(), inc()};
- Unknown ->
- throw({unknown_uuid_algorithm, Unknown})
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ed98610c/src/couch/src/couch_work_queue.erl
----------------------------------------------------------------------
diff --git a/src/couch/src/couch_work_queue.erl b/src/couch/src/couch_work_queue.erl
deleted file mode 100644
index ea871e2..0000000
--- a/src/couch/src/couch_work_queue.erl
+++ /dev/null
@@ -1,187 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-% http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
-
--module(couch_work_queue).
--behaviour(gen_server).
-
--include_lib("couch/include/couch_db.hrl").
-
-% public API
--export([new/1, queue/2, dequeue/1, dequeue/2, close/1, item_count/1, size/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 = [],
- max_size,
- max_items,
- 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) when is_binary(Item) ->
- gen_server:call(Wq, {queue, Item, byte_size(Item)}, infinity);
-queue(Wq, Item) ->
- gen_server:call(Wq, {queue, Item, ?term_size(Item)}, infinity).
-
-
-dequeue(Wq) ->
- dequeue(Wq, all).
-
-
-dequeue(Wq, MaxItems) ->
- try
- gen_server:call(Wq, {dequeue, MaxItems}, infinity)
- catch
- _:_ -> closed
- end.
-
-
-item_count(Wq) ->
- try
- gen_server:call(Wq, item_count, infinity)
- catch
- _:_ -> closed
- end.
-
-
-size(Wq) ->
- try
- gen_server:call(Wq, size, infinity)
- catch
- _:_ -> closed
- end.
-
-
-close(Wq) ->
- gen_server:cast(Wq, close).
-
-
-init(Options) ->
- Q = #q{
- 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, hibernate}.
-
-
-terminate(_Reason, #q{work_waiters=Workers}) ->
- lists:foreach(fun({W, _}) -> gen_server:reply(W, closed) end, Workers).
-
-
-handle_call({queue, Item, Size}, From, #q{work_waiters = []} = Q0) ->
- Q = Q0#q{size = Q0#q.size + Size,
- items = Q0#q.items + 1,
- queue = queue:in({Item, Size}, 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]}, hibernate};
- false ->
- {reply, ok, Q, hibernate}
- 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}, hibernate};
-
-handle_call({dequeue, Max}, From, Q) ->
- #q{work_waiters = Workers, multi_workers = Multi, items = Count} = Q,
- case {Workers, Multi} of
- {[_ | _], false} ->
- exit("Only one caller allowed to wait for this work at a time");
- {[_ | _], true} ->
- {noreply, Q#q{work_waiters=Workers ++ [{From, Max}]}};
- _ ->
- case Count of
- 0 ->
- {noreply, Q#q{work_waiters=Workers ++ [{From, Max}]}};
- C when C > 0 ->
- deliver_queue_items(Max, Q)
- end
- end;
-
-handle_call(item_count, _From, Q) ->
- {reply, Q#q.items, Q};
-
-handle_call(size, _From, Q) ->
- {reply, Q#q.size, Q}.
-
-
-deliver_queue_items(Max, Q) ->
- #q{
- queue = Queue,
- items = Count,
- size = Size,
- close_on_dequeue = Close,
- blocked = Blocked
- } = Q,
- case (Max =:= all) orelse (Max >= Count) of
- false ->
- {Items, Size2, Queue2, Blocked2} = dequeue_items(
- Max, Size, Queue, Blocked, []),
- Q2 = Q#q{
- items = Count - Max, size = Size2, blocked = Blocked2, queue = Queue2
- },
- {reply, {ok, Items}, Q2};
- true ->
- lists:foreach(fun(F) -> gen_server:reply(F, ok) end, Blocked),
- Q2 = Q#q{items = 0, size = 0, blocked = [], queue = queue:new()},
- Items = [Item || {Item, _} <- queue:to_list(Queue)],
- case Close of
- false ->
- {reply, {ok, Items}, Q2};
- true ->
- {stop, normal, {ok, Items}, Q2}
- end
- end.
-
-
-dequeue_items(0, Size, Queue, Blocked, DequeuedAcc) ->
- {lists:reverse(DequeuedAcc), Size, Queue, Blocked};
-
-dequeue_items(NumItems, Size, Queue, Blocked, DequeuedAcc) ->
- {{value, {Item, ItemSize}}, Queue2} = queue:out(Queue),
- case Blocked of
- [] ->
- Blocked2 = Blocked;
- [From | Blocked2] ->
- gen_server:reply(From, ok)
- end,
- dequeue_items(
- NumItems - 1, Size - ItemSize, Queue2, Blocked2, [Item | DequeuedAcc]).
-
-
-handle_cast(close, #q{items = 0} = Q) ->
- {stop, normal, Q};
-
-handle_cast(close, Q) ->
- {noreply, Q#q{close_on_dequeue = true}}.
-
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-handle_info(X, Q) ->
- {stop, X, Q}.