extract couch_httpd changes API in its own module

5 files changed, 206 insertions, 13 deletions

diff --git a/apps/couch_httpd/src/couch_httpd_changes.erl b/apps/couch_httpd/src/couch_httpd_changes.erl
new file mode 100644
index 000000000..1e431e9c7
--- /dev/null
+++ b/apps/couch_httpd/src/couch_httpd_changes.erl
@@ -0,0 +1,174 @@
+% 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_httpd_changes).
+
+-export([handle_changes_req/2]).
+
+-include_lib("couch/include/couch_db.hrl").
+
+handle_changes_req(#httpd{method='POST'}=Req, Db) ->
+    couch_httpd:validate_ctype(Req, "application/json"),
+    handle_changes_req1(Req, Db);
+handle_changes_req(#httpd{method='GET'}=Req, Db) ->
+    handle_changes_req1(Req, Db);
+handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) ->
+    couch_httpd:send_method_not_allowed(Req, "GET,HEAD,POST").
+
+handle_changes_req1(Req, #db{name=DbName}=Db) ->
+    AuthDbName = ?l2b(couch_config:get("couch_httpd_auth", "authentication_db")),
+    case AuthDbName of
+    DbName ->
+        % in the authentication database, _changes is admin-only.
+        ok = couch_db:check_is_admin(Db);
+    _Else ->
+        % on other databases, _changes is free for all.
+        ok
+    end,
+    handle_changes_req2(Req, Db).
+
+handle_changes_req2(Req, Db) ->
+    MakeCallback = fun(Resp) ->
+        fun({change, {ChangeProp}=Change, _}, "eventsource") ->
+            Seq = proplists:get_value(<<"seq">>, ChangeProp),
+            couch_httpd:send_chunk(Resp, ["data: ", ?JSON_ENCODE(Change),
+                              "\n", "id: ", ?JSON_ENCODE(Seq),
+                              "\n\n"]);
+        ({change, Change, _}, "continuous") ->
+            couch_httpd:send_chunk(Resp, [?JSON_ENCODE(Change) | "\n"]);
+        ({change, Change, Prepend}, _) ->
+            couch_httpd:send_chunk(Resp, [Prepend, ?JSON_ENCODE(Change)]);
+        (start, "eventsource") ->
+            ok;
+        (start, "continuous") ->
+            ok;
+        (start, _) ->
+            couch_httpd:send_chunk(Resp, "{\"results\":[\n");
+        ({stop, _EndSeq}, "eventsource") ->
+            couch_httpd:end_json_response(Resp);
+        ({stop, EndSeq}, "continuous") ->
+            couch_httpd:send_chunk(
+                Resp,
+                [?JSON_ENCODE({[{<<"last_seq">>, EndSeq}]}) | "\n"]
+            ),
+            couch_httpd:end_json_response(Resp);
+        ({stop, EndSeq}, _) ->
+            couch_httpd:send_chunk(
+                Resp,
+                io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq])
+            ),
+            couch_httpd:end_json_response(Resp);
+        (timeout, _) ->
+            couch_httpd:send_chunk(Resp, "\n")
+        end
+    end,
+    ChangesArgs = parse_changes_query(Req, Db),
+    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),
+        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;
+    "eventsource" ->
+        Headers = [
+            {"Content-Type", "text/event-stream"},
+            {"Cache-Control", "no-cache"}
+        ],
+        {ok, Resp} = couch_httpd:start_chunked_response(Req, 200, Headers),
+        fun(FeedChangesFun) ->
+            FeedChangesFun(MakeCallback(Resp))
+        end;
+    _ ->
+        % "longpoll" or "continuous"
+        {ok, Resp} = couch_httpd:start_json_response(Req, 200),
+        fun(FeedChangesFun) ->
+            FeedChangesFun(MakeCallback(Resp))
+        end
+    end,
+    couch_stats_collector:increment(
+        {httpd, clients_requesting_changes}
+    ),
+    try
+        WrapperFun(ChangesFun)
+    after
+    couch_stats_collector:decrement(
+        {httpd, clients_requesting_changes}
+    )
+    end.
+
+parse_changes_query(Req, Db) ->
+    ChangesArgs = lists:foldl(fun({Key, Value}, Args) ->
+        case {string:to_lower(Key), Value} of
+        {"feed", _} ->
+            Args#changes_args{feed=Value};
+        {"descending", "true"} ->
+            Args#changes_args{dir=rev};
+        {"since", "now"} ->
+            UpdateSeq = couch_util:with_db(Db#db.name, fun(WDb) ->
+                                        couch_db:get_update_seq(WDb)
+                                end),
+            Args#changes_args{since=UpdateSeq};
+        {"since", _} ->
+            Args#changes_args{since=list_to_integer(Value)};
+        {"last-event-id", _} ->
+            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};
+        {"attachments", "true"} ->
+            Opts = Args#changes_args.doc_options,
+            Args#changes_args{doc_options=[attachments|Opts]};
+        {"att_encoding_info", "true"} ->
+            Opts = Args#changes_args.doc_options,
+            Args#changes_args{doc_options=[att_encoding_info|Opts]};
+        {"conflicts", "true"} ->
+            Args#changes_args{conflicts=true};
+        {"filter", _} ->
+            Args#changes_args{filter=Value};
+        _Else -> % unknown key value pair, ignore.
+            Args
+        end
+    end, #changes_args{}, couch_httpd:qs(Req)),
+    %% if it's an EventSource request with a Last-event-ID header
+    %% that should override the `since` query string, since it's
+    %% probably the browser reconnecting.
+    case ChangesArgs#changes_args.feed of
+        "eventsource" ->
+            case couch_httpd:header_value(Req, "last-event-id") of
+                undefined ->
+                    ChangesArgs;
+                Value ->
+                    ChangesArgs#changes_args{since=list_to_integer(Value)}
+            end;
+        _ ->
+            ChangesArgs
+    end.
diff --git a/apps/couch_httpd/src/couch_httpd_db.erl b/apps/couch_httpd/src/couch_httpd_db.erl
index 45a6dd517..0d1e0f87f 100644
--- a/apps/couch_httpd/src/couch_httpd_db.erl
+++ b/apps/couch_httpd/src/couch_httpd_db.erl
@@ -19,10 +19,10 @@
     handle_design_info_req/3]).
 
 -import(couch_httpd,
-    [send_json/2,send_json/3,send_json/4,send_method_not_allowed/2,
-    start_json_response/2,send_chunk/2,last_chunk/1,end_json_response/1,
-    start_chunked_response/3, absolute_uri/2, send/2,
-    start_response_length/4, send_error/4]).
+        [send_json/2,send_json/3,send_json/4,send_method_not_allowed/2,
+         start_json_response/2,send_chunk/2,last_chunk/1,end_json_response/1,
+         start_chunked_response/3, absolute_uri/2, send/2,
+         start_response_length/4, send_error/4]).
 
 -record(doc_query_args, {
     options = [],
diff --git a/apps/couch_index/src/couch_index.erl b/apps/couch_index/src/couch_index.erl
index c09a1105a..c48c06674 100644
--- a/apps/couch_index/src/couch_index.erl
+++ b/apps/couch_index/src/couch_index.erl
@@ -219,9 +219,18 @@ handle_cast({new_state, NewIdxState}, State) ->
     } = State,
     assert_signature_match(Mod, OldIdxState, NewIdxState),
     CurrSeq = Mod:get(update_seq, NewIdxState),
+
+    DbName = Mod:get(db_name, NewIdxState),
+    DDocId = Mod:get(idx_name, NewIdxState),
+
+    %% notify to event listeners that the index has been
+    %% updated
+    couch_index_event:notify({index_update,
+                              {DbName, DDocId,
+                               Mod}}),
     Args = [
-        Mod:get(db_name, NewIdxState),
-        Mod:get(idx_name, NewIdxState),
+        DbName,
+        DDocId,
         CurrSeq
     ],
     ?LOG_DEBUG("Updated index for db: ~s idx: ~s seq: ~B", Args),
@@ -242,12 +251,27 @@ handle_cast(stop, State) ->
     {stop, normal, State};
 handle_cast(delete, State) ->
     #st{mod=Mod, idx_state=IdxState} = State,
+    DbName = Mod:get(db_name, IdxState),
+    DDocId = Mod:get(idx_name, IdxState),
+
     ok = Mod:delete(IdxState),
+
+    %% notify about the index deletion
+    couch_index_event:notify({index_delete,
+                              {DbName, DDocId, Mod}}),
+
     {stop, normal, State};
 handle_cast(ddoc_updated, State) ->
     #st{mod = Mod, idx_state = IdxState, waiters = Waiters} = State,
     DbName = Mod:get(db_name, IdxState),
     DDocId = Mod:get(idx_name, IdxState),
+
+    %% notify to event listeners that the index has been
+    %% updated
+    couch_index_event:notify({index_update,
+                              {DbName, DDocId,
+                               Mod}}),
+
     Shutdown = couch_util:with_db(DbName, fun(Db) ->
         case couch_db:open_doc(Db, DDocId, [ejson_body]) of
             {not_found, deleted} ->
diff --git a/apps/couch_mrview/src/couch_mrview_updater.erl b/apps/couch_mrview/src/couch_mrview_updater.erl
index a23def6a2..be1055c71 100644
--- a/apps/couch_mrview/src/couch_mrview_updater.erl
+++ b/apps/couch_mrview/src/couch_mrview_updater.erl
@@ -182,7 +182,7 @@ map_docs(Parent, State0) ->
     end.
 
 
-write_results(Parent, #mrst{db_name=DbName, idx_name=IdxName}=State) ->
+write_results(Parent, State) ->
     case couch_work_queue:dequeue(State#mrst.write_queue) of
         closed ->
             Parent ! {new_state, State};
@@ -192,11 +192,6 @@ write_results(Parent, #mrst{db_name=DbName, idx_name=IdxName}=State) ->
                                                            [], dict:new()),
             NewState = write_kvs(State, Seq, ViewKVs, DocIdKeys, Log),
             send_partial(NewState#mrst.partial_resp_pid, NewState),
-
-            % notifify the view update
-            couch_index_event:notify({index_update, {DbName, IdxName,
-                                                     couch_mrview_index}}),
-
             write_results(Parent, NewState)
     end.
 
diff --git a/etc/couchdb/couch.ini b/etc/couchdb/couch.ini
index 4dbe9032d..ad8ac8bfb 100644
--- a/etc/couchdb/couch.ini
+++ b/etc/couchdb/couch.ini
@@ -161,7 +161,7 @@ _plugins = {couch_plugins_httpd, handle_req}
 
 [httpd_db_handlers]
 _all_docs = {couch_mrview_http, handle_all_docs_req}
-_changes = {couch_httpd_db, handle_changes_req}
+_changes = {couch_httpd_changes, handle_changes_req}
 _compact = {couch_httpd_db, handle_compact_req}
 _design = {couch_httpd_db, handle_design_req}
 _temp_view = {couch_mrview_http, handle_temp_view_req}