You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by bb...@apache.org on 2014/10/31 20:53:14 UTC
[5/5] couch-replicator commit: updated refs/heads/master to c9184cf
couch_replicator: add replication using changes in a view
Instead of a database, the replicator can now filter the documents using
a view index. All documents having a key emitted in the view can be
replicated.
View parameters can be used. Which means that you can replicate results
corresponding to a key in a view or a range.
Conflicts:
src/couch_replicator.erl
src/couch_replicator_api_wrap.erl
Project: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/commit/3e3e2cb6
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/tree/3e3e2cb6
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/diff/3e3e2cb6
Branch: refs/heads/master
Commit: 3e3e2cb640a9babf90e6e2ba8af95c2c1cde6d74
Parents: 59a2f2b
Author: benoitc <be...@apache.org>
Authored: Sun Feb 9 00:43:23 2014 +0100
Committer: Benjamin Bastian <be...@gmail.com>
Committed: Thu Oct 30 13:29:56 2014 -0700
----------------------------------------------------------------------
src/couch_replicator.erl | 38 +++++++++++++++++++++++++++++++---
src/couch_replicator.hrl | 2 ++
src/couch_replicator_api_wrap.erl | 21 +++++++++++++++++++
src/couch_replicator_utils.erl | 27 ++++++++++++++++++++++--
4 files changed, 83 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/3e3e2cb6/src/couch_replicator.erl
----------------------------------------------------------------------
diff --git a/src/couch_replicator.erl b/src/couch_replicator.erl
index ee45d0b..f7e9d27 100644
--- a/src/couch_replicator.erl
+++ b/src/couch_replicator.erl
@@ -73,7 +73,9 @@
target_monitor = nil,
source_seq = nil,
use_checkpoints = true,
- checkpoint_interval = 5000
+ checkpoint_interval = 5000,
+ type = db,
+ view = nil
}).
@@ -558,7 +560,8 @@ cancel_timer(#rep_state{timer = Timer} = State) ->
init_state(Rep) ->
#rep{
source = Src, target = Tgt,
- options = Options, user_ctx = UserCtx
+ options = Options, user_ctx = UserCtx,
+ type = Type, view = View
} = Rep,
{ok, Source} = couch_replicator_api_wrap:db_open(Src, [{user_ctx, UserCtx}]),
{ok, Target} = couch_replicator_api_wrap:db_open(Tgt, [{user_ctx, UserCtx}],
@@ -572,6 +575,17 @@ init_state(Rep) ->
{StartSeq0, History} = compare_replication_logs(SourceLog, TargetLog),
StartSeq1 = get_value(since_seq, Options, StartSeq0),
StartSeq = {0, StartSeq1},
+
+ SourceSeq = case Type of
+ db -> get_value(<<"update_seq">>, SourceInfo, ?LOWEST_SEQ);
+ view ->
+ {DDoc, VName} = View,
+ {ok, VInfo} = couch_replicator_api_wrap:get_view_info(Source, DDoc,
+ VName),
+ get_value(<<"update_seq">>, VInfo, ?LOWEST_SEQ)
+ end,
+
+
#doc{body={CheckpointHistory}} = SourceLog,
State = #rep_state{
rep_details = Rep,
@@ -596,8 +610,12 @@ init_state(Rep) ->
start_db_compaction_notifier(Target, self()),
source_monitor = db_monitor(Source),
target_monitor = db_monitor(Target),
+ source_seq = SourceSeq,
use_checkpoints = get_value(use_checkpoints, Options, true),
- checkpoint_interval = get_value(checkpoint_interval, Options, 5000)
+ checkpoint_interval = get_value(checkpoint_interval, Options,
+ 5000),
+ type = Type,
+ view = View
},
State#rep_state{timer = start_timer(State)}.
@@ -883,6 +901,20 @@ db_monitor(#db{} = Db) ->
db_monitor(_HttpDb) ->
nil.
+source_cur_seq(#rep_state{source = #httpdb{} = Db, source_seq = Seq,
+ type = view, view = {DDoc, VName}}) ->
+ case (catch couch_replicator_api_wrap:get_view_info(
+ Db#httpdb{retries = 3}, DDoc, VName)) of
+ {ok, Info} ->
+ get_value(<<"update_seq">>, Info, Seq);
+ _ ->
+ Seq
+ end;
+
+source_cur_seq(#rep_state{source = Db, source_seq = Seq,
+ type = view, view = {DDoc, VName}}) ->
+ {ok, Info} = couch_replicator_api_wrap:get_view_info(Db, DDoc, VName),
+ get_value(<<"update_seq">>, Info, Seq);
get_pending_count(St) ->
Rep = St#rep_state.rep_details,
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/3e3e2cb6/src/couch_replicator.hrl
----------------------------------------------------------------------
diff --git a/src/couch_replicator.hrl b/src/couch_replicator.hrl
index 893d9e4..dbb1793 100644
--- a/src/couch_replicator.hrl
+++ b/src/couch_replicator.hrl
@@ -18,5 +18,7 @@
target,
options,
user_ctx,
+ type = db,
+ view = nil,
doc_id
}).
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/3e3e2cb6/src/couch_replicator_api_wrap.erl
----------------------------------------------------------------------
diff --git a/src/couch_replicator_api_wrap.erl b/src/couch_replicator_api_wrap.erl
index 89c7eb7..97e9b25 100644
--- a/src/couch_replicator_api_wrap.erl
+++ b/src/couch_replicator_api_wrap.erl
@@ -19,6 +19,7 @@
% Many options and apis aren't yet supported here, they are added as needed.
-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
-include("couch_replicator_api_wrap.hrl").
-export([
@@ -27,6 +28,7 @@
db_close/1,
get_db_info/1,
get_pending_count/2,
+ get_view_info/3,
update_doc/3,
update_doc/4,
update_docs/3,
@@ -164,6 +166,15 @@ get_pending_count(#db{name=DbName}=Db, Seq) when is_number(Seq) ->
couch_db:close(CountDb),
{ok, Pending}.
+get_view_info(#httpdb{} = Db, DDocId, ViewName) ->
+ Path = iolist_to_binary([DDocId, "/_view/", ViewName, "/_info"]),
+ send_req(Db, [{path, Path}],
+ fun(200, _, {Props}) ->
+ {ok, Props}
+ end);
+get_view_info(#db{name = DbName}, DDocId, ViewName) ->
+ couch_mrview:get_view_info(DbName, DDocId, ViewName).
+
ensure_full_commit(#httpdb{} = Db) ->
send_req(
@@ -517,6 +528,10 @@ maybe_add_changes_filter_q_args(BaseQS, Options) ->
undefined ->
BaseQS;
FilterName ->
+ %% get list of view attributes
+ ViewFields0 = [atom_to_list(F) || F <- record_info(fields, mrargs)],
+ ViewFields = ["key" | ViewFields0],
+
{Params} = get_value(query_params, Options, {[]}),
[{"filter", ?b2l(FilterName)} | lists:foldl(
fun({K, V}, QSAcc) ->
@@ -524,6 +539,12 @@ maybe_add_changes_filter_q_args(BaseQS, Options) ->
case lists:keymember(Ks, 1, QSAcc) of
true ->
QSAcc;
+ false when FilterName =:= <<"_view">> ->
+ V1 = case lists:member(Ks, ViewFields) of
+ true -> ?JSON_ENCODE(V);
+ false -> couch_util:to_list(V)
+ end,
+ [{Ks, V1} | QSAcc];
false ->
[{Ks, couch_util:to_list(V)} | QSAcc]
end
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/3e3e2cb6/src/couch_replicator_utils.erl
----------------------------------------------------------------------
diff --git a/src/couch_replicator_utils.erl b/src/couch_replicator_utils.erl
index 9d716c9..30afb39 100644
--- a/src/couch_replicator_utils.erl
+++ b/src/couch_replicator_utils.erl
@@ -40,13 +40,34 @@ parse_rep_doc({Props}, UserCtx) ->
true ->
{ok, #rep{options = Options, user_ctx = UserCtx}};
false ->
- Source = parse_rep_db(get_value(<<"source">>, Props), ProxyParams, Options),
- Target = parse_rep_db(get_value(<<"target">>, Props), ProxyParams, Options),
+ Source = parse_rep_db(get_value(<<"source">>, Props),
+ ProxyParams, Options),
+ Target = parse_rep_db(get_value(<<"target">>, Props),
+ ProxyParams, Options),
+
+
+ {RepType, View} = case get_value(<<"filter">>, Props) of
+ <<"_view">> ->
+ {QP} = get_value(query_params, Options, {[]}),
+ ViewParam = get_value(<<"view">>, QP),
+ View1 = case re:split(ViewParam, <<"/">>) of
+ [DName, ViewName] ->
+ {<< "_design/", DName/binary >>, ViewName};
+ _ ->
+ throw({bad_request, "Invalid `view` parameter."})
+ end,
+ {view, View1};
+ _ ->
+ {db, nil}
+ end,
+
Rep = #rep{
source = Source,
target = Target,
options = Options,
user_ctx = UserCtx,
+ type = RepType,
+ view = View,
doc_id = get_value(<<"_id">>, Props, null)
},
{ok, Rep#rep{id = replication_id(Rep)}}
@@ -103,6 +124,8 @@ maybe_append_filters(Base,
DocIds ->
[DocIds]
end;
+ <<"_", _/binary>> = Filter ->
+ [Filter, get_value(query_params, Options, {[]})];
Filter ->
[filter_code(Filter, Source, UserCtx),
get_value(query_params, Options, {[]})]