src/couchdb/couch_native_process.erl - couchdb - Git at Google

 % 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.
	% 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.