<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2011 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

%% @doc Module to manage storage of objects and files

-module(riak_moss_put_fsm).

-behaviour(gen_fsm).

-include("riak_moss.hrl").

%% API
-export([start_link/10,
         augment_data/2,
         block_written/2,
         finalize/1]).

%% gen_fsm callbacks
-export([init/1,
         prepare/2,
         prepare/3,
         not_full/2,
         full/2,
         all_received/2,
         not_full/3,
         all_received/3,
         done/3,
         handle_event/3,
         handle_sync_event/4,
         handle_info/3,
         terminate/3,
         code_change/4]).

-define(SERVER, ?MODULE).
-define(EMPTYORDSET, ordsets:new()).

-record(state, {timeout :: pos_integer(),
                block_size :: pos_integer(),
                caller :: pid(),
                md5 :: binary(),
                reply_pid :: pid(),
                mani_pid :: pid(),
                riakc_pid :: pid(),
                timer_ref :: term(),
                bucket :: binary(),
                key :: binary(),
                metadata :: term(),
                acl :: acl(),
                manifest :: lfs_manifest(),
                content_length :: pos_integer(),
                content_type :: binary(),
                num_bytes_received=0,
                max_buffer_size :: non_neg_integer(),
                current_buffer_size=0,
                buffer_queue=[], %% not actually a queue, but we treat it like one
                remainder_data :: undefined | binary(),
                free_writers :: ordsets:new(),
                unacked_writes=ordsets:new(),
                next_block_id=0,
                all_writer_pids :: list(pid())}).

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

%%--------------------------------------------------------------------
%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Bucket,
           Key,
           ContentLength,
           ContentType,
           Metadata,
           BlockSize,
           Acl,
           Timeout,
           Caller,
           RiakPid) ->
    Args = [Bucket, Key, ContentLength, ContentType,
            Metadata, BlockSize, Acl, Timeout, Caller, RiakPid],
    gen_fsm:start_link(?MODULE, Args, []).

augment_data(Pid, Data) ->
    gen_fsm:sync_send_event(Pid, {augment_data, Data}, infinity).

finalize(Pid) ->
    gen_fsm:sync_send_event(Pid, finalize, infinity).

block_written(Pid, BlockID) ->
    gen_fsm:send_event(Pid, {block_written, BlockID, self()}).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%%
%%--------------------------------------------------------------------

%% TODO:
%% Metadata support is a future feature,
%% but I'm just stubbing it in here
%% so that I can be thinking about how it
%% might be implemented. Does it actually
%% make things more confusing?
init([Bucket, Key, ContentLength, ContentType,
      Metadata, BlockSize, Acl, Timeout, Caller, RiakPid]) ->
    %% We need to do this (the monitor) for two reasons
    %% 1. We're started through a supervisor, so the
    %%    proc that actually intends to start us isn't
    %%    linked to us.
    %% 2. Even if we didn't use a supervisor, the webmachine
    %%    process uses exit(..., normal), even on abnormal
    %%    terminations, so this process would still
    %%    live.
    CallerRef = erlang:monitor(process, Caller),

    {ok, prepare, #state{bucket=Bucket,
                         key=Key,
                         block_size=BlockSize,
                         caller=CallerRef,
                         metadata=Metadata,
                         acl=Acl,
                         content_length=ContentLength,
                         content_type=ContentType,
                         riakc_pid=RiakPid,
                         timeout=Timeout},
                     0}.
%% init([test, Bucket, Key, ContentLength, BlockSize]) ->
%%     {ok, prepare, State1, 0} = init([Bucket, Key, self(), pid]),

%%     %% %% purposely have the timeout happen
%%     %% %% so that we get called in the prepare
%%     %% %% state
%%     %% {ok, ReaderPid} =
%%     %%     riak_moss_dummy_reader:start_link([self(),
%%     %%                                        Bucket,
%%     %%                                        Key,
%%     %%                                        ContentLength,
%%     %%                                        BlockSize]),
%%     %% link(ReaderPid),
%%     {ok, Manifest} = riak_moss_dummy_reader:get_manifest(ReaderPid),
%%     {ok, waiting_value, State1#state{free_readers=[ReaderPid],
%%                                      manifest=Manifest,
%%                                      test=true}}.

%%--------------------------------------------------------------------
%%
%%--------------------------------------------------------------------
prepare(timeout, State=#state{content_length=0}) ->
    NewState = prepare(State),
    Md5 = crypto:md5_final(NewState#state.md5),
    NewManifest = NewState#state.manifest#lfs_manifest_v2{content_md5=Md5,
                                                          state=active,
                                                          last_block_written_time=erlang:now()},
    {next_state, done, NewState#state{md5=Md5, manifest=NewManifest}};
prepare(timeout, State) ->
    NewState = prepare(State),
    {next_state, not_full, NewState}.

prepare(finalize, From, State=#state{content_length=0}) ->
    NewState = prepare(State),
    Md5 = crypto:md5_final(NewState#state.md5),
    NewManifest = NewState#state.manifest#lfs_manifest_v2{content_md5=Md5,
                                                          state=active,
                                                          last_block_written_time=erlang:now()},
    done(finalize, From, NewState#state{md5=Md5, manifest=NewManifest});
prepare({augment_data, NewData}, From, State) ->
    #state{content_length=CLength,<commit_msg>Remove commented-out init garbage in put_fsm.<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2011 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

%% @doc Module to manage storage of objects and files

-module(riak_moss_put_fsm).

-behaviour(gen_fsm).

-include("riak_moss.hrl").

%% API
-export([start_link/10,
         augment_data/2,
         block_written/2,
         finalize/1]).

%% gen_fsm callbacks
-export([init/1,
         prepare/2,
         prepare/3,
         not_full/2,
         full/2,
         all_received/2,
         not_full/3,
         all_received/3,
         done/3,
         handle_event/3,
         handle_sync_event/4,
         handle_info/3,
         terminate/3,
         code_change/4]).

-define(SERVER, ?MODULE).
-define(EMPTYORDSET, ordsets:new()).

-record(state, {timeout :: pos_integer(),
                block_size :: pos_integer(),
                caller :: pid(),
                md5 :: binary(),
                reply_pid :: pid(),
                mani_pid :: pid(),
                riakc_pid :: pid(),
                timer_ref :: term(),
                bucket :: binary(),
                key :: binary(),
                metadata :: term(),
                acl :: acl(),
                manifest :: lfs_manifest(),
                content_length :: pos_integer(),
                content_type :: binary(),
                num_bytes_received=0,
                max_buffer_size :: non_neg_integer(),
                current_buffer_size=0,
                buffer_queue=[], %% not actually a queue, but we treat it like one
                remainder_data :: undefined | binary(),
                free_writers :: ordsets:new(),
                unacked_writes=ordsets:new(),
                next_block_id=0,
                all_writer_pids :: list(pid())}).

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

%%--------------------------------------------------------------------
%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Bucket,
           Key,
           ContentLength,
           ContentType,
           Metadata,
           BlockSize,
           Acl,
           Timeout,
           Caller,
           RiakPid) ->
    Args = [Bucket, Key, ContentLength, ContentType,
            Metadata, BlockSize, Acl, Timeout, Caller, RiakPid],
    gen_fsm:start_link(?MODULE, Args, []).

augment_data(Pid, Data) ->
    gen_fsm:sync_send_event(Pid, {augment_data, Data}, infinity).

finalize(Pid) ->
    gen_fsm:sync_send_event(Pid, finalize, infinity).

block_written(Pid, BlockID) ->
    gen_fsm:send_event(Pid, {block_written, BlockID, self()}).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%%
%%--------------------------------------------------------------------

%% TODO:
%% Metadata support is a future feature,
%% but I'm just stubbing it in here
%% so that I can be thinking about how it
%% might be implemented. Does it actually
%% make things more confusing?
init([Bucket, Key, ContentLength, ContentType,
      Metadata, BlockSize, Acl, Timeout, Caller, RiakPid]) ->
    %% We need to do this (the monitor) for two reasons
    %% 1. We're started through a supervisor, so the
    %%    proc that actually intends to start us isn't
    %%    linked to us.
    %% 2. Even if we didn't use a supervisor, the webmachine
    %%    process uses exit(..., normal), even on abnormal
    %%    terminations, so this process would still
    %%    live.
    CallerRef = erlang:monitor(process, Caller),

    {ok, prepare, #state{bucket=Bucket,
                         key=Key,
                         block_size=BlockSize,
                         caller=CallerRef,
                         metadata=Metadata,
                         acl=Acl,
                         content_length=ContentLength,
                         content_type=ContentType,
                         riakc_pid=RiakPid,
                         timeout=Timeout},
                     0}.

%%--------------------------------------------------------------------
%%
%%--------------------------------------------------------------------
prepare(timeout, State=#state{content_length=0}) ->
    NewState = prepare(State),
    Md5 = crypto:md5_final(NewState#state.md5),
    NewManifest = NewState#state.manifest#lfs_manifest_v2{content_md5=Md5,
                                                          state=active,
                                                          last_block_written_time=erlang:now()},
    {next_state, done, NewState#state{md5=Md5, manifest=NewManifest}};
prepare(timeout, State) ->
    NewState = prepare(State),
    {next_state, not_full, NewState}.

prepare(finalize, From, State=#state{content_length=0}) ->
    NewState = prepare(State),
    Md5 = crypto:md5_final(NewState#state.md5),
    NewManifest = NewState#state.manifest#lfs_manifest_v2{content_md5=Md5,
                                                          state=active,
                                                          last_block_written_time=erlang:now()},
    done(finalize, From, NewState#state{md5=Md5, manifest=NewManifest});
prepare({augment_data, NewData}, From, State) ->
    #state{content_length=CLength,
{'fraction_non_alphanumeric': 0.23886506103596172, 'fraction_numerical': 0.00882547014186737, 'mean_word_length': 1.4984545641871008, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5364044', 'n_tokens_mistral': 3240, 'n_tokens_neox': 3050, 'n_words': 657}