<commit_before>%%%-------------------------------------------------------------------
%%% File    : proxy_handler.erl
%%% Author  : <NAME>
%%% Description : 
%%%
%%% Created :  Tue Oct 13 20:24:21 PDT 2009
%%%-------------------------------------------------------------------

-module (proxy_handler).

-include ("common.hrl").
-include ("http.hrl").
-include ("beehive.hrl").

-export ([
  start_link/1
]).

-export ([terminate1/2]).

-record(state, {
  routing_key,          % subdomain of the app to look for
  host,                 % host
  port,                 % port
  bee,                  % Backend
  request,              % client request
  start_time,           % time proxy started
  client_socket,        % client socket
  server_socket,        % server socket
  client_pid,           % client listening pid
  server_pid,           % server listening pid
  timeout               % timeout
}).

% Start the proxy by spawning off an init with the socket.
% Compatible with the supervisor behaviour
start_link(ClientSock) ->
  Fun = fun() -> proxy_init(ClientSock) end,
  Pid = ?BENCHMARK_LOG("------Spawn-linking proxy_init------", erlang, spawn_link, [Fun]),
  {ok, Pid}.

% Receive the initial request and socket. Use the packet_decoder (http for now) to decode the
% routing key required. If this can be found, then start engaging the bee.
% If this takes more than IDLE_TIMEOUT seconds, then exit with an error.
proxy_init(ClientSock) ->
  receive
    {start, ClientSock, RequestPid} ->
      {ok, RoutingKey, ForwardReq, Req} = ?BENCHMARK_LOG("Handling request", http_request_decoder, handle_request, [ClientSock]),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key", bee_srv, get_bee, [self(), RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee);
    _E ->
      proxy_init(ClientSock)
  after ?IDLE_TIMEOUT ->
    % We only give 30 seconds to the proxy pid to be given control of the socket
    % this is MORE than enough time for the socket to be given a chance to prepare
    % to be handled by the proxy accept request.
    send_and_terminate(ClientSock, error, ?APP_ERROR(503, "Proxy is b0rked because of bee selection timeout in init"))
  end.

% Initiate and engage the chosen bee.
% First, we try to connect to the bee. If this succeeds, then let the rest of the app know that the
% bee has been engaged (nonblocking, through the event handler).
% Let the packet_decoder address the forwarding of the packet to the server and start the proxy loop
% If the bee cannot be reached, send an alert through the event handler that we could not reach
% the bee and try to find a new bee
engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, {ok, #bee{host = Host, port = Port} = Bee}) ->
  SockOpts = [binary, {active, false}, {packet, raw} ],
  case gen_tcp:connect(Host, Port, SockOpts, ?CONNECT_TIMEOUT) of
    {ok, ServerSock} ->
      ?NOTIFY({bee, used, Bee}),
      % Sending raw request to bee server
      gen_tcp:send(ServerSock, ForwardReq),
      
      Timeout = timer:seconds(10),
      ProxyPid = self(),
      
      State = #state{
                  start_time = date_util:now_to_seconds(),
                  client_socket = ClientSock, 
                  server_socket = ServerSock, 
                  request = Req,
                  routing_key = RoutingKey,
                  host = Host,
                  port = Port,
                  timeout = Timeout,
                  bee = Bee},
      
      ClientPid = spawn(fun() -> handle_streaming_data(client, ProxyPid, State) end),
      ServerPid = spawn(fun() -> handle_streaming_data(server, ProxyPid, State) end),
      
      NewState = State#state{server_pid = ServerPid, client_pid = ClientPid},
      
      proxy_loop(NewState);
    {error, emfile} ->
      ?LOG(error, "Maximum number of sockets open. Die instead", []),
      ?NOTIFY({bee, cannot_connect, Bee}),
      send_and_terminate(
        ClientSock, 503,
        ?APP_ERROR(503, "503 Service Unavailable")
      );
    {error,econnreset} ->
      timer:sleep(200),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key after a connection reset", bee_srv, get_bee, [RequestPid, RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee);
    Error ->
      ?LOG(error, "Connection to remote TCP server: ~p:~p ~p", [Host, Port, Error]),
      ?NOTIFY({bee, cannot_connect, Bee}),
      timer:sleep(200),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key after an error", bee_srv, get_bee, [RequestPid, RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee)
  end;
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, ?BACKEND_TIMEOUT_MSG) ->
  ?LOG(error, "Error getting bee because of timeout: ~p", [Hostname]),
  send_and_terminate(
    ClientSock, ?BACKEND_TIMEOUT_MSG, 
    ?APP_ERROR(404, io_lib:format("Error: ~p", [?BACKEND_TIMEOUT_MSG]))
  );
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, {error, Reason}) ->
  send_and_terminate(
    ClientSock, Reason, 
    ?APP_ERROR(404, io_lib:format("Error on ~p: ~p", [Hostname, Reason]))
  );
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, Else) ->
  send_and_terminate(
    ClientSock, Else, 
    ?APP_ERROR(404, io_lib:format("Error on ~p: ~p", [Hostname, Else]))
  ).

% Handle all the proxy functions here
proxy_loop(#state{client_socket = CSock, server_socket = SSock, server_pid = SPid, client_pid = CPid} = State) ->
  receive
	  {tcp, CSock, Data} ->
      % Received data from the client
      ?BENCHMARK_LOG("Sending data to Server", gen_tcp, send, [SSock, Data]),
	    proxy_loop(State);
  	{tcp, SSock, Data} ->
      % Received info from the server
      ?BENCHMARK_LOG("Sending data to client", gen_tcp, send, [CSock, Data]),
      proxy_loop(State);
		{tcp_closed, SPid, _Sock} ->
  	  terminate(normal, State);
		{tcp_closed, CPid, _Sock} ->
		  ?LOG(debug, "Client cut off the connection", []),
  	  terminate(normal, State);
  	{tcp_error, SSock} ->
  	  ?LOG(error, "tcp_error on server: ~p", [SSock]),
      terminate(normal, State);
    ?BACKEND_TIMEOUT_MSG ->
      % ?LOG(info, "Backend timeout message", []),
      terminate(timeout, State);
  	_Else -> proxy_loop(State)
  % If there is no activity for a while and the socket has not already closed, 
  % we'll assume that the connection is tired and should close, so we'll close it
  after 3000 ->
    terminate(normal, State)
  end.
<commit_msg>Removed client cut off connection log in proxy_handler<commit_after>%%%-------------------------------------------------------------------
%%% File    : proxy_handler.erl
%%% Author  : <NAME>
%%% Description : 
%%%
%%% Created :  Tue Oct 13 20:24:21 PDT 2009
%%%-------------------------------------------------------------------

-module (proxy_handler).

-include ("common.hrl").
-include ("http.hrl").
-include ("beehive.hrl").

-export ([
  start_link/1
]).

-export ([terminate1/2]).

-record(state, {
  routing_key,          % subdomain of the app to look for
  host,                 % host
  port,                 % port
  bee,                  % Backend
  request,              % client request
  start_time,           % time proxy started
  client_socket,        % client socket
  server_socket,        % server socket
  client_pid,           % client listening pid
  server_pid,           % server listening pid
  timeout               % timeout
}).

% Start the proxy by spawning off an init with the socket.
% Compatible with the supervisor behaviour
start_link(ClientSock) ->
  Fun = fun() -> proxy_init(ClientSock) end,
  Pid = ?BENCHMARK_LOG("------Spawn-linking proxy_init------", erlang, spawn_link, [Fun]),
  {ok, Pid}.

% Receive the initial request and socket. Use the packet_decoder (http for now) to decode the
% routing key required. If this can be found, then start engaging the bee.
% If this takes more than IDLE_TIMEOUT seconds, then exit with an error.
proxy_init(ClientSock) ->
  receive
    {start, ClientSock, RequestPid} ->
      {ok, RoutingKey, ForwardReq, Req} = ?BENCHMARK_LOG("Handling request", http_request_decoder, handle_request, [ClientSock]),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key", bee_srv, get_bee, [self(), RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee);
    _E ->
      proxy_init(ClientSock)
  after ?IDLE_TIMEOUT ->
    % We only give 30 seconds to the proxy pid to be given control of the socket
    % this is MORE than enough time for the socket to be given a chance to prepare
    % to be handled by the proxy accept request.
    send_and_terminate(ClientSock, error, ?APP_ERROR(503, "Proxy is b0rked because of bee selection timeout in init"))
  end.

% Initiate and engage the chosen bee.
% First, we try to connect to the bee. If this succeeds, then let the rest of the app know that the
% bee has been engaged (nonblocking, through the event handler).
% Let the packet_decoder address the forwarding of the packet to the server and start the proxy loop
% If the bee cannot be reached, send an alert through the event handler that we could not reach
% the bee and try to find a new bee
engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, {ok, #bee{host = Host, port = Port} = Bee}) ->
  SockOpts = [binary, {active, false}, {packet, raw} ],
  case gen_tcp:connect(Host, Port, SockOpts, ?CONNECT_TIMEOUT) of
    {ok, ServerSock} ->
      ?NOTIFY({bee, used, Bee}),
      % Sending raw request to bee server
      gen_tcp:send(ServerSock, ForwardReq),
      
      Timeout = timer:seconds(10),
      ProxyPid = self(),
      
      State = #state{
                  start_time = date_util:now_to_seconds(),
                  client_socket = ClientSock, 
                  server_socket = ServerSock, 
                  request = Req,
                  routing_key = RoutingKey,
                  host = Host,
                  port = Port,
                  timeout = Timeout,
                  bee = Bee},
      
      ClientPid = spawn(fun() -> handle_streaming_data(client, ProxyPid, State) end),
      ServerPid = spawn(fun() -> handle_streaming_data(server, ProxyPid, State) end),
      
      NewState = State#state{server_pid = ServerPid, client_pid = ClientPid},
      
      proxy_loop(NewState);
    {error, emfile} ->
      ?LOG(error, "Maximum number of sockets open. Die instead", []),
      ?NOTIFY({bee, cannot_connect, Bee}),
      send_and_terminate(
        ClientSock, 503,
        ?APP_ERROR(503, "503 Service Unavailable")
      );
    {error,econnreset} ->
      timer:sleep(200),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key after a connection reset", bee_srv, get_bee, [RequestPid, RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee);
    Error ->
      ?LOG(error, "Connection to remote TCP server: ~p:~p ~p", [Host, Port, Error]),
      ?NOTIFY({bee, cannot_connect, Bee}),
      timer:sleep(200),
      GetBee = ?BENCHMARK_LOG("Getting bee for routing key after an error", bee_srv, get_bee, [RequestPid, RoutingKey]),
      engage_bee(ClientSock, RequestPid, RoutingKey, ForwardReq, Req, GetBee)
  end;
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, ?BACKEND_TIMEOUT_MSG) ->
  ?LOG(error, "Error getting bee because of timeout: ~p", [Hostname]),
  send_and_terminate(
    ClientSock, ?BACKEND_TIMEOUT_MSG, 
    ?APP_ERROR(404, io_lib:format("Error: ~p", [?BACKEND_TIMEOUT_MSG]))
  );
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, {error, Reason}) ->
  send_and_terminate(
    ClientSock, Reason, 
    ?APP_ERROR(404, io_lib:format("Error on ~p: ~p", [Hostname, Reason]))
  );
engage_bee(ClientSock, _RequestPid, Hostname, _ForwardReq, _Req, Else) ->
  send_and_terminate(
    ClientSock, Else, 
    ?APP_ERROR(404, io_lib:format("Error on ~p: ~p", [Hostname, Else]))
  ).

% Handle all the proxy functions here
proxy_loop(#state{client_socket = CSock, server_socket = SSock, server_pid = SPid, client_pid = CPid} = State) ->
  receive
	  {tcp, CSock, Data} ->
      % Received data from the client
      ?BENCHMARK_LOG("Sending data to Server", gen_tcp, send, [SSock, Data]),
	    proxy_loop(State);
  	{tcp, SSock, Data} ->
      % Received info from the server
      ?BENCHMARK_LOG("Sending data to client", gen_tcp, send, [CSock, Data]),
      proxy_loop(State);
		{tcp_closed, SPid, _Sock} ->
  	  terminate(normal, State);
		{tcp_closed, CPid, _Sock} ->
  	  terminate(normal, State);
  	{tcp_error, SSock} ->
  	  ?LOG(error, "tcp_error on server: ~p", [SSock]),
      terminate(normal, State);
    ?BACKEND_TIMEOUT_MSG ->
      % ?LOG(info, "Backend timeout message", []),
      terminate(timeout, State);
  	_Else -> proxy_loop(State)
  % If there is no activity for a while and the socket has not already closed, 
  % we'll assume that the connection is tired and should close, so we'll close it
  after 3000 ->
    terminate(normal, State)
  end.

{'fraction_non_alphanumeric': 0.1328214971209213, 'fraction_numerical': 0.00783109404990403, 'mean_word_length': 2.8527062999112687, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 55, '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': '1200976', 'n_tokens_mistral': 4291, 'n_tokens_neox': 4022, 'n_words': 1263}