lib/erl/src/transport/tErlAcceptor.erl

   1 %%% Copyright (c) 2007- Facebook
   2 %%% Distributed under the Thrift Software License
   3 %%%
   4 %%% See accompanying file LICENSE or visit the Thrift site at:
   5 %%% http://developers.facebook.com/thrift/
   6
   7 -module(tErlAcceptor).
   8
   9 -include("oop.hrl").
  10 -include("thrift.hrl").
  11 -include("tApplicationException.hrl").
  12 -include("transport/tTransportException.hrl").
  13 -include("transport/tServerSocket.hrl").
  14 -include("transport/tErlAcceptor.hrl").
  15
  16 -include_lib("kernel/include/inet.hrl").
  17
  18 -behavior(oop).
  19
  20 -export([attr/4, super/0, inspect/1]).
  21
  22 -export([new/3, accept/4]).
  23
  24 %%%
  25 %%% define attributes
  26 %%% 'super' is required unless ?MODULE is a base class
  27 %%%
  28
  29 ?DEFINE_ATTR(super);
  30 ?DEFINE_ATTR(serverPid);
  31 ?DEFINE_ATTR(transportFactory);
  32 ?DEFINE_ATTR(protocolFactory).
  33
  34 %%%
  35 %%% behavior callbacks
  36 %%%
  37
  38 %%% super() -> SuperModule = atom()
  39 %%%             |  none
  40
  41 super() ->
  42     tServerTransport.
  43
  44 %%% inspect(This) -> string()
  45
  46 inspect(This) ->
  47     ?FORMAT_ATTR(serverPid) ++ ", " ++
  48     ?FORMAT_ATTR(transportFactory) ++ ", " ++
  49     ?FORMAT_ATTR(protocolFactory).
  50
  51 %%%
  52 %%% class methods
  53 %%%
  54
  55 new(ServerPid, TF, PF) ->
  56     Super = (super()):new(),
  57     #?MODULE{super = Super,
  58              serverPid = ServerPid,
  59              transportFactory = TF,
  60              protocolFactory = PF
  61             }.
  62
  63 %%%
  64 %%% instance methods
  65 %%%
  66
  67 accept(This, ListenSocket, GP, Handler) ->
  68     ServerPid = oop:get(This, serverPid),
  69
  70     case catch gen_tcp:accept(ListenSocket) of
  71         {ok, Socket} ->
  72             ?C0(ServerPid, effectful_new_acceptor), %% cast to create new acceptor
  73
  74             AddrString = render_addr(Socket),
  75             ?INFO("thrift connection accepted from ~s", [AddrString]),
  76
  77             %% start_new(tSocket, [])
  78             Client = oop:start_new(tSocket, []),
  79             ?R1(Client, effectful_setHandle, Socket), %% TODO(cpiro): should we just let this be a param to the constructor?
  80
  81             %% cpiro: OPAQUE!! Trans = Client
  82             TF      = oop:get(This, transportFactory),
  83             Trans   = ?F1(TF, getTransport, Client),
  84
  85             %% cpiro: OPAQUE!! Prot = start_new(tBinaryProtocol, [Trans])
  86             PF      = oop:get(This, protocolFactory),
  87             Prot    = ?F1(PF, getProtocol, Trans),
  88
  89             %% start_new(, ...)
  90             Processor = oop:start_new(tErlProcessor, [GP, Handler]), %% TODO
  91
  92             case receive_loop(This, Processor, Prot, Prot) of
  93                 conn_timeout ->
  94                     ?INFO("thrift connection timed out from ~s", [AddrString]);
  95                 conn_closed ->
  96                     ?INFO("thrift connection closed from ~s", [AddrString]);
  97                 {Class, Else} ->
  98                     ?ERROR("unhandled ~p in tErlAcceptor: ~p", [Class, Else])
  99             end,
 100             exit(normal);
 101
 102         Else ->
 103             R = thrift_utils:sformat("accept() failed: ~p", [Else]),
 104             exit(tTransportException:new(R))
 105     end.
 106
 107 receive_loop(This, Processor, Iprot, Oprot) ->
 108     try ?R2(Processor, process, Iprot, Oprot) of
 109         {error, TAE} when is_record(TAE, tApplicationException),
 110                           TAE#tApplicationException.type == ?tApplicationException_HANDLER_ERROR ->
 111             ?ERROR("thrift handler returned an error: ~p", [oop:get(TAE, message)]),
 112             receive_loop(This, Processor, Iprot, Oprot);
 113         Value ->
 114             ?INFO("thrift request: ~p", [Value]),
 115             receive_loop(This, Processor, Iprot, Oprot)
 116     catch
 117         exit:{timeout, _} ->
 118             conn_timeout;
 119
 120         %% the following clause must be last
 121         %% cpiro: would be best to implement an is_a/2 guard BIF
 122         %% cpiro: breaks if it's a subclass of tTransportException
 123         %% since unnest_record knows nothing about oop
 124         Class:Else ->
 125             case thrift_utils:unnest_record(Else, tTransportException) of
 126                 {ok, TTE} when TTE#tTransportException.type == ?tTransportException_NOT_OPEN ->
 127                     conn_closed;
 128                 _ ->
 129                     {Class, Else}
 130             end
 131     end.
 132
 133 %% helper functions
 134
 135 %% @param Socket the socket in question
 136 %% TODO(cpiro): there probably needs to be a switch for DoLookup somewhere prominent and outside the lib,
 137 %% probably at the "application" level
 138 render_addr(Socket) ->
 139     DoLookup = true,
 140     {ok, {Peer, Port}} = inet:peername(Socket),
 141
 142     case Peer of
 143         _ when DoLookup ->
 144             case catch inet:gethostbyaddr(Peer) of
 145                 {ok, Hostent} ->
 146                     thrift_utils:sformat("~s:~p", [Hostent#hostent.h_name, Port]);
 147                 _ ->
 148                     "??"
 149             end;
 150
 151         {A,B,C,D} when not DoLookup ->
 152             thrift_utils:sformat("~p.~p.~p.~p:~p", [A,B,C,D,Port])
 153     end.