couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fdman...@apache.org
Subject svn commit: r1173419 - in /couchdb/trunk: src/couchdb/couch_changes.erl src/couchdb/couch_db.hrl test/etap/073-changes.t test/etap/Makefile.am
Date Tue, 20 Sep 2011 22:55:30 GMT
Author: fdmanana
Date: Tue Sep 20 22:55:29 2011
New Revision: 1173419

URL: http://svn.apache.org/viewvc?rev=1173419&view=rev
Log:
Efficient implementation of builtin filters

Currently, the builtin changes filters "_doc_ids" and "_design"
are not very efficient because they fold the entire seq btree
and then filter the values by document ID.
This implementation avoids that by doing direct lookups against
the id btree, and then, for continuous changes requests, it
just listens for database update events and does partial seq
btree folds.

COUCHDB-1288

Added:
    couchdb/trunk/test/etap/073-changes.t   (with props)
Modified:
    couchdb/trunk/src/couchdb/couch_changes.erl
    couchdb/trunk/src/couchdb/couch_db.hrl
    couchdb/trunk/test/etap/Makefile.am

Modified: couchdb/trunk/src/couchdb/couch_changes.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_changes.erl?rev=1173419&r1=1173418&r2=1173419&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_changes.erl (original)
+++ couchdb/trunk/src/couchdb/couch_changes.erl Tue Sep 20 22:55:29 2011
@@ -15,6 +15,10 @@
 
 -export([handle_changes/3]).
 
+% For the builtin filter _docs_ids, this is the maximum number
+% of documents for which we trigger the optimized code path.
+-define(MAX_DOC_IDS, 100).
+
 -record(changes_acc, {
     db,
     seq,
@@ -29,15 +33,21 @@
 }).
 
 %% @type Req -> #httpd{} | {json_req, JsonObj()}
-handle_changes(#changes_args{style=Style}=Args1, Req, Db) ->
-    #changes_args{feed = Feed} = Args = Args1#changes_args{
-        filter = make_filter_fun(Args1#changes_args.filter, Style, Req, Db)
-    },
-    StartSeq = case Args#changes_args.dir of
+handle_changes(Args1, Req, Db) ->
+    #changes_args{
+        style = Style,
+        filter = FilterName,
+        feed = Feed,
+        dir = Dir,
+        since = Since
+    } = Args1,
+    {FilterFun, FilterArgs} = make_filter_fun(FilterName, Style, Req, Db),
+    Args = Args1#changes_args{filter_fun = FilterFun, filter_args = FilterArgs},
+    StartSeq = case Dir of
     rev ->
         couch_db:get_update_seq(Db);
     fwd ->
-        Args#changes_args.since
+        Since
     end,
     if Feed == "continuous" orelse Feed == "longpoll" ->
         fun(CallbackAcc) ->
@@ -61,8 +71,8 @@ handle_changes(#changes_args{style=Style
                     StartSeq,
                     <<"">>,
                     Timeout,
-                    TimeoutFun
-                )
+                    TimeoutFun,
+                    true)
             after
                 couch_db_update_notifier:stop(Notify),
                 get_rest_db_updated(ok) % clean out any remaining update messages
@@ -79,8 +89,8 @@ handle_changes(#changes_args{style=Style
                     UserAcc2,
                     Db,
                     StartSeq,
-                    <<>>
-                ),
+                    <<>>,
+                    true),
             end_sending_changes(Callback, UserAcc3, LastSeq, Feed)
         end
     end.
@@ -94,7 +104,7 @@ get_callback_acc(Callback) when is_funct
 make_filter_fun([$_ | _] = FilterName, Style, Req, Db) ->
     builtin_filter_fun(FilterName, Style, Req, Db);
 make_filter_fun(FilterName, Style, Req, Db) ->
-    os_filter_fun(FilterName, Style, Req, Db).
+    {os_filter_fun(FilterName, Style, Req, Db), []}.
 
 os_filter_fun(FilterName, Style, Req, Db) ->
     case [list_to_binary(couch_httpd:unquote(Part))
@@ -133,19 +143,20 @@ os_filter_fun(FilterName, Style, Req, Db
     end.
 
 builtin_filter_fun("_doc_ids", Style, {json_req, {Props}}, _Db) ->
-    filter_docids(couch_util:get_value(<<"doc_ids">>, Props), Style);
+    DocIds = couch_util:get_value(<<"doc_ids">>, Props),
+    {filter_docids(DocIds, Style), DocIds};
 builtin_filter_fun("_doc_ids", Style, #httpd{method='POST'}=Req, _Db) ->
     {Props} = couch_httpd:json_body_obj(Req),
     DocIds =  couch_util:get_value(<<"doc_ids">>, Props, nil),
-    filter_docids(DocIds, Style);
+    {filter_docids(DocIds, Style), DocIds};
 builtin_filter_fun("_doc_ids", Style, #httpd{method='GET'}=Req, _Db) ->
     DocIds = ?JSON_DECODE(couch_httpd:qs_value(Req, "doc_ids", "null")),
-    filter_docids(DocIds, Style);
+    {filter_docids(DocIds, Style), DocIds};
 builtin_filter_fun("_design", Style, _Req, _Db) ->
-    filter_designdoc(Style);
+    {filter_designdoc(Style), []};
 builtin_filter_fun("_view", Style, Req, Db) ->
     ViewName = couch_httpd:qs_value(Req, "view", ""),
-    filter_view(ViewName, Style, Db);
+    {filter_view(ViewName, Style, Db), []};
 builtin_filter_fun(_FilterName, _Style, _Req, _Db) ->
     throw({bad_request, "unknown builtin filter name"}).
 
@@ -244,49 +255,122 @@ start_sending_changes(_Callback, UserAcc
 start_sending_changes(Callback, UserAcc, ResponseType) ->
     Callback(start, ResponseType, UserAcc).
 
-send_changes(Args, Callback, UserAcc, Db, StartSeq, Prepend) ->
+send_changes(Args, Callback, UserAcc, Db, StartSeq, Prepend, FirstRound) ->
     #changes_args{
         include_docs = IncludeDocs,
         conflicts = Conflicts,
         limit = Limit,
         feed = ResponseType,
         dir = Dir,
-        filter = FilterFun
+        filter = FilterName,
+        filter_args = FilterArgs,
+        filter_fun = FilterFun
     } = Args,
-    couch_db:changes_since(
-        Db,
-        StartSeq,
-        fun changes_enumerator/2,
-        [{dir, Dir}],
-        #changes_acc{
-            db = Db,
-            seq = StartSeq,
-            prepend = Prepend,
-            filter = FilterFun,
-            callback = Callback,
-            user_acc = UserAcc,
-            resp_type = ResponseType,
-            limit = Limit,
-            include_docs = IncludeDocs,
-            conflicts = Conflicts
-        }
-    ).
+    Acc0 = #changes_acc{
+        db = Db,
+        seq = StartSeq,
+        prepend = Prepend,
+        filter = FilterFun,
+        callback = Callback,
+        user_acc = UserAcc,
+        resp_type = ResponseType,
+        limit = Limit,
+        include_docs = IncludeDocs,
+        conflicts = Conflicts
+    },
+    case FirstRound of
+    true ->
+        case FilterName of
+        "_doc_ids" when length(FilterArgs) =< ?MAX_DOC_IDS ->
+            send_changes_doc_ids(
+                FilterArgs, Db, StartSeq, fun changes_enumerator/2, Acc0);
+        "_design" ->
+            send_changes_design_docs(
+                Db, StartSeq, fun changes_enumerator/2, Acc0);
+        _ ->
+            couch_db:changes_since(
+                Db, StartSeq, fun changes_enumerator/2, [{dir, Dir}], Acc0)
+        end;
+    false ->
+        couch_db:changes_since(
+            Db, StartSeq, fun changes_enumerator/2, [{dir, Dir}], Acc0)
+    end.
+
+
+send_changes_doc_ids(DocIds, Db, StartSeq, Fun, Acc0) ->
+    Lookups = couch_btree:lookup(Db#db.fulldocinfo_by_id_btree, DocIds),
+    DocInfos = lists:foldl(
+        fun({ok, FDI}, Acc) ->
+            DocInfo = couch_doc:to_doc_info(FDI),
+            case DocInfo#doc_info.high_seq >= StartSeq of
+            true ->
+                [DocInfo | Acc];
+            false ->
+                Acc
+            end;
+        (not_found, Acc) ->
+            Acc
+        end,
+        [], Lookups),
+    send_lookup_changes(DocInfos, Db, Fun, Acc0).
+
+
+send_changes_design_docs(Db, StartSeq, Fun, Acc0) ->
+    FoldFun = fun(FullDocInfo, _, Acc) ->
+        DocInfo = couch_doc:to_doc_info(FullDocInfo),
+        case DocInfo#doc_info.high_seq >= StartSeq of
+        true ->
+            {ok, [DocInfo | Acc]};
+        false ->
+            {ok, Acc}
+        end
+    end,
+    KeyOpts = [{start_key, <<"_design/">>}, {end_key_gt, <<"_design0">>}],
+    {ok, _, DocInfos} = couch_btree:fold(
+        Db#db.fulldocinfo_by_id_btree, FoldFun, [], KeyOpts),
+    send_lookup_changes(DocInfos, Db, Fun, Acc0).
+
+
+send_lookup_changes(DocInfos, Db, Fun, Acc0) ->
+    SortedDocInfos = lists:keysort(#doc_info.high_seq, DocInfos),
+    FinalAcc = try
+        lists:foldl(
+            fun(DocInfo, Acc) ->
+                case Fun(DocInfo, Acc) of
+                {ok, NewAcc} ->
+                    NewAcc;
+                {stop, NewAcc} ->
+                    throw({stop, NewAcc})
+                end
+            end,
+            Acc0, SortedDocInfos)
+    catch
+    throw:{stop, Acc} ->
+        Acc
+    end,
+    {ok, FinalAcc#changes_acc{seq = couch_db:get_update_seq(Db)}}.
+
 
 keep_sending_changes(Args, Callback, UserAcc, Db, StartSeq, Prepend, Timeout,
-    TimeoutFun) ->
+    TimeoutFun, FirstRound) ->
     #changes_args{
         feed = ResponseType,
         limit = Limit,
         db_open_options = DbOptions
     } = Args,
-    % ?LOG_INFO("send_changes start ~p",[StartSeq]),
 
     {ok, ChangesAcc} = send_changes(
-        Args#changes_args{dir=fwd}, Callback, UserAcc, Db, StartSeq, Prepend),
+        Args#changes_args{dir=fwd},
+        Callback,
+        UserAcc,
+        Db,
+        StartSeq,
+        Prepend,
+        FirstRound),
     #changes_acc{
         seq = EndSeq, prepend = Prepend2, user_acc = UserAcc2, limit = NewLimit
     } = ChangesAcc,
-    % ?LOG_INFO("send_changes last ~p",[EndSeq]),
+
     couch_db:close(Db),
     if Limit > NewLimit, ResponseType == "longpoll" ->
         end_sending_changes(Callback, UserAcc2, EndSeq, ResponseType);
@@ -305,7 +389,8 @@ keep_sending_changes(Args, Callback, Use
                     EndSeq,
                     Prepend2,
                     Timeout,
-                    TimeoutFun
+                    TimeoutFun,
+                    false
                 );
             _Else ->
                 end_sending_changes(Callback, UserAcc2, EndSeq, ResponseType)

Modified: couchdb/trunk/src/couchdb/couch_db.hrl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_db.hrl?rev=1173419&r1=1173418&r2=1173419&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_db.hrl (original)
+++ couchdb/trunk/src/couchdb/couch_db.hrl Tue Sep 20 22:55:29 2011
@@ -280,6 +280,8 @@
     heartbeat,
     timeout,
     filter = "",
+    filter_fun,
+    filter_args = [],
     include_docs = false,
     conflicts = false,
     db_open_options = []

Added: couchdb/trunk/test/etap/073-changes.t
URL: http://svn.apache.org/viewvc/couchdb/trunk/test/etap/073-changes.t?rev=1173419&view=auto
==============================================================================
--- couchdb/trunk/test/etap/073-changes.t (added)
+++ couchdb/trunk/test/etap/073-changes.t Tue Sep 20 22:55:29 2011
@@ -0,0 +1,354 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+% 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.
+
+% Verify that compacting databases that are being used as the source or
+% target of a replication doesn't affect the replication and that the
+% replication doesn't hold their reference counters forever.
+
+-record(user_ctx, {
+    name = null,
+    roles = [],
+    handler
+}).
+
+-record(changes_args, {
+    feed = "normal",
+    dir = fwd,
+    since = 0,
+    limit = 1000000000000000,
+    style = main_only,
+    heartbeat,
+    timeout,
+    filter = "",
+    filter_fun,
+    filter_args = [],
+    include_docs = false,
+    conflicts = false,
+    db_open_options = []
+}).
+
+-record(row, {
+    id,
+    seq
+}).
+
+
+test_db_name() -> <<"couch_test_changes">>.
+
+
+main(_) ->
+    test_util:init_code_path(),
+
+    etap:plan(20),
+    case (catch test()) of
+        ok ->
+            etap:end_tests();
+        Other ->
+            etap:diag(io_lib:format("Test died abnormally: ~p", [Other])),
+            etap:bail(Other)
+    end,
+    ok.
+
+
+test() ->
+    couch_server_sup:start_link(test_util:config_files()),
+
+    test_by_doc_ids(),
+    test_by_doc_ids_with_since(),
+    test_by_doc_ids_continuous(),
+
+    couch_server_sup:stop(),
+    ok.
+
+
+test_by_doc_ids() ->
+    {ok, Db} = create_db(test_db_name()),
+
+    {ok, _Rev1} = save_doc(Db, {[{<<"_id">>, <<"doc1">>}]}),
+    {ok, _Rev2} = save_doc(Db, {[{<<"_id">>, <<"doc2">>}]}),
+    {ok, Rev3} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}]}),
+    {ok, _Rev4} = save_doc(Db, {[{<<"_id">>, <<"doc4">>}]}),
+    {ok, _Rev5} = save_doc(Db, {[{<<"_id">>, <<"doc5">>}]}),
+    {ok, _Rev3_2} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}, {<<"_rev">>,
Rev3}]}),
+    {ok, _Rev6} = save_doc(Db, {[{<<"_id">>, <<"doc6">>}]}),
+    {ok, _Rev7} = save_doc(Db, {[{<<"_id">>, <<"doc7">>}]}),
+    {ok, _Rev8} = save_doc(Db, {[{<<"_id">>, <<"doc8">>}]}),
+
+    ChangesArgs = #changes_args{
+        filter = "_doc_ids"
+    },
+    DocIds = [<<"doc3">>, <<"doc4">>, <<"doc9999">>],
+    Req = {json_req, {[{<<"doc_ids">>, DocIds}]}},
+    Consumer = spawn_consumer(test_db_name(), ChangesArgs, Req),
+
+    Rows = wait_finished(Consumer),
+    etap:is(length(Rows), 2, "Received 2 changes rows"),
+    [#row{seq = Seq1, id = Id1}, #row{seq = Seq2, id = Id2}] = Rows,
+    etap:is(Id1, <<"doc4">>, "First row is for doc doc4"),
+    etap:is(Seq1, 4, "First row has seq 4"),
+    etap:is(Id2, <<"doc3">>, "Second row is for doc doc3"),
+    etap:is(Seq2, 6, "Second row has seq 6"),
+
+    stop(Consumer),
+    delete_db(Db).
+
+
+test_by_doc_ids_with_since() ->
+    {ok, Db} = create_db(test_db_name()),
+
+    {ok, _Rev1} = save_doc(Db, {[{<<"_id">>, <<"doc1">>}]}),
+    {ok, _Rev2} = save_doc(Db, {[{<<"_id">>, <<"doc2">>}]}),
+    {ok, Rev3} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}]}),
+    {ok, _Rev4} = save_doc(Db, {[{<<"_id">>, <<"doc4">>}]}),
+    {ok, _Rev5} = save_doc(Db, {[{<<"_id">>, <<"doc5">>}]}),
+    {ok, _Rev3_2} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}, {<<"_rev">>,
Rev3}]}),
+    {ok, _Rev6} = save_doc(Db, {[{<<"_id">>, <<"doc6">>}]}),
+    {ok, _Rev7} = save_doc(Db, {[{<<"_id">>, <<"doc7">>}]}),
+    {ok, _Rev8} = save_doc(Db, {[{<<"_id">>, <<"doc8">>}]}),
+
+    ChangesArgs = #changes_args{
+        filter = "_doc_ids",
+        since = 5
+    },
+    DocIds = [<<"doc3">>, <<"doc4">>, <<"doc9999">>],
+    Req = {json_req, {[{<<"doc_ids">>, DocIds}]}},
+    Consumer = spawn_consumer(test_db_name(), ChangesArgs, Req),
+
+    Rows = wait_finished(Consumer),
+    etap:is(length(Rows), 1, "Received 1 changes rows"),
+    [#row{seq = Seq1, id = Id1}] = Rows,
+    etap:is(Id1, <<"doc3">>, "First row is for doc doc3"),
+    etap:is(Seq1, 6, "First row has seq 6"),
+
+    stop(Consumer),
+    delete_db(Db).
+
+
+test_by_doc_ids_continuous() ->
+    {ok, Db} = create_db(test_db_name()),
+
+    {ok, _Rev1} = save_doc(Db, {[{<<"_id">>, <<"doc1">>}]}),
+    {ok, _Rev2} = save_doc(Db, {[{<<"_id">>, <<"doc2">>}]}),
+    {ok, Rev3} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}]}),
+    {ok, Rev4} = save_doc(Db, {[{<<"_id">>, <<"doc4">>}]}),
+    {ok, _Rev5} = save_doc(Db, {[{<<"_id">>, <<"doc5">>}]}),
+    {ok, Rev3_2} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}, {<<"_rev">>,
Rev3}]}),
+    {ok, _Rev6} = save_doc(Db, {[{<<"_id">>, <<"doc6">>}]}),
+    {ok, _Rev7} = save_doc(Db, {[{<<"_id">>, <<"doc7">>}]}),
+    {ok, _Rev8} = save_doc(Db, {[{<<"_id">>, <<"doc8">>}]}),
+
+    ChangesArgs = #changes_args{
+        filter = "_doc_ids",
+        feed = "continuous"
+    },
+    DocIds = [<<"doc3">>, <<"doc4">>, <<"doc9999">>],
+    Req = {json_req, {[{<<"doc_ids">>, DocIds}]}},
+    Consumer = spawn_consumer(test_db_name(), ChangesArgs, Req),
+
+    pause(Consumer),
+    Rows = get_rows(Consumer),
+
+    etap:is(length(Rows), 2, "Received 2 changes rows"),
+    [#row{seq = Seq1, id = Id1}, #row{seq = Seq2, id = Id2}] = Rows,
+    etap:is(Id1, <<"doc4">>, "First row is for doc doc4"),
+    etap:is(Seq1, 4, "First row has seq 4"),
+    etap:is(Id2, <<"doc3">>, "Second row is for doc doc3"),
+    etap:is(Seq2, 6, "Second row has seq 6"),
+
+    clear_rows(Consumer),
+    {ok, _Rev9} = save_doc(Db, {[{<<"_id">>, <<"doc9">>}]}),
+    {ok, _Rev10} = save_doc(Db, {[{<<"_id">>, <<"doc10">>}]}),
+    unpause(Consumer),
+    pause(Consumer),
+    etap:is(get_rows(Consumer), [], "No new rows"),
+
+    {ok, Rev4_2} = save_doc(Db, {[{<<"_id">>, <<"doc4">>}, {<<"_rev">>,
Rev4}]}),
+    {ok, _Rev11} = save_doc(Db, {[{<<"_id">>, <<"doc11">>}]}),
+    {ok, _Rev4_3} = save_doc(Db, {[{<<"_id">>, <<"doc4">>}, {<<"_rev">>,
Rev4_2}]}),
+    {ok, _Rev12} = save_doc(Db, {[{<<"_id">>, <<"doc12">>}]}),
+    {ok, Rev3_3} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}, {<<"_rev">>,
Rev3_2}]}),
+    unpause(Consumer),
+    pause(Consumer),
+
+    NewRows = get_rows(Consumer),
+    etap:is(length(NewRows), 2, "Received 2 new rows"),
+    [Row14, Row16] = NewRows,
+    etap:is(Row14#row.seq, 14, "First row has seq 14"),
+    etap:is(Row14#row.id, <<"doc4">>, "First row is for doc doc4"),
+    etap:is(Row16#row.seq, 16, "Second row has seq 16"),
+    etap:is(Row16#row.id, <<"doc3">>, "Second row is for doc doc3"),
+
+    clear_rows(Consumer),
+    {ok, _Rev3_4} = save_doc(Db, {[{<<"_id">>, <<"doc3">>}, {<<"_rev">>,
Rev3_3}]}),
+    unpause(Consumer),
+    pause(Consumer),
+    etap:is(get_rows(Consumer), [#row{seq = 17, id = <<"doc3">>}],
+        "Got row for seq 17, doc doc3"),
+
+    unpause(Consumer),
+    stop(Consumer),
+    delete_db(Db).
+
+
+save_doc(Db, Json) ->
+    Doc = couch_doc:from_json_obj(Json),
+    {ok, Rev} = couch_db:update_doc(Db, Doc, []),
+    {ok, couch_doc:rev_to_str(Rev)}.
+
+
+get_rows(Consumer) ->
+    Ref = make_ref(),
+    Consumer ! {get_rows, Ref},
+    receive
+    {rows, Ref, Rows} ->
+        Rows
+    after 3000 ->
+        etap:bail("Timeout getting rows from consumer")
+    end.
+
+
+clear_rows(Consumer) ->
+    Ref = make_ref(),
+    Consumer ! {reset, Ref},
+    receive
+    {ok, Ref} ->
+        ok
+    after 3000 ->
+        etap:bail("Timeout clearing consumer rows")
+    end.
+
+
+stop(Consumer) ->
+    Ref = make_ref(),
+    Consumer ! {stop, Ref},
+    receive
+    {ok, Ref} ->
+        ok
+    after 3000 ->
+        etap:bail("Timeout stopping consumer")
+    end.
+
+
+pause(Consumer) ->
+    Ref = make_ref(),
+    Consumer ! {pause, Ref},
+    receive
+    {paused, Ref} ->
+        ok
+    after 3000 ->
+        etap:bail("Timeout pausing consumer")
+    end.
+
+
+unpause(Consumer) ->
+    Ref = make_ref(),
+    Consumer ! {continue, Ref},
+    receive
+    {ok, Ref} ->
+        ok
+    after 3000 ->
+        etap:bail("Timeout unpausing consumer")
+    end.
+
+
+wait_finished(_Consumer) ->
+    receive
+    {consumer_finished, Rows} ->
+        Rows
+    after 30000 ->
+        etap:bail("Timeout waiting for consumer to finish")
+    end.
+
+
+spawn_consumer(DbName, ChangesArgs0, Req) ->
+    Parent = self(),
+    spawn(fun() ->
+        Callback = fun({change, {Change}, _}, _, Acc) ->
+            Id = couch_util:get_value(<<"id">>, Change),
+            Seq = couch_util:get_value(<<"seq">>, Change),
+            [#row{id = Id, seq = Seq} | Acc];
+        ({stop, _}, _, Acc) ->
+            Parent ! {consumer_finished, lists:reverse(Acc)},
+            stop_loop(Parent, Acc);
+        (_, _, Acc) ->
+            maybe_pause(Parent, Acc)
+        end,
+        {ok, Db} = couch_db:open_int(DbName, []),
+        ChangesArgs = ChangesArgs0#changes_args{timeout = 10, heartbeat = 10},
+        FeedFun = couch_changes:handle_changes(ChangesArgs, Req, Db),
+        try
+            FeedFun({Callback, []})
+        catch throw:{stop, _} ->
+            ok
+        end,
+        catch couch_db:close(Db)
+    end).
+
+
+maybe_pause(Parent, Acc) ->
+    receive
+    {get_rows, Ref} ->
+        Parent ! {rows, Ref, lists:reverse(Acc)},
+        maybe_pause(Parent, Acc);
+    {reset, Ref} ->
+        Parent ! {ok, Ref},
+        maybe_pause(Parent, []);
+    {pause, Ref} ->
+        Parent ! {paused, Ref},
+        pause_loop(Parent, Acc);
+    {stop, Ref} ->
+        Parent ! {ok, Ref},
+        Acc
+    after 0 ->
+        Acc
+    end.
+
+
+pause_loop(Parent, Acc) ->
+    receive
+    {stop, Ref} ->
+        Parent ! {ok, Ref},
+        throw({stop, Acc});
+    {reset, Ref} ->
+        Parent ! {ok, Ref},
+        pause_loop(Parent, []);
+    {continue, Ref} ->
+        Parent ! {ok, Ref},
+        Acc;
+    {get_rows, Ref} ->
+        Parent ! {rows, Ref, lists:reverse(Acc)},
+        pause_loop(Parent, Acc)
+    end.
+
+
+stop_loop(Parent, Acc) ->
+    receive
+    {get_rows, Ref} ->
+        Parent ! {rows, Ref, lists:reverse(Acc)},
+        stop_loop(Parent, Acc);
+    {stop, Ref} ->
+        Parent ! {ok, Ref},
+        Acc
+    end.
+
+
+create_db(DbName) ->
+    couch_db:create(
+        DbName,
+        [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}, overwrite]).
+
+
+delete_db(Db) ->
+    ok = couch_server:delete(
+        couch_db:name(Db), [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}]).

Propchange: couchdb/trunk/test/etap/073-changes.t
------------------------------------------------------------------------------
    svn:executable = *

Modified: couchdb/trunk/test/etap/Makefile.am
URL: http://svn.apache.org/viewvc/couchdb/trunk/test/etap/Makefile.am?rev=1173419&r1=1173418&r2=1173419&view=diff
==============================================================================
--- couchdb/trunk/test/etap/Makefile.am (original)
+++ couchdb/trunk/test/etap/Makefile.am Tue Sep 20 22:55:29 2011
@@ -55,6 +55,7 @@ EXTRA_DIST = \
     065-kt-stemming.t \
     070-couch-db.t \
     072-cleanup.t \
+    073-changes.t \
     080-config-get-set.t \
     081-config-override.1.ini \
     081-config-override.2.ini \



Mime
View raw message