<commit_before>-module(twitterlinks_stream_listener).
-behaviour(gen_server).
-define(SERVER, ?MODULE).

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.


%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([start_link/2, start_stream/1]).
-record(state, {middleman, url, request_id, new_tweet_callback}).

%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------

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

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

start_link(MiddlemanPid, Url) ->
    gen_server:start_link(?MODULE, [MiddlemanPid, Url], []).

%% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------

init([MiddlemanPid, Url]) ->
    NewTweetCallback = fun(TweetJSONBin) ->
                               twitterlinks_tweet_middleman:new_tweet(MiddlemanPid,
                                                                      TweetJSONBin)
                       end,
    {ok, ReqId} = start_stream(Url),
    {ok, #state{url=Url, request_id=ReqId, new_tweet_callback=NewTweetCallback}}.

handle_call(_Msg, _From, State) ->
    {noreply, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info({http, {_ReqId, stream_start, _Headers}}, State) ->
    io:format("stream listener: stream started~n", []),
    {noreply, State};
handle_info({http, {_, stream, <<"\r\n">>}}, State) ->
    % Twitter Heartbeat
    {noreply, State};
handle_info({http, {_, stream, BinBodyPart}}, State=#state{new_tweet_callback=NewTweetCallback}) ->
    io:format("stream listener: new tweet~n", []),
    NewTweetCallback(BinBodyPart),
    {noreply, State};
handle_info({http, {_, stream_end, _}}, State) ->
    % TODO: start the stream again
    {stop, stream_end, State};
handle_info(Msg, State) ->
    io:format("Unknown Message: ~w", [Msg]),
    {noreply, State}.
<commit_msg>Whoops, I didn't change the module name in the callback<commit_after>-module(twitterlinks_stream_listener).
-behaviour(gen_server).
-define(SERVER, ?MODULE).

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.


%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([start_link/2, start_stream/1]).
-record(state, {middleman, url, request_id, new_tweet_callback}).

%% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------

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

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

start_link(MiddlemanPid, Url) ->
    gen_server:start_link(?MODULE, [MiddlemanPid, Url], []).

%% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------

init([MiddlemanPid, Url]) ->
    NewTweetCallback = fun(TweetJSONBin) ->
                               twitterlinks_middleman:new_tweet(MiddlemanPid,
                                                                      TweetJSONBin)
                       end,
    {ok, ReqId} = start_stream(Url),
    {ok, #state{url=Url, request_id=ReqId, new_tweet_callback=NewTweetCallback}}.

handle_call(_Msg, _From, State) ->
    {noreply, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info({http, {_ReqId, stream_start, _Headers}}, State) ->
    io:format("stream listener: stream started~n", []),
    {noreply, State};
handle_info({http, {_, stream, <<"\r\n">>}}, State) ->
    % Twitter Heartbeat
    {noreply, State};
handle_info({http, {_, stream, BinBodyPart}}, State=#state{new_tweet_callback=NewTweetCallback}) ->
    io:format("stream listener: new tweet~n", []),
    NewTweetCallback(BinBodyPart),
    {noreply, State};
handle_info({http, {_, stream_end, _}}, State) ->
    % TODO: start the stream again
    {stop, stream_end, State};
handle_info(Msg, State) ->
    io:format("Unknown Message: ~w", [Msg]),
    {noreply, State}.

{'fraction_non_alphanumeric': 0.3491048593350384, 'fraction_numerical': 0.0034100596760443308, 'mean_word_length': 1.4883351007423118, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3088295', 'n_tokens_mistral': 1263, 'n_tokens_neox': 1211, 'n_words': 227}