diff options
Diffstat (limited to 'src/couchdb/couch_rep.erl')
-rw-r--r-- | src/couchdb/couch_rep.erl | 86 |
1 files changed, 53 insertions, 33 deletions
diff --git a/src/couchdb/couch_rep.erl b/src/couchdb/couch_rep.erl index 71aa0693..b63a2610 100644 --- a/src/couchdb/couch_rep.erl +++ b/src/couchdb/couch_rep.erl @@ -84,7 +84,8 @@ replicate(Source, Target) -> -record(http_db, { uri, - headers + headers, + oauth }). @@ -496,8 +497,8 @@ make_att_stub_receiver(Url, Headers, Name, Type, Length, Retries, Pause) -> end. -open_db({remote, Url, Headers})-> - {ok, #http_db{uri=?b2l(Url), headers=Headers}, Url}; +open_db({remote, Url, Headers, Auth})-> + {ok, #http_db{uri=?b2l(Url), headers=Headers, oauth=Auth}, Url}; open_db({local, DbName, UserCtx})-> case couch_db:open(DbName, [{user_ctx, UserCtx}]) of {ok, Db} -> {ok, Db, DbName}; @@ -600,19 +601,38 @@ do_checkpoint(Source, Target, Context, NewSeqNum, Stats) -> end. -do_http_request(Url, Action, Headers) -> - do_http_request(Url, Action, Headers, []). - -do_http_request(Url, Action, Headers, JsonBody) -> - do_http_request(Url, Action, Headers, JsonBody, 10, 1000). +do_http_request(Url, Action, Headers, Auth) -> + do_http_request(Url, Action, Headers, Auth, []). + +do_http_request(Url, Action, Headers, Auth, JsonBody) -> + Headers0 = case Auth of + {Props} -> + % Add OAuth header + {OAuth} = proplists:get_value(<<"oauth">>, Props), + ConsumerKey = ?b2l(proplists:get_value(<<"consumer_key">>, OAuth)), + Token = ?b2l(proplists:get_value(<<"token">>, OAuth)), + TokenSecret = ?b2l(proplists:get_value(<<"token_secret">>, OAuth)), + ConsumerSecret = ?b2l(proplists:get_value(<<"consumer_secret">>, OAuth)), + Consumer = {ConsumerKey, ConsumerSecret, hmac_sha1}, + Method = case Action of + get -> "GET"; + post -> "POST"; + put -> "PUT" + end, + Params = oauth:signed_params(Method, Url, [], Consumer, Token, TokenSecret), + [{"Authorization", "OAuth " ++ oauth_uri:params_to_header_string(Params)} | Headers]; + _Else -> + Headers + end, + do_http_request0(Url, Action, Headers0, JsonBody, 10, 1000). -do_http_request(Url, Action, Headers, Body, Retries, Pause) when is_binary(Url) -> - do_http_request(?b2l(Url), Action, Headers, Body, Retries, Pause); -do_http_request(Url, Action, _Headers, _JsonBody, 0, _Pause) -> +do_http_request0(Url, Action, Headers, Body, Retries, Pause) when is_binary(Url) -> + do_http_request0(?b2l(Url), Action, Headers, Body, Retries, Pause); +do_http_request0(Url, Action, _Headers, _JsonBody, 0, _Pause) -> ?LOG_ERROR("couch_rep HTTP ~p request failed after 10 retries: ~s", [Action, Url]), exit({http_request_failed, ?l2b(["failed to replicate ", Url])}); -do_http_request(Url, Action, Headers, JsonBody, Retries, Pause) -> +do_http_request0(Url, Action, Headers, JsonBody, Retries, Pause) -> ?LOG_DEBUG("couch_rep HTTP ~p request: ~s", [Action, Url]), Body = case JsonBody of @@ -638,7 +658,7 @@ do_http_request(Url, Action, Headers, JsonBody, Retries, Pause) -> ResponseCode >= 300, ResponseCode < 400 -> RedirectUrl = mochiweb_headers:get_value("Location", mochiweb_headers:make(ResponseHeaders)), - do_http_request(RedirectUrl, Action, Headers, JsonBody, Retries-1, + do_http_request0(RedirectUrl, Action, Headers, JsonBody, Retries-1, Pause); ResponseCode >= 400, ResponseCode < 500 -> ?JSON_DECODE(ResponseBody); @@ -646,18 +666,18 @@ do_http_request(Url, Action, Headers, JsonBody, Retries, Pause) -> ?LOG_INFO("retrying couch_rep HTTP ~p request in ~p seconds " ++ "due to 500 error: ~s", [Action, Pause/1000, Url]), timer:sleep(Pause), - do_http_request(Url, Action, Headers, JsonBody, Retries - 1, 2*Pause) + do_http_request0(Url, Action, Headers, JsonBody, Retries - 1, 2*Pause) end; {error, Reason} -> ?LOG_INFO("retrying couch_rep HTTP ~p request in ~p seconds due to " ++ "{error, ~p}: ~s", [Action, Pause/1000, Reason, Url]), timer:sleep(Pause), - do_http_request(Url, Action, Headers, JsonBody, Retries - 1, 2*Pause) + do_http_request0(Url, Action, Headers, JsonBody, Retries - 1, 2*Pause) end. -ensure_full_commit(#http_db{uri=DbUrl, headers=Headers}) -> +ensure_full_commit(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}) -> {ResultProps} = do_http_request(DbUrl ++ "_ensure_full_commit", post, - Headers, true), + Headers, OAuth, true), true = proplists:get_value(<<"ok">>, ResultProps), {ok, proplists:get_value(<<"instance_start_time">>, ResultProps)}; ensure_full_commit(Db) -> @@ -687,16 +707,16 @@ enum_docs_since(Pid, DbSource, DbTarget, {StartSeq, RevsCount}) -> -get_db_info(#http_db{uri=DbUrl, headers=Headers}) -> - {DbProps} = do_http_request(DbUrl, get, Headers), +get_db_info(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}) -> + {DbProps} = do_http_request(DbUrl, get, Headers, OAuth), {ok, [{list_to_atom(?b2l(K)), V} || {K,V} <- DbProps]}; get_db_info(Db) -> couch_db:get_db_info(Db). -get_doc_info_list(#http_db{uri=DbUrl, headers=Headers}, StartSeq) -> +get_doc_info_list(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}, StartSeq) -> Url = DbUrl ++ "_all_docs_by_seq?limit=100&startkey=" ++ integer_to_list(StartSeq), - {Results} = do_http_request(Url, get, Headers), + {Results} = do_http_request(Url, get, Headers, OAuth), lists:map(fun({RowInfoList}) -> {RowValueProps} = proplists:get_value(<<"value">>, RowInfoList), Seq = proplists:get_value(<<"key">>, RowInfoList), @@ -719,9 +739,9 @@ get_doc_info_list(DbSource, StartSeq) -> end, {0, []}), lists:reverse(DocInfoList). -get_missing_revs(#http_db{uri=DbUrl, headers=Headers}, DocIdRevsList) -> +get_missing_revs(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}, DocIdRevsList) -> DocIdRevsList2 = [{Id, couch_doc:rev_to_strs(Revs)} || {Id, Revs} <- DocIdRevsList], - {ResponseMembers} = do_http_request(DbUrl ++ "_missing_revs", post, Headers, + {ResponseMembers} = do_http_request(DbUrl ++ "_missing_revs", post, Headers, OAuth, {DocIdRevsList2}), {DocMissingRevsList} = proplists:get_value(<<"missing_revs">>, ResponseMembers), DocMissingRevsList2 = [{Id, couch_doc:parse_revs(MissingRevStrs)} || {Id, MissingRevStrs} <- DocMissingRevsList], @@ -730,9 +750,9 @@ get_missing_revs(Db, DocId) -> couch_db:get_missing_revs(Db, DocId). -open_doc(#http_db{uri=DbUrl, headers=Headers}, DocId, Options) -> +open_doc(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}, DocId, Options) -> [] = Options, - case do_http_request(DbUrl ++ couch_util:url_encode(DocId), get, Headers) of + case do_http_request(DbUrl ++ couch_util:url_encode(DocId), get, Headers, OAuth) of {[{<<"error">>, ErrId}, {<<"reason">>, Reason}]} -> {couch_util:to_existing_atom(ErrId), Reason}; Doc -> @@ -741,7 +761,7 @@ open_doc(#http_db{uri=DbUrl, headers=Headers}, DocId, Options) -> open_doc(Db, DocId, Options) -> couch_db:open_doc(Db, DocId, Options). -open_doc_revs(#http_db{uri=DbUrl, headers=Headers} = DbS, DocId, Revs0, +open_doc_revs(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth} = DbS, DocId, Revs0, [latest]) -> Revs = couch_doc:rev_to_strs(Revs0), BaseUrl = DbUrl ++ couch_util:url_encode(DocId) ++ "?revs=true&latest=true", @@ -752,18 +772,18 @@ open_doc_revs(#http_db{uri=DbUrl, headers=Headers} = DbS, DocId, Revs0, JsonResults = case length(Revs) > MaxN of false -> Url = ?l2b(BaseUrl ++ "&open_revs=" ++ ?JSON_ENCODE(Revs)), - do_http_request(Url, get, Headers); + do_http_request(Url, get, Headers, OAuth); true -> {_, Rest, Acc} = lists:foldl( fun(Rev, {Count, RevsAcc, AccResults}) when Count =:= MaxN -> QSRevs = ?JSON_ENCODE(lists:reverse(RevsAcc)), Url = ?l2b(BaseUrl ++ "&open_revs=" ++ QSRevs), - {1, [Rev], AccResults++do_http_request(Url, get, Headers)}; + {1, [Rev], AccResults++do_http_request(Url, get, Headers, OAuth)}; (Rev, {Count, RevsAcc, AccResults}) -> {Count+1, [Rev|RevsAcc], AccResults} end, {0, [], []}, Revs), Acc ++ do_http_request(?l2b(BaseUrl ++ "&open_revs=" ++ - ?JSON_ENCODE(lists:reverse(Rest))), get, Headers) + ?JSON_ENCODE(lists:reverse(Rest))), get, Headers, OAuth) end, Results = @@ -825,10 +845,10 @@ binary_memory(Pid) -> lists:foldl(fun({_Id, Size, _NRefs}, Acc) -> Size+Acc end, 0, element(2,process_info(Pid, binary))). -update_doc(#http_db{uri=DbUrl, headers=Headers}, #doc{id=DocId}=Doc, Options) -> +update_doc(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}, #doc{id=DocId}=Doc, Options) -> [] = Options, Url = DbUrl ++ couch_util:url_encode(DocId), - {ResponseMembers} = do_http_request(Url, put, Headers, + {ResponseMembers} = do_http_request(Url, put, Headers, OAuth, couch_doc:to_json_obj(Doc, [attachments])), Rev = proplists:get_value(<<"rev">>, ResponseMembers), {ok, couch_doc:parse_rev(Rev)}; @@ -837,10 +857,10 @@ update_doc(Db, Doc, Options) -> update_docs(_, [], _, _) -> {ok, []}; -update_docs(#http_db{uri=DbUrl, headers=Headers}, Docs, [], replicated_changes) -> +update_docs(#http_db{uri=DbUrl, headers=Headers, oauth=OAuth}, Docs, [], replicated_changes) -> JsonDocs = [couch_doc:to_json_obj(Doc, [revs,attachments]) || Doc <- Docs], ErrorsJson = - do_http_request(DbUrl ++ "_bulk_docs", post, Headers, + do_http_request(DbUrl ++ "_bulk_docs", post, Headers, OAuth, {[{new_edits, false}, {docs, JsonDocs}]}), ErrorsList = lists:map( |