C0 code coverage information

Generated on 2009-08-09 21:22:55 with etap 0.3.4.

....1 % Licensed under the Apache License, Version 2.0 (the "License"); you may not
....2 % use this file except in compliance with the License. You may obtain a copy of
....3 % the License at
....4 %
....5 %   http://www.apache.org/licenses/LICENSE-2.0
....6 %
....7 % Unless required by applicable law or agreed to in writing, software
....8 % distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
....9 % WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
...10 % License for the specific language governing permissions and limitations under
...11 % the License.
...12 
...13 -module(couch_rep_changes_feed).
...14 -behaviour(gen_server).
...15 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, 
...16     code_change/3]).
...17 
...18 -export([start_link/4, next/1, stop/1]).
...19 
...20 -define(BUFFER_SIZE, 1000).
...21 
...22 -include("couch_db.hrl").
...23 -include("../ibrowse/ibrowse.hrl").
...24 
...25 -record (state, {
...26     changes_from = nil,
...27     changes_loop = nil,
...28     last_seq,
...29     conn = nil,
...30     reqid = nil,
...31     by_seq_from = nil,
...32     by_seq_loop = nil,
...33     complete = false,
...34     count = 0,
...35     partial_chunk = nil,
...36     reply_to = nil,
...37     rows = queue:new()
...38 }).
...39 
...40 start_link(Parent, Source, StartSeq, PostProps) ->
...41     gen_server:start_link(?MODULE, [Parent, Source, StartSeq, PostProps], []).
...42 
...43 next(Server) ->
...44     gen_server:call(Server, next_changes, infinity).
...45 
...46 stop(Server) ->
...47     gen_server:call(Server, stop).
...48 
...49 init([_Parent, #http_db{}=Source, Since, PostProps]) ->
...50     Feed = case proplists:get_value(<<"continuous">>, PostProps, false) of
...51     false ->
...52         normal;
...53     true ->
...54         continuous
...55     end,
...56     Pid = couch_rep_httpc:spawn_link_worker_process(Source),
...57     Req = Source#http_db{
...58         resource = "_changes",
...59         qs = [{style, all_docs}, {heartbeat, true}, {since, Since},
...60             {feed, Feed}],
...61         conn = Pid,
...62         options = [{stream_to, {self(), once}}, {response_format, binary}],
...63         headers = Source#http_db.headers -- [{"Accept-Encoding", "gzip"}]
...64     },    
...65     {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Req),
...66 
...67     receive
...68     {ibrowse_async_headers, ReqId, "200", _} ->
...69         ibrowse:stream_next(ReqId),
...70         {ok, #state{conn=Pid, last_seq=Since, reqid=ReqId}};
...71     {ibrowse_async_headers, ReqId, "301", Hdrs} ->
...72         catch ibrowse:stop_worker_process(Pid),
...73         Url2 = mochiweb_headers:get_value("Location", mochiweb_headers:make(Hdrs)),
...74         %% TODO use couch_httpc:request instead of start_http_request
...75         {Pid2, ReqId2} = start_http_request(Url2),
...76         receive {ibrowse_async_headers, ReqId2, "200", _} ->
...77             {ok, #state{conn=Pid2, last_seq=Since, reqid=ReqId2}}
...78         after 30000 ->
...79             {stop, changes_timeout}
...80         end;
...81     {ibrowse_async_headers, ReqId, "404", _} ->
...82         catch ibrowse:stop_worker_process(Pid),
...83         ?LOG_INFO("source doesn't have _changes, trying _all_docs_by_seq", []),
...84         Self = self(),
...85         BySeqPid = spawn_link(fun() -> by_seq_loop(Self, Source, Since) end),
...86         {ok, #state{last_seq=Since, by_seq_loop=BySeqPid}};
...87     {ibrowse_async_headers, ReqId, Code, _} ->
...88         {stop, {changes_error_code, list_to_integer(Code)}}
...89     after 10000 ->
...90         {stop, changes_timeout}
...91     end;
...92 
...93 init([_Parent, Source, Since, PostProps]) ->
...94     process_flag(trap_exit, true),
...95     Server = self(),
...96     ChangesPid =
...97     case proplists:get_value(<<"continuous">>, PostProps, false) of
...98     false ->
...99         spawn_link(fun() -> send_local_changes_once(Server, Source, Since) end);
..100     true ->
..101         spawn_link(fun() -> send_local_changes_forever(Server, Source, Since) end)
..102     end,
..103     {ok, #state{changes_loop=ChangesPid}}.
..104 
..105 handle_call({add_change, Row}, From, State) ->
..106     handle_add_change(Row, From, State);
..107 
..108 handle_call(next_changes, From, State) ->
..109     handle_next_changes(From, State);
..110     
..111 handle_call(stop, _From, State) ->
..112     #state{
..113         changes_loop = ChangesPid,
..114         conn = Conn
..115     } = State,
..116     if is_pid(ChangesPid) -> exit(ChangesPid, stop); true -> ok end,
..117     if is_pid(Conn) -> catch ibrowse:stop_worker_process(Conn); true -> ok end,
..118     {stop, normal, ok, State}.
..119 
..120 handle_cast(_Msg, State) ->
..121     {noreply, State}.
..122 
..123 handle_info({ibrowse_async_headers, Id, Code, Hdrs}, #state{reqid=Id}=State) ->
..124     handle_headers(list_to_integer(Code), Hdrs, State);
..125 
..126 handle_info({ibrowse_async_response, Id, Msg}, #state{reqid=Id} = State) ->
..127     handle_response(Msg, State);
..128 
..129 handle_info({ibrowse_async_response_end, Id}, #state{reqid=Id} = State) ->
..130     handle_feed_completion(State);
..131 
..132 handle_info({'EXIT', From, normal}, #state{changes_loop=From} = State) ->
..133     handle_feed_completion(State);
..134 
..135 handle_info({'EXIT', From, Reason}, #state{changes_loop=From} = State) ->
..136     ?LOG_ERROR("changes_loop died with reason ~p", [Reason]),
..137     {stop, changes_loop_died, State};
..138 
..139 handle_info(Msg, State) ->
..140     ?LOG_INFO("unexpected message ~p", [Msg]),
..141     {noreply, State}.
..142 
..143 terminate(_Reason, #state{conn=Pid}) when is_pid(Pid) ->
..144     catch ibrowse:stop_worker_process(Pid),
..145     ok;
..146 terminate(_Reason, _State) ->
..147     ok.
..148 
..149 code_change(_OldVsn, State, _Extra) ->
..150     {ok, State}.
..151 
..152 %internal funs
..153 
..154 handle_add_change(Row, From, #state{reply_to=nil} = State) ->
..155     #state{
..156         count = Count,
..157         rows = Rows
..158     } = State,
..159     NewState = State#state{count=Count+1, rows=queue:in(Row,Rows)},
..160     if Count < ?BUFFER_SIZE ->
..161         {reply, ok, NewState};
..162     true ->
..163         {noreply, NewState#state{changes_from=From}}
..164     end;
..165 handle_add_change(Row, _From, #state{count=0} = State) ->
..166     gen_server:reply(State#state.reply_to, [Row]),
..167     {reply, ok, State#state{reply_to=nil}}.
..168 
..169 handle_next_changes(From, #state{count=0}=State) ->
..170     if State#state.complete ->
..171         {stop, normal, complete, State};
..172     true ->
..173         {noreply, State#state{reply_to=From}}
..174     end;
..175 handle_next_changes(_From, State) ->
..176     #state{
..177         changes_from = ChangesFrom,
..178         rows = Rows
..179     } = State,
..180     NewState = State#state{count=0, changes_from=nil, rows=queue:new()},
..181     ok = maybe_stream_next(NewState),
..182     if ChangesFrom =/= nil -> gen_server:reply(ChangesFrom, ok); true -> ok end,
..183     {reply, queue:to_list(Rows), NewState}.
..184 
..185 handle_headers(200, _, State) ->
..186     ok = maybe_stream_next(State),
..187     {noreply, State};
..188 handle_headers(301, Hdrs, State) ->
..189     catch ibrowse:stop_worker_process(State#state.conn),
..190     Url = mochiweb_headers:get_value("Location", mochiweb_headers:make(Hdrs)),
..191     %% TODO use couch_httpc:request instead of start_http_request
..192     {Pid, ReqId} = start_http_request(Url),
..193     {noreply, State#state{conn=Pid, reqid=ReqId}};
..194 handle_headers(Code, Hdrs, State) ->
..195     ?LOG_ERROR("replicator changes feed failed with code ~s and Headers ~n~p",
..196         [Code,Hdrs]),
..197     {stop, {error, Code}, State}.
..198 
..199 handle_response({error, Reason}, State) ->
..200     {stop, {error, Reason}, State};
..201 handle_response(<<"\n">>, State) ->
..202     ?LOG_DEBUG("got a heartbeat from the remote server", []),
..203     {noreply, State};
..204 handle_response(<<"{\"results\":[\n">>, State) ->
..205     ok = maybe_stream_next(State),
..206     {noreply, State};
..207 handle_response(<<"\n],\n\"last_seq\":", LastSeqStr/binary>>, State) ->
..208     LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
..209     {noreply, State#state{last_seq = LastSeq}};
..210 handle_response(Chunk, #state{partial_chunk=nil} = State) ->
..211     #state{
..212         count = Count,
..213         rows = Rows
..214     } = State,
..215     ok = maybe_stream_next(State),
..216     try
..217         Row = decode_row(Chunk),
..218         case State of
..219         #state{reply_to=nil} ->
..220             {noreply, State#state{count=Count+1, rows = queue:in(Row, Rows)}};
..221         #state{count=0, reply_to=From}->
..222             gen_server:reply(From, [Row]),
..223             {noreply, State#state{reply_to=nil}}
..224         end
..225     catch
..226     throw:{invalid_json, Bad} ->
..227         {noreply, State#state{partial_chunk = Bad}}
..228     end;
..229 handle_response(Chunk, State) ->
..230     #state{
..231         count = Count,
..232         partial_chunk = Partial,
..233         rows = Rows
..234     } = State,
..235     ok = maybe_stream_next(State),
..236     try
..237         Row = decode_row(<>),
..238         {noreply, case State of
..239         #state{reply_to=nil} ->
..240             State#state{count=Count+1, partial_chunk=nil, rows=queue:in(Row,Rows)};
..241         #state{count=0, reply_to=From}->
..242             gen_server:reply(From, [Row]),
..243             State#state{reply_to=nil, partial_chunk=nil}
..244         end}
..245     catch
..246     throw:{invalid_json, Bad} ->
..247         {noreply, State#state{partial_chunk = Bad}}
..248     end.
..249 
..250 handle_feed_completion(#state{reply_to=nil} = State)->
..251     {noreply, State#state{complete=true}};
..252 handle_feed_completion(#state{count=0} = State) ->
..253     gen_server:reply(State#state.reply_to, complete),
..254     {stop, normal, State}.
..255 
..256 by_seq_loop(Server, Source, StartSeq) ->
..257     Req = Source#http_db{
..258         resource = "_all_docs_by_seq",
..259         qs = [{limit, 1000}, {startkey, StartSeq}]
..260     },
..261     {Results} = couch_rep_httpc:request(Req),
..262     if Results =:= [] -> exit(normal); true -> ok end,
..263     EndSeq = lists:foldl(fun({RowInfoList}, _) ->
..264         Id = proplists:get_value(<<"id">>, RowInfoList),
..265         Seq = proplists:get_value(<<"key">>, RowInfoList),
..266         {RowProps} = proplists:get_value(<<"value">>, RowInfoList),
..267         RawRevs = [
..268             proplists:get_value(<<"rev">>, RowProps),
..269             proplists:get_value(<<"conflicts">>, RowProps, []),
..270             proplists:get_value(<<"deleted_conflicts">>, RowProps, [])
..271         ],
..272         ParsedRevs = couch_doc:parse_revs(lists:flatten(RawRevs)),
..273         Change = {[
..274             {<<"seq">>, Seq},
..275             {<<"Id">>, Id},
..276             {<<"changes">>, [{[{<<"rev">>,R}]} || R <- ParsedRevs]}
..277         ]},
..278         gen_server:call(Server, {add_change, Change}),
..279         Seq
..280     end, 0, proplists:get_value(<<"rows">>, Results)),
..281     by_seq_loop(Server, Source, EndSeq+1).
..282 
..283 decode_row(<<",\n", Rest/binary>>) ->
..284     decode_row(Rest);
..285 decode_row(Row) ->
..286     {[Seq, Id, {<<"changes">>,C}]} = ?JSON_DECODE(Row),
..287     C2 = [{[{<<"rev">>,couch_doc:parse_rev(R)}]} || {[{<<"rev">>,R}]} <- C],
..288     {[Seq, Id, {<<"changes">>,C2}]}.
..289 
..290 flush_updated_messages() ->
..291     receive updated -> flush_updated_messages()
..292     after 0 -> ok
..293     end.
..294 
..295 local_update_notification(Self, DbName, {updated, DbName}) ->
..296     Self ! updated;
..297 local_update_notification(Self, DbName, {deleted, DbName}) ->
..298     Self ! deleted;
..299 local_update_notification(_, _, _) ->
..300     ok.
..301 
..302 maybe_stream_next(#state{reqid=nil}) ->
..303     ok;
..304 maybe_stream_next(#state{complete=false, count=N} = S) when N < ?BUFFER_SIZE ->
..305     ibrowse:stream_next(S#state.reqid);
..306 maybe_stream_next(_) ->
..307     ok.
..308 
..309 send_local_changes_forever(Server, Db, Since) ->
..310     #db{name = DbName, user_ctx = UserCtx} = Db,
..311     Self = self(),
..312     {ok, _} = couch_db_update_notifier:start_link(
..313         fun(Msg) -> local_update_notification(Self, DbName, Msg) end),
..314     {ok, NewSeq} = send_local_changes_once(Server, Db, Since),
..315     couch_db:close(Db),
..316     ok = wait_db_updated(),
..317     {ok, NewDb} = couch_db:open(DbName, [{user_ctx, UserCtx}]),
..318     send_local_changes_forever(Server, NewDb, NewSeq).
..319 
..320 send_local_changes_once(Server, Db, Since) ->
..321     FilterFun =
..322     fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
..323         {[{<<"rev">>, Rev}]}
..324     end,
..325 
..326     ChangesFun =
..327     fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, _) ->
..328         Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos],
..329         Results = [Result || Result <- Results0, Result /= null],
..330         if Results /= [] ->
..331             Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]},
..332             gen_server:call(Server, {add_change, Change}, infinity);
..333         true ->
..334             ok
..335         end,
..336         {ok, Seq}
..337     end,
..338 
..339     couch_db:changes_since(Db, all_docs, Since, ChangesFun, Since).
..340 
..341 start_http_request(RawUrl) ->
..342     Url = ibrowse_lib:parse_url(RawUrl),
..343     {ok, Pid} = ibrowse:spawn_link_worker_process(Url#url.host, Url#url.port),
..344     Opts = [
..345         {stream_to, {self(), once}},
..346         {inactivity_timeout, 30000},
..347         {response_format, binary}
..348     ],
..349     {ibrowse_req_id, Id} = 
..350         ibrowse:send_req_direct(Pid, RawUrl, [], get, [], Opts, infinity),
..351     {Pid, Id}.
..352 
..353 wait_db_updated() ->
..354     receive deleted ->
..355         exit(deleted)
..356     after 0 ->
..357         receive updated ->
..358             flush_updated_messages()
..359         end
..360     end.

Generated using etap 0.3.4.