<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_topic_vnode).

-behaviour(riak_core_vnode).

-include_lib("riak_core/include/riak_core_vnode.hrl").
-include("harbinger.hrl").

%% API
-export([publish/2,
         bindings/1,
         bind/2,
         unbind/2]).

%% Callbacks
-export([start_vnode/1,
         init/1,
         handle_command/3,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_exit/3,
         is_empty/1,
         delete/1,
         terminate/2]).

%%
%% Types
%%

-record(s, {partition,
            head = 0                   :: non_neg_integer(),
            tail = 0                   :: non_neg_integer(),
            bindings = gb_sets:empty() :: gb_set()}).

-define(TIMEOUT, 5000).

%%
%% API
%%

-spec publish(binary(), binary()) -> ok.
%% @doc
publish(_Topic, _Payload) ->
    %% Write the payload to the topic and then the vnodes call
    %% queue:handle_publish offset reply
    ok.

-spec bind(binary(), binary()) -> {ok, non_neg_integer(), non_neg_integer()}.
%% @doc
bind(Topic, Queue) ->
    harbinger_coordinator:reply({Topic, Topic},
                                harbinger_topic,
                                fun(R) -> {bind, R, {queue, Queue}} end,
                                ?TOPIC_MASTER).

-spec unbind(binary(), binary()) -> ok.
%% @doc
unbind(Topic, Queue) ->
    harbinger_coordinator:noreply({Topic, Topic},
                               harbinger_topic,
                               fun(R) -> {unbind, R, {queue, Queue}} end,
                               ?TOPIC_MASTER).

bindings(Topic) ->
    harbinger_coordinator:reply({Topic, Topic},
                                harbinger_topic,
                                fun(R) -> {bindings, R} end,
                                ?TOPIC_MASTER).

%%
%% Callbacks
%%

start_vnode(Index) -> riak_core_vnode_master:get_vnode_pid(Index, ?MODULE).

init([Partition]) -> {ok, #s{partition = Partition}}.

handle_command({bind, ReqId, Binding}, _Sender, State = #s{bindings = Bindings}) ->
    NewState = State#s{bindings = gb_sets:add(Binding, Bindings)},
    {reply, {ok, ReqId, {State#s.head, State#s.tail}}, NewState};
handle_command({unbind, ReqId, Binding}, _Sender, State = #s{bindings = Bindings}) ->
    NewState = State#s{bindings = gb_sets:delete_any(Binding, Bindings)},
    {reply, {ok, ReqId}, NewState};
handle_command({bindings, ReqId}, _Sender, State) ->
    {reply, {ok, ReqId, gb_sets:to_list(State#s.bindings)}, State};
handle_command(ping, _Sender, State) ->
    {reply, {pong, State#s.partition}, State}.

handle_handoff_command(_Message, _Sender, State) ->
    {noreply, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

handle_handoff_data(_Data, State) ->
    {reply, ok, State}.

encode_handoff_item(_ObjectName, _ObjectValue) ->
    <<>>.

handle_coverage(_Req, _KeySpaces, _Sender, State) ->
    {stop, not_implemented, State}.

handle_exit(_Pid, _Reason, State) ->
    {noreply, State}.

is_empty(State) ->
    {true, State}.

delete(State) ->
    {ok, State}.

terminate(_Reason, _State) ->
    ok.
<commit_msg>Adding basic topic vnode 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_topic_vnode).

-behaviour(riak_core_vnode).

-include_lib("riak_core/include/riak_core_vnode.hrl").
-include("harbinger.hrl").

%% API
-export([publish/2,
         bindings/1,
         bind/2,
         unbind/2]).

%% Callbacks
-export([start_vnode/1,
         init/1,
         handle_command/3,
         handle_handoff_command/3,
         handoff_starting/2,
         handoff_cancelled/1,
         handoff_finished/2,
         handle_handoff_data/2,
         encode_handoff_item/2,
         handle_coverage/4,
         handle_exit/3,
         is_empty/1,
         delete/1,
         terminate/2]).

%%
%% Types
%%

-record(s, {partition = 0               :: non_neg_integer(),
            head      = 0               :: non_neg_integer(),
            tail      = 0               :: non_neg_integer(),
            bindings  = gb_sets:empty() :: gb_set(),
            log       = []              :: [binary()]}).

-define(TIMEOUT, 5000).

%%
%% API
%%

-spec publish(binary(), binary()) -> ok.
%% @doc
publish(Topic, Payload) ->
    harbinger_coordinator:noreply({Topic, Topic},
                               harbinger_topic,
                               fun(R) -> {publish, R, Payload} end,
                               ?TOPIC_MASTER).

-spec bind(binary(), binary()) -> {ok, non_neg_integer(), non_neg_integer()}.
%% @doc
bind(Topic, Queue) ->
    harbinger_coordinator:reply({Topic, Topic},
                                harbinger_topic,
                                fun(R) -> {bind, R, {queue, Queue}} end,
                                ?TOPIC_MASTER).

-spec unbind(binary(), binary()) -> ok.
%% @doc
unbind(Topic, Queue) ->
    harbinger_coordinator:noreply({Topic, Topic},
                               harbinger_topic,
                               fun(R) -> {unbind, R, {queue, Queue}} end,
                               ?TOPIC_MASTER).

-spec bindings(binary()) -> {ok, [{queue, binary()}]}.
%% @doc
bindings(Topic) ->
    harbinger_coordinator:reply({Topic, Topic},
                                harbinger_topic,
                                fun(R) -> {bindings, R} end,
                                ?TOPIC_MASTER).

%%
%% Callbacks
%%

start_vnode(Index) -> riak_core_vnode_master:get_vnode_pid(Index, ?MODULE).

init([Partition]) -> {ok, #s{partition = Partition}}.

%% Publish
handle_command({publish, ReqId, Payload}, _Sender, State = #s{tail = Tail, log = Log}) ->
    NewState = State#s{log = [Payload|Log], tail = Tail + 1},
    {reply, {ok, ReqId}, NewState};

%% Bind
handle_command({bind, ReqId, Binding}, _Sender, State = #s{bindings = Bindings}) ->
    NewState = State#s{bindings = gb_sets:add(Binding, Bindings)},
    {reply, {ok, ReqId, {State#s.head, State#s.tail}}, NewState};

%% Unbind
handle_command({unbind, ReqId, Binding}, _Sender, State = #s{bindings = Bindings}) ->
    NewState = State#s{bindings = gb_sets:delete_any(Binding, Bindings)},
    {reply, {ok, ReqId}, NewState};

%% Bindings
handle_command({bindings, ReqId}, _Sender, State) ->
    {reply, {ok, ReqId, gb_sets:to_list(State#s.bindings)}, State};

%% Ping
handle_command(ping, _Sender, State) ->
    {reply, {pong, State#s.partition}, State}.

handle_handoff_command(_Message, _Sender, State) ->
    {noreply, State}.

handoff_starting(_TargetNode, State) ->
    {true, State}.

handoff_cancelled(State) ->
    {ok, State}.

handoff_finished(_TargetNode, State) ->
    {ok, State}.

handle_handoff_data(_Data, State) ->
    {reply, ok, State}.

encode_handoff_item(_ObjectName, _ObjectValue) ->
    <<>>.

handle_coverage(_Req, _KeySpaces, _Sender, State) ->
    {stop, not_implemented, State}.

handle_exit(_Pid, _Reason, State) ->
    {noreply, State}.

is_empty(State) ->
    {true, State}.

delete(State) ->
    {ok, State}.

terminate(_Reason, _State) ->
    ok.

{'fraction_non_alphanumeric': 0.15356773526370218, 'fraction_numerical': 0.008531540847983453, 'mean_word_length': 2.384514435695538, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 72, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2879916', 'n_tokens_mistral': 2679, 'n_tokens_neox': 2608, 'n_words': 532}