diff options
Diffstat (limited to 'apps/couch/src/couch_rep_changes_feed.erl')
| -rw-r--r-- | apps/couch/src/couch_rep_changes_feed.erl | 247 |
1 files changed, 174 insertions, 73 deletions
diff --git a/apps/couch/src/couch_rep_changes_feed.erl b/apps/couch/src/couch_rep_changes_feed.erl index 032f62a3..36fe82aa 100644 --- a/apps/couch/src/couch_rep_changes_feed.erl +++ b/apps/couch/src/couch_rep_changes_feed.erl @@ -18,6 +18,7 @@ -export([start_link/4, next/1, stop/1]). -define(BUFFER_SIZE, 1000). +-define(DOC_IDS_FILTER_NAME, "_doc_ids"). -include("couch_db.hrl"). -include_lib("ibrowse/include/ibrowse.hrl"). @@ -33,9 +34,15 @@ count = 0, partial_chunk = <<>>, reply_to = nil, - rows = queue:new() + rows = queue:new(), + doc_ids = nil }). +-import(couch_util, [ + get_value/2, + get_value/3 +]). + start_link(Parent, Source, StartSeq, PostProps) -> gen_server:start_link(?MODULE, [Parent, Source, StartSeq, PostProps], []). @@ -46,9 +53,9 @@ stop(Server) -> catch gen_server:call(Server, stop), ok. -init([Parent, #http_db{}=Source, Since, PostProps]) -> +init([Parent, #http_db{headers = Headers0} = Source, Since, PostProps]) -> process_flag(trap_exit, true), - Feed = case couch_util:get_value(<<"continuous">>, PostProps, false) of + Feed = case get_value(<<"continuous">>, PostProps, false) of false -> normal; true -> @@ -60,51 +67,59 @@ init([Parent, #http_db{}=Source, Since, PostProps]) -> {"since", Since}, {"feed", Feed} ], - QS = case couch_util:get_value(<<"filter">>, PostProps) of + {QS, Method, Body, Headers} = case get_value(<<"doc_ids">>, PostProps) of undefined -> - BaseQS; - FilterName -> - {Params} = couch_util:get_value(<<"query_params">>, PostProps, {[]}), - lists:foldr( - fun({K, V}, QSAcc) -> - Ks = couch_util:to_list(K), - case proplists:is_defined(Ks, QSAcc) of - true -> - QSAcc; - false -> - [{Ks, V} | QSAcc] - end - end, - [{"filter", FilterName} | BaseQS], - Params - ) + {maybe_add_filter_qs_params(PostProps, BaseQS), get, nil, Headers0}; + DocIds when is_list(DocIds) -> + Headers1 = [{"Content-Type", "application/json"} | Headers0], + QS1 = [{"filter", ?l2b(?DOC_IDS_FILTER_NAME)} | BaseQS], + {QS1, post, {[{<<"doc_ids">>, DocIds}]}, Headers1} end, Pid = couch_rep_httpc:spawn_link_worker_process(Source), Req = Source#http_db{ + method = Method, + body = Body, resource = "_changes", qs = QS, conn = Pid, options = [{stream_to, {self(), once}}] ++ lists:keydelete(inactivity_timeout, 1, Source#http_db.options), - headers = Source#http_db.headers -- [{"Accept-Encoding", "gzip"}] + headers = Headers -- [{"Accept-Encoding", "gzip"}] }, {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Req), Args = [Parent, Req, Since, PostProps], + State = #state{ + conn = Pid, + last_seq = Since, + reqid = ReqId, + init_args = Args, + doc_ids = get_value(<<"doc_ids">>, PostProps, nil) + }, receive {ibrowse_async_headers, ReqId, "200", _} -> ibrowse:stream_next(ReqId), - {ok, #state{conn=Pid, last_seq=Since, reqid=ReqId, init_args=Args}}; - {ibrowse_async_headers, ReqId, Code, Hdrs} when Code=="301"; Code=="302" -> - stop_link_worker(Pid), - Url2 = couch_rep_httpc:redirect_url(Hdrs, Req#http_db.url), - Req2 = couch_rep_httpc:redirected_request(Req, Url2), - Pid2 = couch_rep_httpc:spawn_link_worker_process(Req2), - Req3 = Req2#http_db{conn = Pid2}, - {ibrowse_req_id, ReqId2} = couch_rep_httpc:request(Req3), - Args2 = [Parent, Req3, Since, PostProps], - receive {ibrowse_async_headers, ReqId2, "200", _} -> - {ok, #state{conn=Pid2, last_seq=Since, reqid=ReqId2, init_args=Args2}} + {ok, State}; + {ibrowse_async_headers, ReqId, Code, Hdrs} + when Code =:= "301"; Code =:= "302"; Code =:= "303" -> + {ReqId2, Req2} = redirect_req(Req, Code, Hdrs), + receive + {ibrowse_async_headers, ReqId2, "200", _} -> + {ok, State#state{ + conn = Req2#http_db.conn, + reqid = ReqId2, + init_args = [Parent, Req2, Since, PostProps]}}; + {ibrowse_async_headers, ReqId2, "405", _} when Method =:= post -> + {ReqId3, Req3} = req_no_builtin_doc_ids(Req2, ReqId2), + receive + {ibrowse_async_headers, ReqId3, "200", _} -> + {ok, State#state{ + conn = Req3#http_db.conn, + reqid = ReqId3, + init_args = [Parent, Req3, Since, PostProps]}} + after 30000 -> + {stop, changes_timeout} + end after 30000 -> {stop, changes_timeout} end; @@ -113,7 +128,30 @@ init([Parent, #http_db{}=Source, Since, PostProps]) -> ?LOG_INFO("source doesn't have _changes, trying _all_docs_by_seq", []), Self = self(), BySeqPid = spawn_link(fun() -> by_seq_loop(Self, Source, Since) end), - {ok, #state{last_seq=Since, changes_loop=BySeqPid, init_args=Args}}; + {ok, State#state{changes_loop = BySeqPid}}; + {ibrowse_async_headers, ReqId, "405", _} when Method =:= post -> + {ReqId2, Req2} = req_no_builtin_doc_ids(Req, ReqId), + receive + {ibrowse_async_headers, ReqId2, "200", _} -> + {ok, State#state{ + conn = Req2#http_db.conn, + reqid = ReqId2, + init_args = [Parent, Req2, Since, PostProps]}}; + {ibrowse_async_headers, ReqId, Code, Hdrs} + when Code =:= "301"; Code =:= "302"; Code =:= "303" -> + {ReqId3, Req3} = redirect_req(Req2, Code, Hdrs), + receive + {ibrowse_async_headers, ReqId3, "200", _} -> + {ok, State#state{ + conn = Req3#http_db.conn, + reqid = ReqId3, + init_args = [Parent, Req3, Since, PostProps]}} + after 30000 -> + {stop, changes_timeout} + end + after 30000 -> + {stop, changes_timeout} + end; {ibrowse_async_headers, ReqId, Code, _} -> {stop, {changes_error_code, list_to_integer(Code)}} after 10000 -> @@ -123,11 +161,17 @@ init([Parent, #http_db{}=Source, Since, PostProps]) -> init([_Parent, Source, Since, PostProps] = InitArgs) -> process_flag(trap_exit, true), Server = self(), + Filter = case get_value(<<"doc_ids">>, PostProps) of + undefined -> + ?b2l(get_value(<<"filter">>, PostProps, <<>>)); + DocIds when is_list(DocIds) -> + ?DOC_IDS_FILTER_NAME + end, ChangesArgs = #changes_args{ style = all_docs, since = Since, - filter = ?b2l(couch_util:get_value(<<"filter">>, PostProps, <<>>)), - feed = case couch_util:get_value(<<"continuous">>, PostProps, false) of + filter = Filter, + feed = case get_value(<<"continuous">>, PostProps, false) of true -> "continuous"; false -> @@ -138,7 +182,7 @@ init([_Parent, Source, Since, PostProps] = InitArgs) -> ChangesPid = spawn_link(fun() -> ChangesFeedFun = couch_changes:handle_changes( ChangesArgs, - {json_req, filter_json_req(Source, PostProps)}, + {json_req, filter_json_req(Filter, Source, PostProps)}, Source ), ChangesFeedFun(fun({change, Change, _}, _) -> @@ -149,29 +193,49 @@ init([_Parent, Source, Since, PostProps] = InitArgs) -> end), {ok, #state{changes_loop=ChangesPid, init_args=InitArgs}}. -filter_json_req(Db, PostProps) -> - case couch_util:get_value(<<"filter">>, PostProps) of +maybe_add_filter_qs_params(PostProps, BaseQS) -> + case get_value(<<"filter">>, PostProps) of undefined -> - {[]}; + BaseQS; FilterName -> - {Query} = couch_util:get_value(<<"query_params">>, PostProps, {[]}), - {ok, Info} = couch_db:get_db_info(Db), - % simulate a request to db_name/_changes - {[ - {<<"info">>, {Info}}, - {<<"id">>, null}, - {<<"method">>, 'GET'}, - {<<"path">>, [couch_db:name(Db), <<"_changes">>]}, - {<<"query">>, {[{<<"filter">>, FilterName} | Query]}}, - {<<"headers">>, []}, - {<<"body">>, []}, - {<<"peer">>, <<"replicator">>}, - {<<"form">>, []}, - {<<"cookie">>, []}, - {<<"userCtx">>, couch_util:json_user_ctx(Db)} - ]} + {Params} = get_value(<<"query_params">>, PostProps, {[]}), + lists:foldr( + fun({K, V}, QSAcc) -> + Ks = couch_util:to_list(K), + case proplists:is_defined(Ks, QSAcc) of + true -> + QSAcc; + false -> + [{Ks, V} | QSAcc] + end + end, + [{"filter", FilterName} | BaseQS], + Params + ) end. +filter_json_req([], _Db, _PostProps) -> + {[]}; +filter_json_req(?DOC_IDS_FILTER_NAME, _Db, PostProps) -> + {[{<<"doc_ids">>, get_value(<<"doc_ids">>, PostProps)}]}; +filter_json_req(FilterName, Db, PostProps) -> + {Query} = get_value(<<"query_params">>, PostProps, {[]}), + {ok, Info} = couch_db:get_db_info(Db), + % simulate a request to db_name/_changes + {[ + {<<"info">>, {Info}}, + {<<"id">>, null}, + {<<"method">>, 'GET'}, + {<<"path">>, [couch_db:name(Db), <<"_changes">>]}, + {<<"query">>, {[{<<"filter">>, FilterName} | Query]}}, + {<<"headers">>, []}, + {<<"body">>, []}, + {<<"peer">>, <<"replicator">>}, + {<<"form">>, []}, + {<<"cookie">>, []}, + {<<"userCtx">>, couch_util:json_user_ctx(Db)} + ]}. + handle_call({add_change, Row}, From, State) -> handle_add_change(Row, From, State); @@ -191,6 +255,10 @@ handle_info({ibrowse_async_response, Id, {error, sel_conn_closed}}, #state{reqid=Id}=State) -> handle_retry(State); +handle_info({ibrowse_async_response, Id, {error, connection_closed}}, + #state{reqid=Id}=State) -> + handle_retry(State); + handle_info({ibrowse_async_response, Id, {error,E}}, #state{reqid=Id}=State) -> {stop, {error, E}, State}; @@ -240,12 +308,9 @@ code_change(_OldVsn, State, _Extra) -> %internal funs handle_add_change(Row, From, #state{reply_to=nil} = State) -> - #state{ - count = Count, - rows = Rows - } = State, - NewState = State#state{count=Count+1, rows=queue:in(Row,Rows)}, - if Count < ?BUFFER_SIZE -> + {Rows2, Count2} = queue_changes_row(Row, State), + NewState = State#state{count = Count2, rows = Rows2}, + if Count2 =< ?BUFFER_SIZE -> {reply, ok, NewState}; true -> {noreply, NewState#state{changes_from=From}} @@ -274,11 +339,10 @@ handle_headers(200, _, State) -> maybe_stream_next(State), {noreply, State}; handle_headers(Code, Hdrs, #state{init_args = InitArgs} = State) - when Code =:= 301 ; Code =:= 302 -> + when Code =:= 301 ; Code =:= 302 ; Code =:= 303 -> stop_link_worker(State#state.conn), - [Parent, #http_db{url = Url1} = Source, Since, PostProps] = InitArgs, - Url = couch_rep_httpc:redirect_url(Hdrs, Url1), - Source2 = couch_rep_httpc:redirected_request(Source, Url), + [Parent, Source, Since, PostProps] = InitArgs, + Source2 = couch_rep_httpc:redirected_request(Code, Hdrs, Source), Pid2 = couch_rep_httpc:spawn_link_worker_process(Source2), Source3 = Source2#http_db{conn = Pid2}, {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Source3), @@ -298,21 +362,17 @@ handle_messages([<<"]">>, <<"\"last_seq\":", _/binary>>], State) -> handle_feed_completion(State); handle_messages([<<"{\"last_seq\":", _/binary>>], State) -> handle_feed_completion(State); -handle_messages([Chunk|Rest], State) -> - #state{ - count = Count, - partial_chunk = Partial, - rows = Rows - } = State, +handle_messages([Chunk|Rest], #state{partial_chunk = Partial} = State) -> NewState = try Row = {Props} = decode_row(<<Partial/binary, Chunk/binary>>), case State of #state{reply_to=nil} -> + {Rows2, Count2} = queue_changes_row(Row, State), State#state{ - count = Count+1, last_seq = couch_util:get_value(<<"seq">>, Props), partial_chunk = <<>>, - rows=queue:in(Row,Rows) + rows = Rows2, + count = Count2 }; #state{count=0, reply_to=From}-> gen_server:reply(From, [Row]), @@ -400,3 +460,44 @@ stop_link_worker(Conn) when is_pid(Conn) -> catch ibrowse:stop_worker_process(Conn); stop_link_worker(_) -> ok. + +redirect_req(#http_db{conn = WorkerPid} = Req, Code, Headers) -> + stop_link_worker(WorkerPid), + Req2 = couch_rep_httpc:redirected_request(Code, Headers, Req), + WorkerPid2 = couch_rep_httpc:spawn_link_worker_process(Req2), + Req3 = Req2#http_db{conn = WorkerPid2}, + {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Req3), + {ReqId, Req3}. + +req_no_builtin_doc_ids(#http_db{conn = WorkerPid, qs = QS} = Req, ReqId) -> + % CouchDB versions prior to 1.1.0 don't have the builtin filter _doc_ids + % and don't allow POSTing to /database/_changes + purge_req_messages(ReqId), + stop_link_worker(WorkerPid), + Req2 = Req#http_db{method = get, qs = lists:keydelete("filter", 1, QS)}, + WorkerPid2 = couch_rep_httpc:spawn_link_worker_process(Req2), + Req3 = Req2#http_db{conn = WorkerPid2}, + {ibrowse_req_id, ReqId2} = couch_rep_httpc:request(Req3), + {ReqId2, Req3}. + +purge_req_messages(ReqId) -> + ibrowse:stream_next(ReqId), + receive + {ibrowse_async_response, ReqId, {error, _}} -> + ok; + {ibrowse_async_response, ReqId, _Data} -> + purge_req_messages(ReqId); + {ibrowse_async_response_end, ReqId} -> + ok + end. + +queue_changes_row(Row, #state{doc_ids = nil, count = Count, rows = Rows}) -> + {queue:in(Row, Rows), Count + 1}; +queue_changes_row({RowProps} = Row, + #state{doc_ids = Ids, count = Count, rows = Rows}) -> + case lists:member(get_value(<<"id">>, RowProps), Ids) of + true -> + {queue:in(Row, Rows), Count + 1}; + false -> + {Rows, Count} + end. |