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

-module(couchbeam_view_stream).

-export([start_link/4]).

-export([init_stream/5,
         maybe_continue/1,
         system_continue/3,
         system_terminate/4,
         system_code_change/4]).


-export([init/1,
         handle_event/2,
         wait_rows/2,
         wait_rows1/2,
         wait_val/2,
         collect_object/2,
         maybe_continue_decoding/1]).


-include("couchbeam.hrl").

-record(state, {parent,
                owner,
                req,
                ref,
                mref,
                client_ref=nil,
                decoder,
                async=normal}).

-record(viewst, {parent,
                 owner,
                 ref,
                 mref,
                 client_ref,
                 async=false}).


-define(TIMEOUT, 10000).
-define(DEFAULT_CHANGES_POLL, 5000). % we check every 5secs



start_link(Owner, StreamRef, {Db, Url, Args}, StreamOptions) ->
    proc_lib:start_link(?MODULE, init_stream, [self(), Owner, StreamRef,
                                               {Db, Url, Args},
                                               StreamOptions]).

init_stream(Parent, Owner, StreamRef, {_Db, _Url, _Args}=Req,
            StreamOptions) ->


    Async = proplists:get_value(async, StreamOptions, normal),

    %% monitor the process receiving§ the messages
    MRef = erlang:monitor(process, Owner),

    InitState = #state{parent=Parent,
                       owner=Owner,
                       req=Req,
                       ref=StreamRef,
                       mref=MRef,
                       async=Async},

    %% connect to the view
    case do_init_stream(Req, InitState) of
        {ok, State} ->
            %% register the stream
            ets:insert(couchbeam_view_streams, [{StreamRef, self()}]),
            %% tell to the parent that we are ok
            proc_lib:init_ack(Parent, {ok, self()}),
            %% start the loop
            loop(State);
        Error ->
            proc_lib:init_ack(Parent, Error)
    end,
    %% stop to monitor the parent
    erlang:demonitor(MRef),
    ok.

do_init_stream({#db{options=Opts}, Url, Args}, #state{mref=MRef}=State) ->
    %% we are doing the request asynchronously
    FinalOpts = [{async, once} | Opts],
    Reply = case Args#view_query_args.method of
        get ->
            hackney:request(get, Url, [], <<>>, FinalOpts);
        post ->
            Body = couchbeam_ejson:encode({[{<<"keys">>,
                                             Args#view_query_args.keys}]}),
            Headers = [{<<"Content-Type">>, <<"application/json">>}],
            hackney:request(post, Url, Headers, Body, FinalOpts)
    end,

    case Reply of
        {ok, Ref} ->
            receive
                {'DOWN', MRef, _, _, _} ->
                    %% parent exited there is no need to continue
                    exit(normal);
                {hackney_response, Ref, {status, 200, _}} ->
                    #state{parent=Parent,
                           owner=Owner,
                           ref=StreamRef,
                           async=Async} = State,

                    DecoderFun = jsx:decoder(?MODULE, [Parent, Owner,
                                                       StreamRef, MRef, Ref,
                                                       Async], [stream]),
                    {ok, State#state{client_ref=Ref,
                                     decoder=DecoderFun}};

                {hackney_response, Ref, {status, 404, _}} ->
                    {error, not_found};
                {hackney_response, Ref, {status, Status, Reason}} ->
                    {error, {http_error, Status, Reason}};
                {hackney_response, Ref, {error, Reason}} ->
                    exit(Reason)
            after ?TIMEOUT ->
                    exit(timeout)
            end;
        Error ->
            {error, Error}
    end.



loop(#state{owner=Owner,
            ref=StreamRef,
            mref=MRef,
            client_ref=ClientRef}=State) ->

    hackney:stream_next(ClientRef),
    receive
        {'DOWN', MRef, _, _, _} ->
            %% parent exited there is no need to continue
            exit(normal);
        {hackney_response, ClientRef, {headers, _Headers}} ->
            loop(State);
        {hackney_response, ClientRef, done} ->
            %% unregister the stream
            ets:delete(couchbeam_view_streams, StreamRef),
            %% tell to the owner that we are done and exit,
            Owner ! {StreamRef, done};
        {hackney_response, ClientRef, Data} when is_binary(Data) ->
            decode_data(Data, State);
        {hackney_response, ClientRef, Error} ->
            ets:delete(couchbeam_view_streams, StreamRef),<commit_msg>do not exit on do_init_stream<commit_after>%%% -*- erlang -*-
%%%
%%% This file is part of couchbeam released under the MIT license.
%%% See the NOTICE for more information.

-module(couchbeam_view_stream).

-export([start_link/4]).

-export([init_stream/5,
         maybe_continue/1,
         system_continue/3,
         system_terminate/4,
         system_code_change/4]).


-export([init/1,
         handle_event/2,
         wait_rows/2,
         wait_rows1/2,
         wait_val/2,
         collect_object/2,
         maybe_continue_decoding/1]).


-include("couchbeam.hrl").

-record(state, {parent,
                owner,
                req,
                ref,
                mref,
                client_ref=nil,
                decoder,
                async=normal}).

-record(viewst, {parent,
                 owner,
                 ref,
                 mref,
                 client_ref,
                 async=false}).


-define(TIMEOUT, 10000).
-define(DEFAULT_CHANGES_POLL, 5000). % we check every 5secs



start_link(Owner, StreamRef, {Db, Url, Args}, StreamOptions) ->
    proc_lib:start_link(?MODULE, init_stream, [self(), Owner, StreamRef,
                                               {Db, Url, Args},
                                               StreamOptions]).

init_stream(Parent, Owner, StreamRef, {_Db, _Url, _Args}=Req,
            StreamOptions) ->


    Async = proplists:get_value(async, StreamOptions, normal),

    %% monitor the process receiving§ the messages
    MRef = erlang:monitor(process, Owner),

    InitState = #state{parent=Parent,
                       owner=Owner,
                       req=Req,
                       ref=StreamRef,
                       mref=MRef,
                       async=Async},

    %% connect to the view
    case do_init_stream(Req, InitState) of
        {ok, State} ->
            %% register the stream
            ets:insert(couchbeam_view_streams, [{StreamRef, self()}]),
            %% tell to the parent that we are ok
            proc_lib:init_ack(Parent, {ok, self()}),
            %% start the loop
            loop(State);
        Error ->
            proc_lib:init_ack(Parent, Error)
    end,
    %% stop to monitor the parent
    erlang:demonitor(MRef),
    ok.

do_init_stream({#db{options=Opts}, Url, Args}, #state{mref=MRef}=State) ->
    %% we are doing the request asynchronously
    FinalOpts = [{async, once} | Opts],
    Reply = case Args#view_query_args.method of
        get ->
            hackney:request(get, Url, [], <<>>, FinalOpts);
        post ->
            Body = couchbeam_ejson:encode({[{<<"keys">>,
                                             Args#view_query_args.keys}]}),
            Headers = [{<<"Content-Type">>, <<"application/json">>}],
            hackney:request(post, Url, Headers, Body, FinalOpts)
    end,

    case Reply of
        {ok, Ref} ->
            receive
                {'DOWN', MRef, _, _, _} ->
                    %% parent exited there is no need to continue
                    exit(normal);
                {hackney_response, Ref, {status, 200, _}} ->
                    #state{parent=Parent,
                           owner=Owner,
                           ref=StreamRef,
                           async=Async} = State,

                    DecoderFun = jsx:decoder(?MODULE, [Parent, Owner,
                                                       StreamRef, MRef, Ref,
                                                       Async], [stream]),
                    {ok, State#state{client_ref=Ref,
                                     decoder=DecoderFun}};

                {hackney_response, Ref, {status, 404, _}} ->
                    {error, not_found};
                {hackney_response, Ref, {status, Status, Reason}} ->
                    {error, {http_error, Status, Reason}};
                {hackney_response, Ref, {error, Reason}} ->
                    {error, Reason}
            after ?TIMEOUT ->
                    {error, timeout}
            end;
        Error ->
            {error, Error}
    end.



loop(#state{owner=Owner,
            ref=StreamRef,
            mref=MRef,
            client_ref=ClientRef}=State) ->

    hackney:stream_next(ClientRef),
    receive
        {'DOWN', MRef, _, _, _} ->
            %% parent exited there is no need to continue
            exit(normal);
        {hackney_response, ClientRef, {headers, _Headers}} ->
            loop(State);
        {hackney_response, ClientRef, done} ->
            %% unregister the stream
            ets:delete(couchbeam_view_streams, StreamRef),
            %% tell to the owner that we are done and exit,
            Owner ! {StreamRef, done};
        {hackney_response, ClientRef, Data} when is_binary(Data) ->
            decode_data(Data, State);
        {hackney_response, ClientRef, Error} ->
            ets:delete(couchbeam_view_streams, StreamRef),
{'fraction_non_alphanumeric': 0.11408709875281128, 'fraction_numerical': 0.006133714986710284, 'mean_word_length': 1.4673392181588902, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1529247', 'n_tokens_mistral': 2822, 'n_tokens_neox': 2644, 'n_words': 601}