diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/couchdb/Makefile.am | 2 | ||||
| -rw-r--r-- | src/couchdb/couch_db.hrl | 12 | ||||
| -rw-r--r-- | src/couchdb/couch_httpd_db.erl | 265 | ||||
| -rw-r--r-- | src/couchdb/couch_query_servers.erl | 7 | ||||
| -rw-r--r-- | src/couchdb/couch_rep.erl | 6 | ||||
| -rw-r--r-- | src/couchdb/couch_rep_changes_feed.erl | 136 | ||||
| -rw-r--r-- | src/couchdb/couch_rep_missing_revs.erl | 20 |
7 files changed, 188 insertions, 260 deletions
diff --git a/src/couchdb/Makefile.am b/src/couchdb/Makefile.am index 9c7e9d4d..b5c84f33 100644 --- a/src/couchdb/Makefile.am +++ b/src/couchdb/Makefile.am @@ -30,6 +30,7 @@ source_files = \ couch.erl \ couch_app.erl \ couch_btree.erl \ + couch_changes.erl \ couch_config.erl \ couch_config_writer.erl \ couch_db.erl \ @@ -86,6 +87,7 @@ compiled_files = \ couch.beam \ couch_app.beam \ couch_btree.beam \ + couch_changes.beam \ couch_config.beam \ couch_config_writer.beam \ couch_db.beam \ diff --git a/src/couchdb/couch_db.hrl b/src/couchdb/couch_db.hrl index f74bade7..bb6b4f73 100644 --- a/src/couchdb/couch_db.hrl +++ b/src/couchdb/couch_db.hrl @@ -264,3 +264,15 @@ % small value used in revision trees to indicate the revision isn't stored -define(REV_MISSING, []). + +-record(changes_args, { + feed = "normal", + dir = fwd, + since = 0, + limit = 1000000000000000, + style = main_only, + heartbeat, + timeout, + filter = "", + include_docs = false +}). diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl index 9ad34752..1028e857 100644 --- a/src/couchdb/couch_httpd_db.erl +++ b/src/couchdb/couch_httpd_db.erl @@ -55,200 +55,62 @@ handle_request(#httpd{path_parts=[DbName|RestParts],method=Method, do_db_req(Req, Handler) end. -get_changes_timeout(Req, Resp) -> - DefaultTimeout = list_to_integer( - couch_config:get("httpd", "changes_timeout", "60000")), - case couch_httpd:qs_value(Req, "heartbeat") of - undefined -> - case couch_httpd:qs_value(Req, "timeout") of - undefined -> - {DefaultTimeout, fun() -> stop end}; - TimeoutList -> - {lists:min([DefaultTimeout, list_to_integer(TimeoutList)]), - fun() -> stop end} - end; - "true" -> - {DefaultTimeout, fun() -> send_chunk(Resp, "\n"), ok end}; - TimeoutList -> - {lists:min([DefaultTimeout, list_to_integer(TimeoutList)]), - fun() -> send_chunk(Resp, "\n"), ok end} - end. - - -start_sending_changes(_Resp, "continuous") -> - ok; -start_sending_changes(Resp, _Else) -> - send_chunk(Resp, "{\"results\":[\n"). - -handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) -> - FilterFun = make_filter_fun(Req, Db), - {ok, Info} = couch_db:get_db_info(Db), - Seq = proplists:get_value(update_seq, Info), - {Dir, StartSeq} = case couch_httpd:qs_value(Req, "descending", "false") of - "false" -> - {fwd, list_to_integer(couch_httpd:qs_value(Req, "since", "0"))}; - "true" -> - {rev, Seq}; - _Bad -> throw({bad_request, "descending must be true or false"}) +handle_changes_req(#httpd{method='GET'}=Req, Db) -> + MakeCallback = fun(Resp) -> + fun({change, Change, _}, "continuous") -> + send_chunk(Resp, [?JSON_ENCODE(Change) | "\n"]); + ({change, Change, Prepend}, _) -> + send_chunk(Resp, [Prepend, ?JSON_ENCODE(Change)]); + (start, "continuous") -> + ok; + (start, _) -> + send_chunk(Resp, "{\"results\":[\n"); + ({stop, EndSeq}, "continuous") -> + send_chunk( + Resp, + [?JSON_ENCODE({[{<<"last_seq">>, EndSeq}]}) | "\n"] + ), + end_json_response(Resp); + ({stop, EndSeq}, _) -> + send_chunk( + Resp, + io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq]) + ), + end_json_response(Resp); + (timeout, _) -> + send_chunk(Resp, "\n") + end end, - Limit = list_to_integer(couch_httpd:qs_value(Req, "limit", "1000000000000000")), - ResponseType = couch_httpd:qs_value(Req, "feed", "normal"), - if ResponseType == "continuous" orelse ResponseType == "longpoll" -> - {ok, Resp} = start_json_response(Req, 200), - start_sending_changes(Resp, ResponseType), - - Self = self(), - {ok, Notify} = couch_db_update_notifier:start_link( - fun({_, DbName0}) when DbName0 == DbName -> - Self ! db_updated; - (_) -> - ok - end), - {Timeout, TimeoutFun} = get_changes_timeout(Req, Resp), - couch_stats_collector:track_process_count(Self, - {httpd, clients_requesting_changes}), - try - keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, - TimeoutFun, ResponseType, Limit, FilterFun) - after - couch_db_update_notifier:stop(Notify), - get_rest_db_updated() % clean out any remaining update messages - end; - true -> + ChangesArgs = parse_changes_query(Req), + ChangesFun = couch_changes:handle_changes(ChangesArgs, Req, Db), + WrapperFun = case ChangesArgs#changes_args.feed of + "normal" -> + {ok, Info} = couch_db:get_db_info(Db), CurrentEtag = couch_httpd:make_etag(Info), - couch_httpd:etag_respond(Req, CurrentEtag, fun() -> - % send the etag - {ok, Resp} = start_json_response(Req, 200, [{"Etag", CurrentEtag}]), - start_sending_changes(Resp, ResponseType), - {ok, {_, LastSeq, _Prepend, _, _, _, _, _}} = - send_changes(Req, Resp, Db, Dir, StartSeq, <<"">>, "normal", - Limit, FilterFun), - end_sending_changes(Resp, LastSeq, ResponseType) - end) - end; + fun(FeedChangesFun) -> + couch_httpd:etag_respond( + Req, + CurrentEtag, + fun() -> + {ok, Resp} = couch_httpd:start_json_response( + Req, 200, [{"Etag", CurrentEtag}] + ), + FeedChangesFun(MakeCallback(Resp)) + end + ) + end; + _ -> + % "longpoll" or "continuous" + {ok, Resp} = couch_httpd:start_json_response(Req, 200), + fun(FeedChangesFun) -> + FeedChangesFun(MakeCallback(Resp)) + end + end, + WrapperFun(ChangesFun); handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) -> send_method_not_allowed(Req, "GET,HEAD"). -% waits for a db_updated msg, if there are multiple msgs, collects them. -wait_db_updated(Timeout, TimeoutFun) -> - receive db_updated -> get_rest_db_updated() - after Timeout -> - case TimeoutFun() of - ok -> wait_db_updated(Timeout, TimeoutFun); - stop -> stop - end - end. - -get_rest_db_updated() -> - receive db_updated -> get_rest_db_updated() - after 0 -> updated - end. - -end_sending_changes(Resp, EndSeq, "continuous") -> - send_chunk(Resp, [?JSON_ENCODE({[{<<"last_seq">>, EndSeq}]}) | "\n"]), - end_json_response(Resp); -end_sending_changes(Resp, EndSeq, _Else) -> - send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq])), - end_json_response(Resp). - -keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, - Db, StartSeq, Prepend, Timeout, TimeoutFun, ResponseType, Limit, Filter) -> - {ok, {_, EndSeq, Prepend2, _, _, _, NewLimit, _}} = send_changes(Req, Resp, Db, fwd, StartSeq, - Prepend, ResponseType, Limit, Filter), - couch_db:close(Db), - if - Limit > NewLimit, ResponseType == "longpoll" -> - end_sending_changes(Resp, EndSeq, ResponseType); - true -> - case wait_db_updated(Timeout, TimeoutFun) of - updated -> - case couch_db:open(DbName, [{user_ctx, UserCtx}]) of - {ok, Db2} -> - keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, - TimeoutFun, ResponseType, NewLimit, Filter); - _Else -> - end_sending_changes(Resp, EndSeq, ResponseType) - end; - stop -> - end_sending_changes(Resp, EndSeq, ResponseType) - end - end. - -changes_enumerator(DocInfos, {Db, _, _, FilterFun, Resp, "continuous", Limit, IncludeDocs}) -> - [#doc_info{id=Id, high_seq=Seq, revs=[#rev_info{deleted=Del,rev=Rev}|_]}|_] = DocInfos, - Results0 = FilterFun(DocInfos), - Results = [Result || Result <- Results0, Result /= null], - Go = if Limit =< 1 -> stop; true -> ok end, - case Results of - [] -> - {Go, {Db, Seq, nil, FilterFun, Resp, "continuous", Limit, IncludeDocs}}; - _ -> - send_chunk(Resp, [?JSON_ENCODE(changes_row(Db, Seq, Id, Del, Results, Rev, IncludeDocs)) - |"\n"]), - {Go, {Db, Seq, nil, FilterFun, Resp, "continuous", Limit-1, IncludeDocs}} - end; -changes_enumerator(DocInfos, {Db, _, Prepend, FilterFun, Resp, _, Limit, IncludeDocs}) -> - [#doc_info{id=Id, high_seq=Seq, revs=[#rev_info{deleted=Del,rev=Rev}|_]}|_] = DocInfos, - Results0 = FilterFun(DocInfos), - Results = [Result || Result <- Results0, Result /= null], - Go = if Limit =< 1 -> stop; true -> ok end, - case Results of - [] -> - {Go, {Db, Seq, Prepend, FilterFun, Resp, nil, Limit, IncludeDocs}}; - _ -> - send_chunk(Resp, [Prepend, ?JSON_ENCODE( - changes_row(Db, Seq, Id, Del, Results, Rev, IncludeDocs))]), - {Go, {Db, Seq, <<",\n">>, FilterFun, Resp, nil, Limit-1, IncludeDocs}} - end. - -changes_row(Db, Seq, Id, Del, Results, Rev, true) -> - {[{seq,Seq},{id,Id},{changes,Results}] ++ deleted_item(Del) ++ - couch_httpd_view:doc_member(Db, {Id, Rev})}; -changes_row(_, Seq, Id, Del, Results, _, false) -> - {[{seq,Seq},{id,Id},{changes,Results}] ++ deleted_item(Del)}. - -deleted_item(true) -> [{deleted,true}]; -deleted_item(_) -> []. - -send_changes(Req, Resp, Db, Dir, StartSeq, Prepend, ResponseType, Limit, FilterFun) -> - Style = list_to_existing_atom( - couch_httpd:qs_value(Req, "style", "main_only")), - IncludeDocs = list_to_existing_atom( - couch_httpd:qs_value(Req, "include_docs", "false")), - couch_db:changes_since(Db, Style, StartSeq, fun changes_enumerator/2, - [{dir, Dir}], {Db, StartSeq, Prepend, FilterFun, Resp, ResponseType, Limit, IncludeDocs}). - -make_filter_fun(Req, Db) -> - Filter = couch_httpd:qs_value(Req, "filter", ""), - case [list_to_binary(couch_httpd:unquote(Part)) - || Part <- string:tokens(Filter, "/")] of - [] -> - fun(DocInfos) -> - % doing this as a batch is more efficient for external filters - [{[{rev, couch_doc:rev_to_str(Rev)}]} || - #doc_info{revs=[#rev_info{rev=Rev}|_]} <- DocInfos] - end; - [DName, FName] -> - DesignId = <<"_design/", DName/binary>>, - DDoc = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []), - % validate that the ddoc has the filter fun - #doc{body={Props}} = DDoc, - couch_util:get_nested_json_value({Props}, [<<"filters">>, FName]), - fun(DocInfos) -> - Docs = [Doc || {ok, Doc} <- [ - {ok, _Doc} = couch_db:open_doc(Db, DInfo, [deleted, conflicts]) - || DInfo <- DocInfos]], - {ok, Passes} = couch_query_servers:filter_docs(Req, Db, DDoc, FName, Docs), - [{[{rev, couch_doc:rev_to_str(Rev)}]} - || #doc_info{revs=[#rev_info{rev=Rev}|_]} <- DocInfos, - Pass <- Passes, Pass == true] - end; - _Else -> - throw({bad_request, - "filter parameter must be of the form `designname/filtername`"}) - end. - handle_compact_req(#httpd{method='POST',path_parts=[DbName,_,Id|_]}=Req, _Db) -> ok = couch_view_compactor:start_compact(DbName, Id), send_json(Req, 202, {[{ok, true}]}); @@ -1188,6 +1050,33 @@ parse_doc_query(Req) -> end end, #doc_query_args{}, couch_httpd:qs(Req)). +parse_changes_query(Req) -> + lists:foldl(fun({Key, Value}, Args) -> + case {Key, Value} of + {"feed", _} -> + Args#changes_args{feed=Value}; + {"descending", "true"} -> + Args#changes_args{dir=rev}; + {"since", _} -> + Args#changes_args{since=list_to_integer(Value)}; + {"limit", _} -> + Args#changes_args{limit=list_to_integer(Value)}; + {"style", _} -> + Args#changes_args{style=list_to_existing_atom(Value)}; + {"heartbeat", "true"} -> + Args#changes_args{heartbeat=true}; + {"heartbeat", _} -> + Args#changes_args{heartbeat=list_to_integer(Value)}; + {"timeout", _} -> + Args#changes_args{timeout=list_to_integer(Value)}; + {"include_docs", "true"} -> + Args#changes_args{include_docs=true}; + {"filter", _} -> + Args#changes_args{filter=Value}; + _Else -> % unknown key value pair, ignore. + Args + end + end, #changes_args{}, couch_httpd:qs(Req)). extract_header_rev(Req, ExplicitRev) when is_binary(ExplicitRev) or is_list(ExplicitRev)-> extract_header_rev(Req, couch_doc:parse_rev(ExplicitRev)); diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl index a90681c5..3095b199 100644 --- a/src/couchdb/couch_query_servers.erl +++ b/src/couchdb/couch_query_servers.erl @@ -183,7 +183,12 @@ json_doc(Doc) -> couch_doc:to_json_obj(Doc, [revs]). filter_docs(Req, Db, DDoc, FName, Docs) -> - JsonReq = couch_httpd_external:json_req_obj(Req, Db), + JsonReq = case Req of + {json_req, JsonObj} -> + JsonObj; + #httpd{} = HttpReq -> + couch_httpd_external:json_req_obj(HttpReq, Db) + end, JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs], JsonCtx = couch_util:json_user_ctx(Db), [true, Passes] = ddoc_prompt(DDoc, [<<"filters">>, FName], [JsonDocs, JsonReq, JsonCtx]), diff --git a/src/couchdb/couch_rep.erl b/src/couchdb/couch_rep.erl index 4bad4a9f..c0010a94 100644 --- a/src/couchdb/couch_rep.erl +++ b/src/couchdb/couch_rep.erl @@ -451,7 +451,11 @@ make_replication_id({Props}, UserCtx) -> % Port = mochiweb_socket_server:get(couch_httpd, port), Src = get_rep_endpoint(UserCtx, proplists:get_value(<<"source">>, Props)), Tgt = get_rep_endpoint(UserCtx, proplists:get_value(<<"target">>, Props)), - Base = couch_util:to_hex(erlang:md5(term_to_binary([HostName, Src, Tgt]))), + Filter = proplists:get_value(<<"filter">>, Props), + QueryParams = proplists:get_value(<<"query_params">>, Props), + Base = couch_util:to_hex(erlang:md5( + term_to_binary([HostName, Src, Tgt, Filter, QueryParams]) + )), Extension = maybe_append_options( [<<"continuous">>, <<"create_target">>], Props), {Base, Extension}. diff --git a/src/couchdb/couch_rep_changes_feed.erl b/src/couchdb/couch_rep_changes_feed.erl index cdfed6a0..67cfabe4 100644 --- a/src/couchdb/couch_rep_changes_feed.erl +++ b/src/couchdb/couch_rep_changes_feed.erl @@ -53,11 +53,35 @@ init([_Parent, #http_db{}=Source, Since, PostProps] = Args) -> true -> continuous end, + BaseQS = [ + {"style", all_docs}, + {"heartbeat", 10000}, + {"since", Since}, + {"feed", Feed} + ], + QS = case proplists:get_value(<<"filter">>, PostProps) of + undefined -> + BaseQS; + FilterName -> + {Params} = proplists: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, Pid = couch_rep_httpc:spawn_link_worker_process(Source), Req = Source#http_db{ resource = "_changes", - qs = [{style, all_docs}, {heartbeat, 10000}, {since, Since}, - {feed, Feed}], + qs = QS, conn = Pid, options = [{stream_to, {self(), once}}, {response_format, binary}, {inactivity_timeout, 31000}], % miss 3 heartbeats, assume death @@ -94,20 +118,54 @@ init([_Parent, #http_db{}=Source, Since, PostProps] = Args) -> init([_Parent, Source, Since, PostProps] = InitArgs) -> process_flag(trap_exit, true), Server = self(), - ChangesPid = - case proplists:get_value(<<"continuous">>, PostProps, false) of - false -> - spawn_link(fun() -> send_local_changes_once(Server, Source, Since) end); - true -> - spawn_link(fun() -> - Self = self(), - {ok, _} = couch_db_update_notifier:start_link(fun(Msg) -> - local_update_notification(Self, Source#db.name, Msg) end), - send_local_changes_forever(Server, Source, Since) + ChangesArgs = #changes_args{ + style = all_docs, + since = Since, + filter = ?b2l(proplists:get_value(<<"filter">>, PostProps, <<>>)), + feed = case proplists:get_value(<<"continuous">>, PostProps, false) of + true -> + "continuous"; + false -> + "normal" + end + }, + ChangesPid = spawn_link(fun() -> + ChangesFeedFun = couch_changes:handle_changes( + ChangesArgs, + {json_req, filter_json_req(Source, PostProps)}, + Source + ), + ChangesFeedFun(fun({change, Change, _}, _) -> + gen_server:call(Server, {add_change, Change}, infinity); + (_, _) -> + ok end) - end, + end), {ok, #state{changes_loop=ChangesPid, init_args=InitArgs}}. +filter_json_req(Db, PostProps) -> + case proplists:get_value(<<"filter">>, PostProps) of + undefined -> + {[]}; + FilterName -> + {Query} = proplists: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)} + ]} + end. + handle_call({add_change, Row}, From, State) -> handle_add_change(Row, From, State); @@ -302,23 +360,7 @@ by_seq_loop(Server, Source, StartSeq) -> decode_row(<<",", Rest/binary>>) -> decode_row(Rest); decode_row(Row) -> - {Props} = ?JSON_DECODE(Row), - % [Seq, Id, {<<"changes">>,C}] - Seq = proplists:get_value(<<"seq">>, Props), - Id = proplists:get_value(<<"id">>, Props), - C = proplists:get_value(<<"changes">>, Props), - C2 = [{[{<<"rev">>,couch_doc:parse_rev(R)}]} || {[{<<"rev">>,R}]} <- C], - {[{<<"seq">>, Seq}, {<<"id">>,Id}, {<<"changes">>,C2}]}. - -flush_updated_messages() -> - receive updated -> flush_updated_messages() - after 0 -> ok - end. - -local_update_notification(Self, DbName, {updated, DbName}) -> - Self ! updated; -local_update_notification(_, _, _) -> - ok. + ?JSON_DECODE(Row). maybe_stream_next(#state{reqid=nil}) -> ok; @@ -327,35 +369,6 @@ maybe_stream_next(#state{complete=false, count=N} = S) when N < ?BUFFER_SIZE -> maybe_stream_next(_) -> ok. -send_local_changes_forever(Server, Db, Since) -> - #db{name = DbName, user_ctx = UserCtx} = Db, - {ok, NewSeq} = send_local_changes_once(Server, Db, Since), - couch_db:close(Db), - ok = wait_db_updated(), - {ok, NewDb} = couch_db:open(DbName, [{user_ctx, UserCtx}]), - send_local_changes_forever(Server, NewDb, NewSeq). - -send_local_changes_once(Server, Db, Since) -> - FilterFun = - fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) -> - {[{<<"rev">>, Rev}]} - end, - - ChangesFun = - fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, _) -> - Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos], - Results = [Result || Result <- Results0, Result /= null], - if Results /= [] -> - Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]}, - gen_server:call(Server, {add_change, Change}, infinity); - true -> - ok - end, - {ok, Seq} - end, - - couch_db:changes_since(Db, all_docs, Since, ChangesFun, Since). - start_http_request(RawUrl) -> Url = ibrowse_lib:parse_url(RawUrl), {ok, Pid} = ibrowse:spawn_link_worker_process(Url#url.host, Url#url.port), @@ -367,8 +380,3 @@ start_http_request(RawUrl) -> {ibrowse_req_id, Id} = ibrowse:send_req_direct(Pid, RawUrl, [], get, [], Opts, infinity), {Pid, Id}. - -wait_db_updated() -> - receive updated -> - flush_updated_messages() - end. diff --git a/src/couchdb/couch_rep_missing_revs.erl b/src/couchdb/couch_rep_missing_revs.erl index 5790dd71..7b4956e2 100644 --- a/src/couchdb/couch_rep_missing_revs.erl +++ b/src/couchdb/couch_rep_missing_revs.erl @@ -144,11 +144,15 @@ changes_loop(OurServer, SourceChangesServer, Target) -> changes_loop(OurServer, SourceChangesServer, Target). get_missing_revs(#http_db{}=Target, Changes) -> - Transform = fun({[{<<"seq">>,_}, {<<"id">>,Id}, {<<"changes">>,C}]}) -> - {Id, [couch_doc:rev_to_str(R) || {[{<<"rev">>, R}]} <- C]} end, + Transform = fun({Props}) -> + C = proplists:get_value(<<"changes">>, Props), + Id = proplists:get_value(<<"id">>, Props), + {Id, [R || {[{<<"rev">>, R}]} <- C]} + end, IdRevsList = [Transform(Change) || Change <- Changes], SeqDict = changes_dictionary(Changes), - {[{<<"seq">>, HighSeq}, _, _]} = lists:last(Changes), + {LastProps} = lists:last(Changes), + HighSeq = proplists:get_value(<<"seq">>, LastProps), Request = Target#http_db{ resource = "_missing_revs", method = post, @@ -165,11 +169,15 @@ get_missing_revs(#http_db{}=Target, Changes) -> end; get_missing_revs(Target, Changes) -> - Transform = fun({[{<<"seq">>,_}, {<<"id">>,Id}, {<<"changes">>,C}]}) -> - {Id, [R || {[{<<"rev">>, R}]} <- C]} end, + Transform = fun({Props}) -> + C = proplists:get_value(<<"changes">>, Props), + Id = proplists:get_value(<<"id">>, Props), + {Id, [couch_doc:parse_rev(R) || {[{<<"rev">>, R}]} <- C]} + end, IdRevsList = [Transform(Change) || Change <- Changes], SeqDict = changes_dictionary(Changes), - {[{<<"seq">>, HighSeq}, _, _]} = lists:last(Changes), + {LastProps} = lists:last(Changes), + HighSeq = proplists:get_value(<<"seq">>, LastProps), {ok, Results} = couch_db:get_missing_revs(Target, IdRevsList), {HighSeq, [{Id, dict:fetch(Id, SeqDict), Revs} || {Id, Revs, _} <- Results]}. |