<commit_before>%% This Source Code Form is subject to the terms of
%% the Mozilla Public License, v. 2.0.
%% A copy of the MPL can be found in the LICENSE file or
%% you can obtain it at http://mozilla.org/MPL/2.0/.
%%
%% @author <NAME>
%% @copyright (c) 2012 <NAME> <<EMAIL>>
%% @doc
%%

-module(harbinger_processor).

-behaviour(gen_server).

-include("restomp.hrl").
-include("harbinger.hrl").

%% API
-export([start_link/2,
         process/2,
         stop/1]).

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

%%
%% Types
%%

-record(s, {sup      :: pid(),
            sock     :: inet:socket(),
            session  :: string()}).

-define(VERSIONS,       ["1.0", "1.1"]).
-define(VERSION_HEADER, {"version", string:join(?VERSIONS, ",")}).
-define(SERVER_HEADER,  {"server", "Harbinger/0.1.0"}).
-define(CONTENT_TEXT,   {"content-type", "text/plain"}).

%%
%% API
%%

-spec start_link(pid(), inet:socket()) -> {ok, pid()} | ignore | {error, _}.
%% @doc
start_link(SupPid, Sock) -> gen_server:start_link(?MODULE, {SupPid, Sock}, []).

-spec process(pid(), #stomp_frame{}) -> ok.
%% @doc
process(Pid, Frame) -> gen_server:cast(Pid, Frame).

-spec stop(pid()) -> ok.
%% @doc
stop(Pid) -> gen_server:cast(Pid, stop).

%%
%% Callbacks
%%

-spec init({pid(), inet:socket()}) -> {ok, #s{}}.
%% @hidden
init({SupPid, Sock}) ->
    process_flag(trap_exit, true),
    {ok, #s{sup = SupPid, sock = Sock}}.

-spec handle_call(_, {pid(), _}, #s{}) -> {stop, {unhandled_call, _}, #s{}}.
%% @hidden
handle_call(Msg, _From, State) -> {stop, {unhandled_call, Msg}, State}.

-spec handle_cast(stop | _, #s{}) -> {stop, normal, #s{}}.
%% @hidden
handle_cast(Frame = #stomp_frame{command = Cmd}, State) ->
    case Cmd of
        "STOMP"       -> connect(Frame, State);
        "CONNECT"     -> connect(Frame, State);
        "SEND"        -> send(Frame, State);
        "SUBSCRIBE"   -> subscribe(Frame, State);
        "UNSUBSCRIBE" -> unsubscribe(Frame, State);
        "ACK"         -> ack(Frame, State);
        "NACK"        -> nack(Frame, State);
        "BEGIN"       -> txbegin(State);
        "COMMIT"      -> txcommit(State);
        "ABORT"       -> txabort(State);
        "DISCONNECT"  -> disconnect(State);
        _Unknown      -> unsupported(State)
    end;
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(Msg, State) ->
    {stop, {unhandled_cast, Msg}, State}.

-spec handle_info(_, #s{}) -> {stop, {unhandled_call, _}, #s{}}.
%% @hidden
handle_info(Info, State) -> {stop, {unhandled_info, Info}, State}.

-spec terminate(_, #s{}) -> ok.
%% @hidden
terminate(_Reason, _State) -> ok.

-spec code_change(_, #s{}, _) -> {ok, #s{}}.
%% @hidden
code_change(_OldVsn, State, _Extra) -> {ok, State}.

%%
%% Handlers
%%

%% @private
connect(_Frame, State = #s{session = undefined}) ->
    Session = riak_core_util:unique_id_62(),
    Headers = [{"session", Session},
               ?SERVER_HEADER,
               ?VERSION_HEADER],
    send_frame("CONNECTED", Headers, "", State#s{session = Session});
connect(_Frame, State) ->
    send_error(already_connected, State).

%% @private
send(Frame = #stomp_frame{command = _Cmd, headers = _Headers, body = _Body},
     State) ->
    %% Check destination header exists
    case restomp:header(Frame, "destination") of
        {ok, Dest} ->
            case split_destination(Dest) of
                {topic, _Topic}  -> {noreply, State};
                _Other           -> send_error(invalid_destination, State)
            end;
        not_found ->
            send_error(no_destination, State)
    end.

%% @private
subscribe(Frame, State) ->
    Fun = fun(H) -> restomp:header(Frame, H) end,
    case {Fun("ack"), Fun("id"), Fun("destination")} of
        {{ok, "client"}, {ok, _Id}, {ok, Dest}} ->
            case split_destination(Dest) of
                {queue, _Topic, _Queue} -> {noreply, State};
                _Other                  -> send_error(invalid_destination, State)
            end;
        {_Ack, not_found, _Dest} ->
            send_error(no_id, State);
        {_Ack, _Id, _Dest} ->
            send_error(must_set_ack_client, State)
    end.

%% @private
unsubscribe(Frame, State) ->
    case restomp:header(Frame, "id") of
        {ok, _Id} ->
            {noreply, State};
        not_found ->
            send_error(no_id, State)
    end.

%% @private
ack(_Frame, State) -> send_error(not_implemented, State).

%% @private
nack(_Frame, State) -> send_error(not_implemented, State).

%% @private
txbegin(State) -> send_error(not_implemented, State).

%% @private
txcommit(State) -> send_error(not_implemented, State).

%% @private
txabort(State) -> send_error(not_implemented, State).

%% @private
disconnect(State) -> {stop, normal, State}.

%% @private
unsupported(State) -> send_error(unsupported_command, State).
<commit_msg>Finish initial skeletons for supported STOMP commands<commit_after>%% This Source Code Form is subject to the terms of
%% the Mozilla Public License, v. 2.0.
%% A copy of the MPL can be found in the LICENSE file or
%% you can obtain it at http://mozilla.org/MPL/2.0/.
%%
%% @author <NAME>
%% @copyright (c) 2012 <NAME> <<EMAIL>>
%% @doc
%%

-module(harbinger_processor).

-behaviour(gen_server).

-include("restomp.hrl").
-include("harbinger.hrl").

%% API
-export([start_link/2,
         process/2,
         stop/1]).

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

%%
%% Types
%%

-record(s, {sup      :: pid(),
            sock     :: inet:socket(),
            session  :: string()}).

-define(VERSIONS,       ["1.0", "1.1"]).
-define(VERSION_HEADER, {"version", string:join(?VERSIONS, ",")}).
-define(SERVER_HEADER,  {"server", "Harbinger/0.1.0"}).
-define(CONTENT_TEXT,   {"content-type", "text/plain"}).

%%
%% API
%%

-spec start_link(pid(), inet:socket()) -> {ok, pid()} | ignore | {error, _}.
%% @doc
start_link(SupPid, Sock) -> gen_server:start_link(?MODULE, {SupPid, Sock}, []).

-spec process(pid(), #stomp_frame{}) -> ok.
%% @doc
process(Pid, Frame) -> gen_server:cast(Pid, Frame).

-spec stop(pid()) -> ok.
%% @doc
stop(Pid) -> gen_server:cast(Pid, stop).

%%
%% Callbacks
%%

-spec init({pid(), inet:socket()}) -> {ok, #s{}}.
%% @hidden
init({SupPid, Sock}) ->
    process_flag(trap_exit, true),
    {ok, #s{sup = SupPid, sock = Sock}}.

-spec handle_call(_, {pid(), _}, #s{}) -> {stop, {unhandled_call, _}, #s{}}.
%% @hidden
handle_call(Msg, _From, State) -> {stop, {unhandled_call, Msg}, State}.

-spec handle_cast(stop | _, #s{}) -> {stop, normal, #s{}}.
%% @hidden
handle_cast(Frame = #stomp_frame{command = Cmd}, State) ->
    case Cmd of
        "STOMP"       -> connect(Frame, State);
        "CONNECT"     -> connect(Frame, State);
        "SEND"        -> send(Frame, State);
        "SUBSCRIBE"   -> subscribe(Frame, State);
        "UNSUBSCRIBE" -> unsubscribe(Frame, State);
        "ACK"         -> ack(Frame, State);
        "NACK"        -> nack(Frame, State);
        "DISCONNECT"  -> disconnect(State);
        _Unknown      -> unsupported(State)
    end;
handle_cast(stop, State) ->
    {stop, normal, State};
handle_cast(Msg, State) ->
    {stop, {unhandled_cast, Msg}, State}.

-spec handle_info(_, #s{}) -> {stop, {unhandled_call, _}, #s{}}.
%% @hidden
handle_info(Info, State) -> {stop, {unhandled_info, Info}, State}.

-spec terminate(_, #s{}) -> ok.
%% @hidden
terminate(_Reason, _State) -> ok.

-spec code_change(_, #s{}, _) -> {ok, #s{}}.
%% @hidden
code_change(_OldVsn, State, _Extra) -> {ok, State}.

%%
%% Handlers
%%

%% @private
connect(_Frame, State = #s{session = undefined}) ->
    Session = riak_core_util:unique_id_62(),
    Headers = [{"session", Session},
               ?SERVER_HEADER,
               ?VERSION_HEADER],
    send_frame("CONNECTED", Headers, "", State#s{session = Session});
connect(_Frame, State) ->
    send_error(already_connected, State).

%% @private
send(Frame = #stomp_frame{command = _Cmd, headers = _Headers, body = _Body},
     State) ->
    %% Check destination header exists
    case restomp:header(Frame, "destination") of
        {ok, Dest} ->
            case split_destination(Dest) of
                {topic, _Topic}  -> {noreply, State};
                _Other           -> send_error(invalid_destination, State)
            end;
        not_found ->
            send_error(no_destination, State)
    end.

%% @private
subscribe(Frame, State) ->
    Fun = fun(H) -> restomp:header(Frame, H) end,
    case {Fun("ack"), Fun("id"), Fun("destination")} of
        {{ok, "client"}, {ok, _Id}, {ok, Dest}} ->
            case split_destination(Dest) of
                {queue, _Topic, _Queue} -> {noreply, State};
                _Other                  -> send_error(invalid_destination, State)
            end;
        {_Ack, not_found, _Dest} ->
            send_error(no_id, State);
        {_Ack, _Id, _Dest} ->
            send_error(must_set_ack_client, State)
    end.

%% @private
unsubscribe(Frame, State) ->
    case restomp:header(Frame, "id") of
        {ok, _Id} ->
            {noreply, State};
        not_found ->
            send_error(no_id, State)
    end.

%% @private
ack(Frame, State) ->
    Fun = fun(H) -> restomp:header(Frame, H) end,
    case {Fun("message-id"), Fun("subscription")} of
        {{ok, _Id}, {ok, _Sub}} ->
            {noreply, State};
        {not_found, _Sub} ->
            send_error(no_message_id, State);
        {_Id, not_found} ->
            send_error(no_subscription, State)
    end.

%% @private
nack(Frame, State) ->
    Fun = fun(H) -> restomp:header(Frame, H) end,
    case {Fun("message-id"), Fun("subscription")} of
        {{ok, _Id}, {ok, _Sub}} ->
            {noreply, State};
        {not_found, _Sub} ->
            send_error(no_message_id, State);
        {_Id, not_found} ->
            send_error(no_subscription, State)
    end.

%% @private
disconnect(State) -> {stop, normal, State}.

%% @private
unsupported(State) -> send_error(unsupported_command, State).

{'fraction_non_alphanumeric': 0.18520710059171597, 'fraction_numerical': 0.005128205128205128, 'mean_word_length': 2.7132918344928596, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 125, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4429248', 'n_tokens_mistral': 3552, 'n_tokens_neox': 3484, 'n_words': 776}