gen_changes.erl

%%% -*- erlang -*-
%%%
%%% This file is part of couchbeam released under the MIT license.
%%% See the NOTICE for more information.

%% @doc gen_changes CouchDB continuous changes consumer behavior
%% This behaviour allows you to create easily a server that consume
%% Couchdb continuous changes

-module(gen_changes).

-include("couchbeam.hrl").

-behavior(gen_server).

-export([start_link/4]).
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).
-export([behaviour_info/1]).

-export([call/2,
         call/3,
         cast/2]).

-export([stop/1, get_seq/1]).


behaviour_info(callbacks) ->
    [{init, 1},
     {handle_change, 2},
     {handle_call, 3},
     {handle_cast, 2},
     {handle_info, 2},
     {terminate, 2}];
behaviour_info(_) ->
    undefined.

call(Name, Request) ->
    gen_server:call(Name, Request).

call(Name, Request, Timeout) ->
    gen_server:call(Name, Request, Timeout).

cast(Dest, Request) ->
    gen_server:cast(Dest, Request).

%% @doc create a gen_changes process as part of a supervision tree.
%% The function should be called, directly or indirectly, by the supervisor.
%% @spec start_link(Module, Db::db(), Options::changesoptions(),
%%                  InitArgs::list()) -> term()
%%       changesoptions() = [changeoption()]
%%       changeoption() = {include_docs, string()} |
%%                  {filter, string()} |
%%                  {since, integer()|string()} |
%%                  {heartbeat, string()|boolean()}
start_link(Module, Db, Options, InitArgs) ->
    gen_server:start_link(?MODULE, [Module, Db, Options, InitArgs], []).

init([Module, Db, Options, InitArgs]) ->
    case Module:init(InitArgs) of
        {ok, ModState} ->
            case couchbeam_changes:follow(Db, Options) of
            {ok, StreamRef} ->
                LastSeq = proplists:get_value(since, Options, 0),
                {ok, #gen_changes_state{stream_ref=StreamRef,
                                        mod=Module,
                                        modstate=ModState,
                                        db=Db,
                                        options=Options,
                                        last_seq=LastSeq}};
            {error, Error} ->
                Module:terminate(Error, ModState),
                {stop, Error}
            end;
        Error ->
            Error
    end.

stop(Pid) when is_pid(Pid) ->
    gen_server:cast(Pid, stop).

get_seq(Pid) when is_pid(Pid) ->
    gen_server:call(Pid, get_seq).

handle_call(get_seq, _From, State=#gen_changes_state{last_seq=Seq}) ->
    {reply, Seq, State};
handle_call(Request, From,
            State=#gen_changes_state{mod=Module, modstate=ModState}) ->
    case Module:handle_call(Request, From, ModState) of
        {reply, Reply, NewModState} ->
            {reply, Reply, State#gen_changes_state{modstate=NewModState}};
        {reply, Reply, NewModState, A}
          when A =:= hibernate orelse is_number(A) ->
            {reply, Reply, State#gen_changes_state{modstate=NewModState}, A};
        {noreply, NewModState} ->
            {noreply, State#gen_changes_state{modstate=NewModState}};
        {noreply, NewModState, A} when A =:= hibernate orelse is_number(A) ->
            {noreply, State#gen_changes_state{modstate=NewModState}, A};
        {stop, Reason, NewModState} ->
            {stop, Reason, State#gen_changes_state{modstate=NewModState}};
        {stop, Reason, Reply, NewModState} ->
            {stop, Reason, Reply, State#gen_changes_state{modstate=NewModState}}
  end.

handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(Msg, State=#gen_changes_state{mod=Module, modstate=ModState}) ->
    case Module:handle_cast(Msg, ModState) of
        {noreply, NewModState} ->
            {noreply, State#gen_changes_state{modstate=NewModState}};
        {noreply, NewModState, A} when A =:= hibernate orelse is_number(A) ->
            {noreply, State#gen_changes_state{modstate=NewModState}, A};
        {stop, Reason, NewModState} ->
            {stop, Reason, State#gen_changes_state{modstate=NewModState}}
    end.


handle_info({Ref, Msg},
        State=#gen_changes_state{mod=Module, modstate=ModState,
            stream_ref=Ref}) ->

    State2 = case Msg of
        {done, LastSeq} ->
            State#gen_changes_state{last_seq=LastSeq};
        {change, Change} ->
            Seq = couchbeam_doc:get_value(<<"seq">>, Change),
            State#gen_changes_state{last_seq=Seq}
    end,

    case catch Module:handle_change(Msg, ModState) of
        {noreply, NewModState} ->
            {noreply, State2#gen_changes_state{modstate=NewModState}};
        {noreply, NewModState, A} when A =:= hibernate orelse is_number(A) ->
            {noreply, State2#gen_changes_state{modstate=NewModState}, A};
        {stop, Reason, NewModState} ->
            {stop, Reason, State2#gen_changes_state{modstate=NewModState}}
    end;


handle_info({Ref, {error, Error}},
        State=#gen_changes_state{stream_ref=Ref, last_seq=LastSeq}) ->
    handle_info({error, [Error, {last_seq, LastSeq}]}, State);

handle_info(Info, State=#gen_changes_state{mod=Module, modstate=ModState}) ->
    case Module:handle_info(Info, ModState) of
        {noreply, NewModState} ->
            {noreply, State#gen_changes_state{modstate=NewModState}};
        {noreply, NewModState, A} when A =:= hibernate orelse is_number(A) ->
            {noreply, State#gen_changes_state{modstate=NewModState}, A};
        {stop, Reason, NewModState} ->
            {stop, Reason, State#gen_changes_state{modstate=NewModState}}
    end.

code_change(_OldVersion, State, _Extra) ->
    %% TODO:  support code changes?
    {ok, State}.

terminate(Reason, #gen_changes_state{stream_ref=Ref,
        mod=Module, modstate=ModState}) ->
    Module:terminate(Reason, ModState),
    couchbeam_changes:cancel_stream(Ref),
    ok.