<commit_before>%% Copyright 2014 Erlio GmbH Basel Switzerland (http://erl.io)
%%
%% 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(vmq_session_proxy).
-include("vmq_server.hrl").

-behaviour(gen_server).

%% API
-export([start_link/3,
         derefed/2,
         deliver/2]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {node,
                repl_pid,
                repl_mon,
                queue_pid,
                queue_mon,
                subscriber_id,
                waiting}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Node, QPid, SubscriberId) ->
    gen_server:start_link(?MODULE, [Node, QPid, SubscriberId], []).

derefed(SessionProxy, MsgRef) ->
    gen_server:cast(SessionProxy, {derefed, MsgRef}).

deliver(SessionProxy, Term) ->
    gen_server:call(SessionProxy, {deliver, Term}, infinity).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Node, QPid, SubscriberId]) ->
    QueueMon = monitor(process, QPid),
    {ok, #state{node=Node, subscriber_id=SubscriberId, queue_pid=QPid,
                queue_mon=QueueMon}, 0}.

handle_call({deliver, Term}, From, State) ->
    {noreply, deliver(From, Term, State)}.

handle_cast({derefed, MsgRef}, #state{waiting={MsgRef, From}} = State) ->
    gen_server:reply(From, ok),
    {noreply, State#state{waiting=undefined}}.

handle_info({'DOWN', _MRef, process, Pid, Reason}, State) ->
    #state{repl_pid=ReplPid, subscriber_id=SubscriberId} = State,
    case {Pid == ReplPid, Reason} of
        {true, normal} ->
            %% finished replicating
            {stop, normal, State};
        {true, OtherReason} ->
            lager:warning("replication process for subscriber ~p died due to ~p",
                         [SubscriberId, OtherReason]),
            {stop, OtherReason, State};
        {false, Reason} ->
            %% session stopped during replication
            {stop, Reason, State}
    end;

handle_info(timeout, #state{node=Node, subscriber_id=SubscriberId}= State) ->
    Self = self(),
    {ReplPid, ReplMon} =
    case node() of
        Node ->
            spawn_monitor(vmq_msg_store, deliver_from_store,
                          [SubscriberId, Self]);
        _ ->
            spawn_monitor(
              fun() ->
                      rpc:call(Node, vmq_msg_store, deliver_from_store,
                               [SubscriberId, Self])
              end)
    end,
    {noreply, State#state{repl_pid=ReplPid, repl_mon=ReplMon}}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.<commit_msg>improved message delivery from remote store<commit_after>%% Copyright 2014 Erlio GmbH Basel Switzerland (http://erl.io)
%%
%% 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(vmq_session_proxy).
-include("vmq_server.hrl").

-behaviour(gen_server).

%% API
-export([start_link/3,
         derefed/2,
         deliver/2]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-record(state, {node,
                repl_pid,
                repl_mon,
                queue_pid,
                queue_mon,
                subscriber_id,
                waiting}).

-define(RETRY, 10000). % wait 10 seconds to retry

%%%===================================================================
%%% API
%%%===================================================================

start_link(Node, QPid, SubscriberId) ->
    gen_server:start_link(?MODULE, [Node, QPid, SubscriberId], []).

derefed(SessionProxy, MsgRef) ->
    gen_server:cast(SessionProxy, {derefed, MsgRef}).

deliver(SessionProxy, Term) ->
    gen_server:call(SessionProxy, {deliver, Term}, infinity).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init([Node, QPid, SubscriberId]) ->
    QueueMon = monitor(process, QPid),
    self() ! deliver_from_store, % kick of delivery
    {ok, #state{node=Node, subscriber_id=SubscriberId, queue_pid=QPid,
                queue_mon=QueueMon}}.

handle_call({deliver, Term}, From, State) ->
    {noreply, deliver(From, Term, State)}.

handle_cast({derefed, MsgRef}, #state{waiting={MsgRef, From}} = State) ->
    gen_server:reply(From, ok),
    {noreply, State#state{waiting=undefined}}.

handle_info({'DOWN', _MRef, process, Pid, Reason}, State) ->
    #state{repl_pid=ReplPid, subscriber_id=SubscriberId, node=Node} = State,
    case {Pid == ReplPid, Reason} of
        {true, normal} ->
            %% finished replicating
            {stop, normal, State};
        {true, {deliver_from_store, retry}} ->
            case lists:member(Node, vmq_cluster:nodes()) of
                true ->
                    erlang:send_after(?RETRY, self(), deliver_from_store),
                    {noreply, State};
                false ->
                    {stop, normal, State}
            end;
        {true, OtherReason} ->
            lager:warning("replication process for subscriber ~p died due to ~p",
                         [SubscriberId, OtherReason]),
            {stop, OtherReason, State};
        {false, Reason} ->
            %% session stopped during replication
            {stop, Reason, State}
    end;

handle_info(deliver_from_store, #state{node=Node,
                                       subscriber_id=SubscriberId} = State) ->
    Self = self(),
    {ReplPid, ReplMon} =
    case node() of
        Node ->
            spawn_monitor(vmq_msg_store, deliver_from_store,
                          [SubscriberId, Self]);
        _ ->
            spawn_monitor(
              fun() ->
                      case rpc:call(Node, vmq_msg_store, deliver_from_store,
                                    [SubscriberId, Self]) of
                          {badrpc, timeout} ->
                              % maybe netsplit
                              exit({deliver_from_store, retry});
                          R -> R
                      end
              end)
    end,
    {noreply, State#state{repl_pid=ReplPid, repl_mon=ReplMon}}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.
{'fraction_non_alphanumeric': 0.1853215835854268, 'fraction_numerical': 0.005524135209785611, 'mean_word_length': 2.551611396543671, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1847690', 'n_tokens_mistral': 2252, 'n_tokens_neox': 2135, 'n_words': 529}