Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tag: 0.10.2
Fetching contributors…

Cannot retrieve contributors at this time

393 lines (368 sloc) 11.845 kb
% 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.
%
% This file drew much inspiration from erlview, which was written by and
% copyright Michael McDaniel [http://autosys.us], and is also under APL 2.0
%
%
% This module provides the smallest possible native view-server.
% With this module in-place, you can add the following to your couch INI files:
% [native_query_servers]
% erlang={couch_native_process, start_link, []}
%
% Which will then allow following example map function to be used:
%
% fun({Doc}) ->
% % Below, we emit a single record - the _id as key, null as value
% DocId = proplists:get_value(Doc, <<"_id">>, null),
% Emit(DocId, null)
% end.
%
% which should be roughly the same as the javascript:
% emit(doc._id, null);
%
% This module exposes enough functions such that a native erlang server can
% act as a fully-fleged view server, but no 'helper' functions specifically
% for simplifying your erlang view code. It is expected other third-party
% extensions will evolve which offer useful layers on top of this view server
% to help simplify your view code.
-module(couch_native_process).
-export([start_link/0]).
-export([set_timeout/2, prompt/2, stop/1]).
-export([loop/0]).
-define(STATE, native_proc_state).
-record(evstate, {funs=[], query_config=[], list_pid=nil, timeout=5000}).
-include("couch_db.hrl").
start_link() ->
{ok, spawn_link(?MODULE, loop, [])}.
stop(Pid) ->
Pid ! {stop, self()},
receive
{ok, Pid} -> ok
after 1000 ->
throw({error, timeout})
end.
prompt(Pid, Data) ->
Pid ! {prompt, self(), Data},
receive
{ok, Pid, Resp} -> Resp
after 1000 ->
throw({error, timeout})
end.
set_timeout(Pid, TimeOut) ->
Pid ! {set_timeout, self(), TimeOut},
receive
{ok, Pid} -> ok
after 1000 ->
throw({error, timeout})
end.
loop() ->
receive
{prompt, From, Data} ->
case (catch prompt(Data)) of
{error, Reason} ->
From ! {ok, self(), {[{error, Reason}]}};
{NewState, Resp} ->
put(?STATE, NewState),
From ! {ok, self(), Resp},
loop()
end;
{set_timeout, From, TimeOut} ->
NewState = case get(?STATE) of
undefined -> #evstate{timeout=TimeOut};
State -> State#evstate{timeout=TimeOut}
end,
put(?STATE, NewState),
From ! {ok, self()},
loop();
{stop, From} ->
From ! {ok, self()}
end.
prompt(Data) when is_list(Data) ->
case get(?STATE) of
undefined ->
State = #evstate{},
put(?STATE, State);
State ->
State
end,
case is_pid(State#evstate.list_pid) of
true ->
case hd(Data) of
<<"list_row">> -> ok;
<<"list_end">> -> ok;
_ -> throw({error, query_server_error})
end;
_ ->
ok % Not listing
end,
run(State, to_binary(Data)).
run(_, [<<"reset">>]) ->
{#evstate{}, true};
run(_, [<<"reset">>, QueryConfig]) ->
{#evstate{query_config=QueryConfig}, true};
run(#evstate{funs=Funs}=State, [<<"add_fun">> , BinFunc]) ->
FunInfo = makefun(State, BinFunc),
{State#evstate{funs=Funs ++ [FunInfo]}, true};
run(State, [<<"map_doc">> , Doc]) ->
Resp = lists:map(fun({Sig, Fun}) ->
erlang:put(Sig, []),
Fun(Doc),
lists:reverse(erlang:get(Sig))
end, State#evstate.funs),
{State, Resp};
run(State, [<<"reduce">>, Funs, KVs]) ->
{Keys, Vals} =
lists:foldl(fun([K, V], {KAcc, VAcc}) ->
{[K | KAcc], [V | VAcc]}
end, {[], []}, KVs),
Keys2 = lists:reverse(Keys),
Vals2 = lists:reverse(Vals),
{State, catch reduce(State, Funs, Keys2, Vals2, false)};
run(State, [<<"rereduce">>, Funs, Vals]) ->
{State, catch reduce(State, Funs, null, Vals, true)};
run(State, [<<"validate">>, BFun, NDoc, ODoc, Ctx]) ->
{_Sig, Fun} = makefun(State, BFun),
{State, catch Fun(NDoc, ODoc, Ctx)};
run(State, [<<"filter">>, Docs, Req]) ->
{_Sig, Fun} = hd(State#evstate.funs),
Resp = lists:map(fun(Doc) ->
case (catch Fun(Doc, Req)) of
true -> true;
_ -> false
end
end, Docs),
{State, [true, Resp]};
run(State, [<<"show">>, BFun, Doc, Req]) ->
{_Sig, Fun} = makefun(State, BFun),
Resp = case (catch Fun(Doc, Req)) of
FunResp when is_list(FunResp) ->
FunResp;
FunResp when is_tuple(FunResp), size(FunResp) == 1 ->
[<<"resp">>, FunResp];
FunResp ->
FunResp
end,
{State, Resp};
run(State, [<<"update">>, BFun, Doc, Req]) ->
{_Sig, Fun} = makefun(State, BFun),
Resp = case (catch Fun(Doc, Req)) of
[JsonDoc, JsonResp] ->
[<<"up">>, JsonDoc, JsonResp]
end,
{State, Resp};
run(State, [<<"list">>, Head, Req]) ->
{Sig, Fun} = hd(State#evstate.funs),
% This is kinda dirty
case is_function(Fun, 2) of
false -> throw({error, render_error});
true -> ok
end,
Self = self(),
SpawnFun = fun() ->
LastChunk = (catch Fun(Head, Req)),
case start_list_resp(Self, Sig) of
started ->
receive
{Self, list_row, _Row} -> ignore;
{Self, list_end} -> ignore
after State#evstate.timeout ->
throw({timeout, list_cleanup_pid})
end;
_ ->
ok
end,
LastChunks =
case erlang:get(Sig) of
undefined -> [LastChunk];
OtherChunks -> [LastChunk | OtherChunks]
end,
Self ! {self(), list_end, lists:reverse(LastChunks)}
end,
erlang:put(do_trap, process_flag(trap_exit, true)),
Pid = spawn_link(SpawnFun),
Resp =
receive
{Pid, start, Chunks, JsonResp} ->
[<<"start">>, Chunks, JsonResp]
after State#evstate.timeout ->
throw({timeout, list_start})
end,
{State#evstate{list_pid=Pid}, Resp};
run(#evstate{list_pid=Pid}=State, [<<"list_row">>, Row]) when is_pid(Pid) ->
Pid ! {self(), list_row, Row},
receive
{Pid, chunks, Data} ->
{State, [<<"chunks">>, Data]};
{Pid, list_end, Data} ->
receive
{'EXIT', Pid, normal} -> ok
after State#evstate.timeout ->
throw({timeout, list_cleanup})
end,
process_flag(trap_exit, erlang:get(do_trap)),
{State#evstate{list_pid=nil}, [<<"end">>, Data]}
after State#evstate.timeout ->
throw({timeout, list_row})
end;
run(#evstate{list_pid=Pid}=State, [<<"list_end">>]) when is_pid(Pid) ->
Pid ! {self(), list_end},
Resp =
receive
{Pid, list_end, Data} ->
receive
{'EXIT', Pid, normal} -> ok
after State#evstate.timeout ->
throw({timeout, list_cleanup})
end,
[<<"end">>, Data]
after State#evstate.timeout ->
throw({timeout, list_end})
end,
process_flag(trap_exit, erlang:get(do_trap)),
{State#evstate{list_pid=nil}, Resp};
run(_, Unknown) ->
?LOG_ERROR("Native Process: Unknown command: ~p~n", [Unknown]),
throw({error, query_server_error}).
bindings(State, Sig) ->
Self = self(),
Log = fun(Msg) ->
?LOG_INFO(Msg, [])
end,
Emit = fun(Id, Value) ->
Curr = erlang:get(Sig),
erlang:put(Sig, [[Id, Value] | Curr])
end,
Start = fun(Headers) ->
erlang:put(list_headers, Headers)
end,
Send = fun(Chunk) ->
Curr =
case erlang:get(Sig) of
undefined -> [];
Else -> Else
end,
erlang:put(Sig, [Chunk | Curr])
end,
GetRow = fun() ->
case start_list_resp(Self, Sig) of
started ->
ok;
_ ->
Chunks =
case erlang:get(Sig) of
undefined -> [];
CurrChunks -> CurrChunks
end,
Self ! {self(), chunks, lists:reverse(Chunks)}
end,
erlang:put(Sig, []),
receive
{Self, list_row, Row} -> Row;
{Self, list_end} -> nil
after State#evstate.timeout ->
throw({timeout, list_pid_getrow})
end
end,
FoldRows = fun(Fun, Acc) -> foldrows(GetRow, Fun, Acc) end,
[
{'Log', Log},
{'Emit', Emit},
{'Start', Start},
{'Send', Send},
{'GetRow', GetRow},
{'FoldRows', FoldRows}
].
% thanks to erlview, via:
% http://erlang.org/pipermail/erlang-questions/2003-November/010544.html
makefun(State, Source) ->
Sig = erlang:md5(Source),
BindFuns = bindings(State, Sig),
{Sig, makefun(State, Source, BindFuns)}.
makefun(_State, Source, BindFuns) ->
FunStr = binary_to_list(Source),
{ok, Tokens, _} = erl_scan:string(FunStr),
Form = case (catch erl_parse:parse_exprs(Tokens)) of
{ok, [ParsedForm]} ->
ParsedForm;
{error, {LineNum, _Mod, [Mesg, Params]}}=Error ->
io:format(standard_error, "Syntax error on line: ~p~n", [LineNum]),
io:format(standard_error, "~s~p~n", [Mesg, Params]),
throw(Error)
end,
Bindings = lists:foldl(fun({Name, Fun}, Acc) ->
erl_eval:add_binding(Name, Fun, Acc)
end, erl_eval:new_bindings(), BindFuns),
{value, Fun, _} = erl_eval:expr(Form, Bindings),
Fun.
reduce(State, BinFuns, Keys, Vals, ReReduce) ->
Funs = case is_list(BinFuns) of
true ->
lists:map(fun(BF) -> makefun(State, BF) end, BinFuns);
_ ->
[makefun(State, BinFuns)]
end,
Reds = lists:map(fun({_Sig, Fun}) ->
Fun(Keys, Vals, ReReduce)
end, Funs),
[true, Reds].
foldrows(GetRow, ProcRow, Acc) ->
case GetRow() of
nil ->
{ok, Acc};
Row ->
case (catch ProcRow(Row, Acc)) of
{ok, Acc2} ->
foldrows(GetRow, ProcRow, Acc2);
{stop, Acc2} ->
{ok, Acc2}
end
end.
start_list_resp(Self, Sig) ->
case erlang:get(list_started) of
undefined ->
Headers =
case erlang:get(list_headers) of
undefined -> {[{<<"headers">>, {[]}}]};
CurrHdrs -> CurrHdrs
end,
Chunks =
case erlang:get(Sig) of
undefined -> [];
CurrChunks -> CurrChunks
end,
Self ! {self(), start, lists:reverse(Chunks), Headers},
erlang:put(list_started, true),
erlang:put(Sig, []),
started;
_ ->
ok
end.
to_binary({Data}) ->
Pred = fun({Key, Value}) ->
{to_binary(Key), to_binary(Value)}
end,
{lists:map(Pred, Data)};
to_binary(Data) when is_list(Data) ->
lists:map(fun to_binary/1, Data);
to_binary(null) ->
null;
to_binary(true) ->
true;
to_binary(false) ->
false;
to_binary(Data) when is_atom(Data) ->
list_to_binary(atom_to_list(Data));
to_binary(Data) ->
Data.
Jump to Line
Something went wrong with that request. Please try again.