diff options
Diffstat (limited to 'src/couchdb/couch_rep.erl')
| -rw-r--r-- | src/couchdb/couch_rep.erl | 748 |
1 files changed, 0 insertions, 748 deletions
diff --git a/src/couchdb/couch_rep.erl b/src/couchdb/couch_rep.erl deleted file mode 100644 index 65573e8c..00000000 --- a/src/couchdb/couch_rep.erl +++ /dev/null @@ -1,748 +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_rep). --behaviour(gen_server). --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, - code_change/3]). - --export([replicate/2, checkpoint/1]). - --include("couch_db.hrl"). - --record(state, { - changes_feed, - missing_revs, - reader, - writer, - - source, - target, - continuous, - create_target, - init_args, - checkpoint_scheduled = nil, - - start_seq, - history, - source_log, - target_log, - rep_starttime, - src_starttime, - tgt_starttime, - checkpoint_history = nil, - - listeners = [], - complete = false, - committed_seq = 0, - - stats = nil, - doc_ids = nil -}). - -%% convenience function to do a simple replication from the shell -replicate(Source, Target) when is_list(Source) -> - replicate(?l2b(Source), Target); -replicate(Source, Target) when is_binary(Source), is_list(Target) -> - replicate(Source, ?l2b(Target)); -replicate(Source, Target) when is_binary(Source), is_binary(Target) -> - replicate({[{<<"source">>, Source}, {<<"target">>, Target}]}, #user_ctx{}); - -%% function handling POST to _replicate -replicate({Props}=PostBody, UserCtx) -> - {BaseId, Extension} = make_replication_id(PostBody, UserCtx), - Replicator = {BaseId ++ Extension, - {gen_server, start_link, [?MODULE, [BaseId, PostBody, UserCtx], []]}, - temporary, - 1, - worker, - [?MODULE] - }, - - case couch_util:get_value(<<"cancel">>, Props, false) of - true -> - case supervisor:terminate_child(couch_rep_sup, BaseId ++ Extension) of - {error, not_found} -> - {error, not_found}; - ok -> - ok = supervisor:delete_child(couch_rep_sup, BaseId ++ Extension), - {ok, {cancelled, ?l2b(BaseId)}} - end; - false -> - Server = start_replication_server(Replicator), - - case couch_util:get_value(<<"continuous">>, Props, false) of - true -> - {ok, {continuous, ?l2b(BaseId)}}; - false -> - get_result(Server, PostBody, UserCtx) - end - end. - -checkpoint(Server) -> - gen_server:cast(Server, do_checkpoint). - -get_result(Server, PostBody, UserCtx) -> - try gen_server:call(Server, get_result, infinity) of - retry -> replicate(PostBody, UserCtx); - Else -> Else - catch - exit:{noproc, {gen_server, call, [Server, get_result , infinity]}} -> - %% oops, this replication just finished -- restart it. - replicate(PostBody, UserCtx); - exit:{normal, {gen_server, call, [Server, get_result , infinity]}} -> - %% we made the call during terminate - replicate(PostBody, UserCtx) - end. - -init(InitArgs) -> - try do_init(InitArgs) - catch throw:{db_not_found, DbUrl} -> {stop, {db_not_found, DbUrl}} end. - -do_init([RepId, {PostProps}, UserCtx] = InitArgs) -> - process_flag(trap_exit, true), - - SourceProps = couch_util:get_value(<<"source">>, PostProps), - TargetProps = couch_util:get_value(<<"target">>, PostProps), - - DocIds = couch_util:get_value(<<"doc_ids">>, PostProps, nil), - Continuous = couch_util:get_value(<<"continuous">>, PostProps, false), - CreateTarget = couch_util:get_value(<<"create_target">>, PostProps, false), - - ProxyParams = parse_proxy_params( - couch_util:get_value(<<"proxy">>, PostProps, [])), - Source = open_db(SourceProps, UserCtx, ProxyParams), - Target = open_db(TargetProps, UserCtx, ProxyParams, CreateTarget), - - SourceInfo = dbinfo(Source), - TargetInfo = dbinfo(Target), - - case DocIds of - List when is_list(List) -> - % Fast replication using only a list of doc IDs to replicate. - % Replication sessions, checkpoints and logs are not created - % since the update sequence number of the source DB is not used - % for determining which documents are copied into the target DB. - SourceLog = nil, - TargetLog = nil, - - StartSeq = nil, - History = nil, - - ChangesFeed = nil, - MissingRevs = nil, - - {ok, Reader} = - couch_rep_reader:start_link(self(), Source, DocIds, PostProps); - - _ -> - % Replication using the _changes API (DB sequence update numbers). - SourceLog = open_replication_log(Source, RepId), - TargetLog = open_replication_log(Target, RepId), - - {StartSeq, History} = compare_replication_logs(SourceLog, TargetLog), - - {ok, ChangesFeed} = - couch_rep_changes_feed:start_link(self(), Source, StartSeq, PostProps), - {ok, MissingRevs} = - couch_rep_missing_revs:start_link(self(), Target, ChangesFeed, PostProps), - {ok, Reader} = - couch_rep_reader:start_link(self(), Source, MissingRevs, PostProps) - end, - - {ok, Writer} = - couch_rep_writer:start_link(self(), Target, Reader, PostProps), - - Stats = ets:new(replication_stats, [set, private]), - ets:insert(Stats, {total_revs,0}), - ets:insert(Stats, {missing_revs, 0}), - ets:insert(Stats, {docs_read, 0}), - ets:insert(Stats, {docs_written, 0}), - ets:insert(Stats, {doc_write_failures, 0}), - - {ShortId, _} = lists:split(6, RepId), - couch_task_status:add_task("Replication", io_lib:format("~s: ~s -> ~s", - [ShortId, dbname(Source), dbname(Target)]), "Starting"), - - State = #state{ - changes_feed = ChangesFeed, - missing_revs = MissingRevs, - reader = Reader, - writer = Writer, - - source = Source, - target = Target, - continuous = Continuous, - create_target = CreateTarget, - init_args = InitArgs, - stats = Stats, - checkpoint_scheduled = nil, - - start_seq = StartSeq, - history = History, - source_log = SourceLog, - target_log = TargetLog, - rep_starttime = httpd_util:rfc1123_date(), - src_starttime = couch_util:get_value(instance_start_time, SourceInfo), - tgt_starttime = couch_util:get_value(instance_start_time, TargetInfo), - doc_ids = DocIds - }, - {ok, State}. - -handle_call(get_result, From, #state{complete=true, listeners=[]} = State) -> - {stop, normal, State#state{listeners=[From]}}; -handle_call(get_result, From, State) -> - Listeners = State#state.listeners, - {noreply, State#state{listeners=[From|Listeners]}}. - -handle_cast(do_checkpoint, State) -> - {noreply, do_checkpoint(State)}; - -handle_cast(_Msg, State) -> - {noreply, State}. - -handle_info({missing_revs_checkpoint, SourceSeq}, State) -> - couch_task_status:update("MR Processed source update #~p", [SourceSeq]), - {noreply, schedule_checkpoint(State#state{committed_seq = SourceSeq})}; - -handle_info({writer_checkpoint, SourceSeq}, #state{committed_seq=N} = State) - when SourceSeq > N -> - MissingRevs = State#state.missing_revs, - ok = gen_server:cast(MissingRevs, {update_committed_seq, SourceSeq}), - couch_task_status:update("W Processed source update #~p", [SourceSeq]), - {noreply, schedule_checkpoint(State#state{committed_seq = SourceSeq})}; -handle_info({writer_checkpoint, _}, State) -> - {noreply, State}; - -handle_info({update_stats, Key, N}, State) -> - ets:update_counter(State#state.stats, Key, N), - {noreply, State}; - -handle_info({'DOWN', _, _, _, _}, State) -> - ?LOG_INFO("replication terminating because local DB is shutting down", []), - timer:cancel(State#state.checkpoint_scheduled), - {stop, shutdown, State}; - -handle_info({'EXIT', Writer, normal}, #state{writer=Writer} = State) -> - case State#state.listeners of - [] -> - {noreply, State#state{complete = true}}; - _Else -> - {stop, normal, State} - end; - -handle_info({'EXIT', _, normal}, State) -> - {noreply, State}; -handle_info({'EXIT', _Pid, {Err, Reason}}, State) when Err == source_error; - Err == target_error -> - ?LOG_INFO("replication terminating due to ~p: ~p", [Err, Reason]), - timer:cancel(State#state.checkpoint_scheduled), - {stop, shutdown, State}; -handle_info({'EXIT', _Pid, Reason}, State) -> - {stop, Reason, State}. - -terminate(normal, #state{checkpoint_scheduled=nil} = State) -> - do_terminate(State); - -terminate(normal, State) -> - timer:cancel(State#state.checkpoint_scheduled), - do_terminate(do_checkpoint(State)); - -terminate(Reason, State) -> - #state{ - listeners = Listeners, - source = Source, - target = Target, - stats = Stats - } = State, - [gen_server:reply(L, {error, Reason}) || L <- Listeners], - ets:delete(Stats), - close_db(Target), - close_db(Source). - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -% internal funs - -start_replication_server(Replicator) -> - RepId = element(1, Replicator), - case supervisor:start_child(couch_rep_sup, Replicator) of - {ok, Pid} -> - ?LOG_INFO("starting new replication ~p at ~p", [RepId, Pid]), - Pid; - {error, already_present} -> - case supervisor:restart_child(couch_rep_sup, RepId) of - {ok, Pid} -> - ?LOG_INFO("starting replication ~p at ~p", [RepId, Pid]), - Pid; - {error, running} -> - %% this error occurs if multiple replicators are racing - %% each other to start and somebody else won. Just grab - %% the Pid by calling start_child again. - {error, {already_started, Pid}} = - supervisor:start_child(couch_rep_sup, Replicator), - ?LOG_DEBUG("replication ~p already running at ~p", [RepId, Pid]), - Pid; - {error, {db_not_found, DbUrl}} -> - throw({db_not_found, <<"could not open ", DbUrl/binary>>}) - end; - {error, {already_started, Pid}} -> - ?LOG_DEBUG("replication ~p already running at ~p", [RepId, Pid]), - Pid; - {error, {{db_not_found, DbUrl}, _}} -> - throw({db_not_found, <<"could not open ", DbUrl/binary>>}) - end. - -compare_replication_logs(SrcDoc, TgtDoc) -> - #doc{body={RepRecProps}} = SrcDoc, - #doc{body={RepRecPropsTgt}} = TgtDoc, - case couch_util:get_value(<<"session_id">>, RepRecProps) == - couch_util:get_value(<<"session_id">>, RepRecPropsTgt) of - true -> - % if the records have the same session id, - % then we have a valid replication history - OldSeqNum = couch_util:get_value(<<"source_last_seq">>, RepRecProps, 0), - OldHistory = couch_util:get_value(<<"history">>, RepRecProps, []), - {OldSeqNum, OldHistory}; - false -> - SourceHistory = couch_util:get_value(<<"history">>, RepRecProps, []), - TargetHistory = couch_util:get_value(<<"history">>, RepRecPropsTgt, []), - ?LOG_INFO("Replication records differ. " - "Scanning histories to find a common ancestor.", []), - ?LOG_DEBUG("Record on source:~p~nRecord on target:~p~n", - [RepRecProps, RepRecPropsTgt]), - compare_rep_history(SourceHistory, TargetHistory) - end. - -compare_rep_history(S, T) when S =:= [] orelse T =:= [] -> - ?LOG_INFO("no common ancestry -- performing full replication", []), - {0, []}; -compare_rep_history([{S}|SourceRest], [{T}|TargetRest]=Target) -> - SourceId = couch_util:get_value(<<"session_id">>, S), - case has_session_id(SourceId, Target) of - true -> - RecordSeqNum = couch_util:get_value(<<"recorded_seq">>, S, 0), - ?LOG_INFO("found a common replication record with source_seq ~p", - [RecordSeqNum]), - {RecordSeqNum, SourceRest}; - false -> - TargetId = couch_util:get_value(<<"session_id">>, T), - case has_session_id(TargetId, SourceRest) of - true -> - RecordSeqNum = couch_util:get_value(<<"recorded_seq">>, T, 0), - ?LOG_INFO("found a common replication record with source_seq ~p", - [RecordSeqNum]), - {RecordSeqNum, TargetRest}; - false -> - compare_rep_history(SourceRest, TargetRest) - end - end. - -close_db(#http_db{}) -> - ok; -close_db(Db) -> - couch_db:close(Db). - -dbname(#http_db{url = Url}) -> - strip_password(Url); -dbname(#db{name = Name}) -> - Name. - -strip_password(Url) -> - re:replace(Url, - "http(s)?://([^:]+):[^@]+@(.*)$", - "http\\1://\\2:*****@\\3", - [{return, list}]). - -dbinfo(#http_db{} = Db) -> - {DbProps} = couch_rep_httpc:request(Db), - [{list_to_existing_atom(?b2l(K)), V} || {K,V} <- DbProps]; -dbinfo(Db) -> - {ok, Info} = couch_db:get_db_info(Db), - Info. - -do_terminate(#state{doc_ids=DocIds} = State) when is_list(DocIds) -> - #state{ - listeners = Listeners, - rep_starttime = ReplicationStartTime, - stats = Stats - } = State, - - RepByDocsJson = {[ - {<<"start_time">>, ?l2b(ReplicationStartTime)}, - {<<"end_time">>, ?l2b(httpd_util:rfc1123_date())}, - {<<"docs_read">>, ets:lookup_element(Stats, docs_read, 2)}, - {<<"docs_written">>, ets:lookup_element(Stats, docs_written, 2)}, - {<<"doc_write_failures">>, - ets:lookup_element(Stats, doc_write_failures, 2)} - ]}, - - terminate_cleanup(State), - [gen_server:reply(L, {ok, RepByDocsJson}) || L <- lists:reverse(Listeners)]; - -do_terminate(State) -> - #state{ - checkpoint_history = CheckpointHistory, - committed_seq = NewSeq, - listeners = Listeners, - source = Source, - continuous = Continuous, - source_log = #doc{body={OldHistory}} - } = State, - - NewRepHistory = case CheckpointHistory of - nil -> - {[{<<"no_changes">>, true} | OldHistory]}; - _Else -> - CheckpointHistory - end, - - %% reply to original requester - OtherListeners = case Continuous of - true -> - []; % continuous replications have no listeners - _ -> - [Original|Rest] = lists:reverse(Listeners), - gen_server:reply(Original, {ok, NewRepHistory}), - Rest - end, - - %% maybe trigger another replication. If this replicator uses a local - %% source Db, changes to that Db since we started will not be included in - %% this pass. - case up_to_date(Source, NewSeq) of - true -> - [gen_server:reply(R, {ok, NewRepHistory}) || R <- OtherListeners]; - false -> - [gen_server:reply(R, retry) || R <- OtherListeners] - end, - terminate_cleanup(State). - -terminate_cleanup(#state{source=Source, target=Target, stats=Stats}) -> - couch_task_status:update("Finishing"), - close_db(Target), - close_db(Source), - ets:delete(Stats). - -has_session_id(_SessionId, []) -> - false; -has_session_id(SessionId, [{Props} | Rest]) -> - case couch_util:get_value(<<"session_id">>, Props, nil) of - SessionId -> - true; - _Else -> - has_session_id(SessionId, Rest) - end. - -maybe_append_options(Options, Props) -> - lists:foldl(fun(Option, Acc) -> - Acc ++ - case couch_util:get_value(Option, Props, false) of - true -> - "+" ++ ?b2l(Option); - false -> - "" - end - end, [], Options). - -make_replication_id({Props}, UserCtx) -> - %% funky algorithm to preserve backwards compatibility - {ok, HostName} = inet:gethostname(), - % Port = mochiweb_socket_server:get(couch_httpd, port), - Src = get_rep_endpoint(UserCtx, couch_util:get_value(<<"source">>, Props)), - Tgt = get_rep_endpoint(UserCtx, couch_util:get_value(<<"target">>, Props)), - Base = [HostName, Src, Tgt] ++ - case couch_util:get_value(<<"filter">>, Props) of - undefined -> - case couch_util:get_value(<<"doc_ids">>, Props) of - undefined -> - []; - DocIds -> - [DocIds] - end; - Filter -> - [Filter, couch_util:get_value(<<"query_params">>, Props, {[]})] - end, - Extension = maybe_append_options( - [<<"continuous">>, <<"create_target">>], Props), - {couch_util:to_hex(couch_util:md5(term_to_binary(Base))), Extension}. - -maybe_add_trailing_slash(Url) -> - re:replace(Url, "[^/]$", "&/", [{return, list}]). - -get_rep_endpoint(_UserCtx, {Props}) -> - Url = maybe_add_trailing_slash(couch_util:get_value(<<"url">>, Props)), - {BinHeaders} = couch_util:get_value(<<"headers">>, Props, {[]}), - {Auth} = couch_util:get_value(<<"auth">>, Props, {[]}), - case couch_util:get_value(<<"oauth">>, Auth) of - undefined -> - {remote, Url, [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders]}; - {OAuth} -> - {remote, Url, [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders], OAuth} - end; -get_rep_endpoint(_UserCtx, <<"http://",_/binary>>=Url) -> - {remote, maybe_add_trailing_slash(Url), []}; -get_rep_endpoint(_UserCtx, <<"https://",_/binary>>=Url) -> - {remote, maybe_add_trailing_slash(Url), []}; -get_rep_endpoint(UserCtx, <<DbName/binary>>) -> - {local, DbName, UserCtx}. - -open_replication_log(#http_db{}=Db, RepId) -> - DocId = ?LOCAL_DOC_PREFIX ++ RepId, - Req = Db#http_db{resource=couch_util:url_encode(DocId)}, - case couch_rep_httpc:request(Req) of - {[{<<"error">>, _}, {<<"reason">>, _}]} -> - ?LOG_DEBUG("didn't find a replication log for ~s", [Db#http_db.url]), - #doc{id=?l2b(DocId)}; - Doc -> - ?LOG_DEBUG("found a replication log for ~s", [Db#http_db.url]), - couch_doc:from_json_obj(Doc) - end; -open_replication_log(Db, RepId) -> - DocId = ?l2b(?LOCAL_DOC_PREFIX ++ RepId), - case couch_db:open_doc(Db, DocId, []) of - {ok, Doc} -> - ?LOG_DEBUG("found a replication log for ~s", [Db#db.name]), - Doc; - _ -> - ?LOG_DEBUG("didn't find a replication log for ~s", [Db#db.name]), - #doc{id=DocId} - end. - -open_db(Props, UserCtx, ProxyParams) -> - open_db(Props, UserCtx, ProxyParams, false). - -open_db({Props}, _UserCtx, ProxyParams, CreateTarget) -> - Url = maybe_add_trailing_slash(couch_util:get_value(<<"url">>, Props)), - {AuthProps} = couch_util:get_value(<<"auth">>, Props, {[]}), - {BinHeaders} = couch_util:get_value(<<"headers">>, Props, {[]}), - Headers = [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders], - DefaultHeaders = (#http_db{})#http_db.headers, - Db1 = #http_db{ - url = Url, - auth = AuthProps, - headers = lists:ukeymerge(1, Headers, DefaultHeaders) - }, - Db = Db1#http_db{options = Db1#http_db.options ++ ProxyParams}, - couch_rep_httpc:db_exists(Db, CreateTarget); -open_db(<<"http://",_/binary>>=Url, _, ProxyParams, CreateTarget) -> - open_db({[{<<"url">>,Url}]}, [], ProxyParams, CreateTarget); -open_db(<<"https://",_/binary>>=Url, _, ProxyParams, CreateTarget) -> - open_db({[{<<"url">>,Url}]}, [], ProxyParams, CreateTarget); -open_db(<<DbName/binary>>, UserCtx, _ProxyParams, CreateTarget) -> - case CreateTarget of - true -> - ok = couch_httpd:verify_is_server_admin(UserCtx), - couch_server:create(DbName, [{user_ctx, UserCtx}]); - false -> ok - end, - - case couch_db:open(DbName, [{user_ctx, UserCtx}]) of - {ok, Db} -> - couch_db:monitor(Db), - Db; - {not_found, no_db_file} -> throw({db_not_found, DbName}) - end. - -schedule_checkpoint(#state{checkpoint_scheduled = nil} = State) -> - Server = self(), - case timer:apply_after(5000, couch_rep, checkpoint, [Server]) of - {ok, TRef} -> - State#state{checkpoint_scheduled = TRef}; - Error -> - ?LOG_ERROR("tried to schedule a checkpoint but got ~p", [Error]), - State - end; -schedule_checkpoint(State) -> - State. - -do_checkpoint(State) -> - #state{ - source = Source, - target = Target, - committed_seq = NewSeqNum, - start_seq = StartSeqNum, - history = OldHistory, - source_log = SourceLog, - target_log = TargetLog, - rep_starttime = ReplicationStartTime, - src_starttime = SrcInstanceStartTime, - tgt_starttime = TgtInstanceStartTime, - stats = Stats - } = State, - case commit_to_both(Source, Target, NewSeqNum) of - {SrcInstanceStartTime, TgtInstanceStartTime} -> - ?LOG_INFO("recording a checkpoint for ~s -> ~s at source update_seq ~p", - [dbname(Source), dbname(Target), NewSeqNum]), - SessionId = couch_uuids:random(), - NewHistoryEntry = {[ - {<<"session_id">>, SessionId}, - {<<"start_time">>, list_to_binary(ReplicationStartTime)}, - {<<"end_time">>, list_to_binary(httpd_util:rfc1123_date())}, - {<<"start_last_seq">>, StartSeqNum}, - {<<"end_last_seq">>, NewSeqNum}, - {<<"recorded_seq">>, NewSeqNum}, - {<<"missing_checked">>, ets:lookup_element(Stats, total_revs, 2)}, - {<<"missing_found">>, ets:lookup_element(Stats, missing_revs, 2)}, - {<<"docs_read">>, ets:lookup_element(Stats, docs_read, 2)}, - {<<"docs_written">>, ets:lookup_element(Stats, docs_written, 2)}, - {<<"doc_write_failures">>, - ets:lookup_element(Stats, doc_write_failures, 2)} - ]}, - % limit history to 50 entries - NewRepHistory = {[ - {<<"session_id">>, SessionId}, - {<<"source_last_seq">>, NewSeqNum}, - {<<"history">>, lists:sublist([NewHistoryEntry | OldHistory], 50)} - ]}, - - try - {SrcRevPos,SrcRevId} = - update_local_doc(Source, SourceLog#doc{body=NewRepHistory}), - {TgtRevPos,TgtRevId} = - update_local_doc(Target, TargetLog#doc{body=NewRepHistory}), - State#state{ - checkpoint_scheduled = nil, - checkpoint_history = NewRepHistory, - source_log = SourceLog#doc{revs={SrcRevPos, [SrcRevId]}}, - target_log = TargetLog#doc{revs={TgtRevPos, [TgtRevId]}} - } - catch throw:conflict -> - ?LOG_ERROR("checkpoint failure: conflict (are you replicating to " - "yourself?)", []), - State - end; - _Else -> - ?LOG_INFO("rebooting ~s -> ~s from last known replication checkpoint", - [dbname(Source), dbname(Target)]), - #state{ - changes_feed = CF, - missing_revs = MR, - reader = Reader, - writer = Writer - } = State, - Pids = [Writer, Reader, MR, CF], - [unlink(Pid) || Pid <- Pids], - [exit(Pid, shutdown) || Pid <- Pids], - close_db(Target), - close_db(Source), - {ok, NewState} = init(State#state.init_args), - NewState#state{listeners=State#state.listeners} - end. - -commit_to_both(Source, Target, RequiredSeq) -> - % commit the src async - ParentPid = self(), - SrcCommitPid = spawn_link(fun() -> - ParentPid ! {self(), ensure_full_commit(Source, RequiredSeq)} end), - - % commit tgt sync - TargetStartTime = ensure_full_commit(Target), - - SourceStartTime = - receive - {SrcCommitPid, Timestamp} -> - Timestamp; - {'EXIT', SrcCommitPid, {http_request_failed, _}} -> - exit(replication_link_failure) - end, - {SourceStartTime, TargetStartTime}. - -ensure_full_commit(#http_db{headers = Headers} = Target) -> - Req = Target#http_db{ - resource = "_ensure_full_commit", - method = post, - headers = couch_util:proplist_apply_field({"Content-Type", "application/json"}, Headers) - }, - {ResultProps} = couch_rep_httpc:request(Req), - true = couch_util:get_value(<<"ok">>, ResultProps), - couch_util:get_value(<<"instance_start_time">>, ResultProps); -ensure_full_commit(Target) -> - {ok, NewDb} = couch_db:open_int(Target#db.name, []), - UpdateSeq = couch_db:get_update_seq(Target), - CommitSeq = couch_db:get_committed_update_seq(NewDb), - InstanceStartTime = NewDb#db.instance_start_time, - couch_db:close(NewDb), - if UpdateSeq > CommitSeq -> - ?LOG_DEBUG("target needs a full commit: update ~p commit ~p", - [UpdateSeq, CommitSeq]), - {ok, DbStartTime} = couch_db:ensure_full_commit(Target), - DbStartTime; - true -> - ?LOG_DEBUG("target doesn't need a full commit", []), - InstanceStartTime - end. - -ensure_full_commit(#http_db{headers = Headers} = Source, RequiredSeq) -> - Req = Source#http_db{ - resource = "_ensure_full_commit", - method = post, - qs = [{seq, RequiredSeq}], - headers = couch_util:proplist_apply_field({"Content-Type", "application/json"}, Headers) - }, - {ResultProps} = couch_rep_httpc:request(Req), - case couch_util:get_value(<<"ok">>, ResultProps) of - true -> - couch_util:get_value(<<"instance_start_time">>, ResultProps); - undefined -> nil end; -ensure_full_commit(Source, RequiredSeq) -> - {ok, NewDb} = couch_db:open_int(Source#db.name, []), - CommitSeq = couch_db:get_committed_update_seq(NewDb), - InstanceStartTime = NewDb#db.instance_start_time, - couch_db:close(NewDb), - if RequiredSeq > CommitSeq -> - ?LOG_DEBUG("source needs a full commit: required ~p committed ~p", - [RequiredSeq, CommitSeq]), - {ok, DbStartTime} = couch_db:ensure_full_commit(Source), - DbStartTime; - true -> - ?LOG_DEBUG("source doesn't need a full commit", []), - InstanceStartTime - end. - -update_local_doc(#http_db{} = Db, #doc{id=DocId} = Doc) -> - Req = Db#http_db{ - resource = couch_util:url_encode(DocId), - method = put, - body = couch_doc:to_json_obj(Doc, [attachments]), - headers = [{"x-couch-full-commit", "false"} | Db#http_db.headers] - }, - {ResponseMembers} = couch_rep_httpc:request(Req), - Rev = couch_util:get_value(<<"rev">>, ResponseMembers), - couch_doc:parse_rev(Rev); -update_local_doc(Db, Doc) -> - {ok, Result} = couch_db:update_doc(Db, Doc, [delay_commit]), - Result. - -up_to_date(#http_db{}, _Seq) -> - true; -up_to_date(Source, Seq) -> - {ok, NewDb} = couch_db:open_int(Source#db.name, []), - T = NewDb#db.update_seq == Seq, - couch_db:close(NewDb), - T. - -parse_proxy_params(ProxyUrl) when is_binary(ProxyUrl) -> - parse_proxy_params(?b2l(ProxyUrl)); -parse_proxy_params([]) -> - []; -parse_proxy_params(ProxyUrl) -> - {url, _, Base, Port, User, Passwd, _Path, _Proto} = - ibrowse_lib:parse_url(ProxyUrl), - [{proxy_host, Base}, {proxy_port, Port}] ++ - case is_list(User) andalso is_list(Passwd) of - false -> - []; - true -> - [{proxy_user, User}, {proxy_password, Passwd}] - end. |