lib/erl/src/thrift_processor.erl

%%
%% Licensed to the Apache Software Foundation (ASF) under one
%% or more contributor license agreements. See the NOTICE file
%% distributed with this work for additional information
%% regarding copyright ownership. The ASF licenses this file
%% to you under the Apache License, Version 2.0 (the
%% "License"); you may not use this file except in compliance
%% with the License. You may obtain a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied. See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%

-module(thrift_processor).

-export([init/1]).

-include("thrift_constants.hrl").
-include("thrift_protocol.hrl").

-record(thrift_processor, {handler, in_protocol, out_protocol, service}).

init({Server, ProtoGen, Service, Handler}) when is_function(ProtoGen, 0) ->
    {ok, IProt, OProt} = ProtoGen(),
    loop(#thrift_processor{in_protocol = IProt,
                           out_protocol = OProt,
                           service = Service,
                           handler = Handler}).

loop(State = #thrift_processor{in_protocol  = IProto,
                               out_protocol = OProto}) ->
    case thrift_protocol:read(IProto, message_begin) of
        #protocol_message_begin{name = Function,
                                type = ?tMessageType_CALL} ->
            ok = handle_function(State, list_to_atom(Function)),
            loop(State);
        #protocol_message_begin{name = Function,
                                type = ?tMessageType_ONEWAY} ->
            ok = handle_function(State, list_to_atom(Function)),
            loop(State);
        {error, timeout} ->
            thrift_protocol:close_transport(OProto),
            ok;
        {error, closed} ->
            %% error_logger:info_msg("Client disconnected~n"),
            thrift_protocol:close_transport(OProto),
            exit(shutdown)
    end.

handle_function(State=#thrift_processor{in_protocol = IProto,
                                        out_protocol = OProto,
                                        handler = Handler,
                                        service = Service},
                Function) ->
    InParams = Service:function_info(Function, params_type),

    {ok, Params} = thrift_protocol:read(IProto, InParams),

    try
        Result = Handler:handle_function(Function, Params),
        %% {Micro, Result} = better_timer(Handler, handle_function, [Function, Params]),
        %% error_logger:info_msg("Processed ~p(~p) in ~.4fms~n",
        %%                       [Function, Params, Micro/1000.0]),
        handle_success(State, Function, Result)
    catch
        Type:Data ->
            handle_function_catch(State, Function, Type, Data)
    end,
    after_reply(OProto).

handle_function_catch(State = #thrift_processor{service = Service},
                      Function, ErrType, ErrData) ->
    IsOneway = Service:function_info(Function, reply_type) =:= oneway_void,

    case {ErrType, ErrData} of
        _ when IsOneway ->
            Stack = erlang:get_stacktrace(),
            error_logger:warning_msg(
              "oneway void ~p threw error which must be ignored: ~p",
              [Function, {ErrType, ErrData, Stack}]),
            ok;

        {throw, Exception} when is_tuple(Exception), size(Exception) > 0 ->
            error_logger:warning_msg("~p threw exception: ~p~n", [Function, Exception]),
            handle_exception(State, Function, Exception),
            ok;   % we still want to accept more requests from this client

        {error, Error} ->
            ok = handle_error(State, Function, Error)
    end.

handle_success(State = #thrift_processor{out_protocol = OProto,
                                         service = Service},
               Function,
               Result) ->
    ReplyType  = Service:function_info(Function, reply_type),
    StructName = atom_to_list(Function) ++ "_result",

    ok = case Result of
             {reply, ReplyData} ->
                 Reply = {{struct, [{0, ReplyType}]}, {StructName, ReplyData}},
                 send_reply(OProto, Function, ?tMessageType_REPLY, Reply);

             ok when ReplyType == {struct, []} ->
                 send_reply(OProto, Function, ?tMessageType_REPLY, {ReplyType, {StructName}});

             ok when ReplyType == oneway_void ->
                 %% no reply for oneway void
                 ok
         end.

handle_exception(State = #thrift_processor{out_protocol = OProto,
                                           service = Service},
                 Function,
                 Exception) ->
    ExceptionType = element(1, Exception),
    %% Fetch a structure like {struct, [{-2, {struct, {Module, Type}}},
    %%                                  {-3, {struct, {Module, Type}}}]}

    ReplySpec = Service:function_info(Function, exceptions),
    {struct, XInfo} = ReplySpec,

    true = is_list(XInfo),

    %% Assuming we had a type1 exception, we'd get: [undefined, Exception, undefined]
    %% e.g.: [{-1, type0}, {-2, type1}, {-3, type2}]
    ExceptionList = [case Type of
                         ExceptionType -> Exception;
                         _ -> undefined
                     end
                     || {_Fid, {struct, {_Module, Type}}} <- XInfo],

    ExceptionTuple = list_to_tuple([Function | ExceptionList]),

                                                % Make sure we got at least one defined
    case lists:all(fun(X) -> X =:= undefined end, ExceptionList) of
        true ->
            ok = handle_unknown_exception(State, Function, Exception);
        false ->
            ok = send_reply(OProto, Function, ?tMessageType_REPLY, {ReplySpec, ExceptionTuple})
    end.

%%
%% Called when an exception has been explicitly thrown by the service, but it was
%% not one of the exceptions that was defined for the function.
%%
handle_unknown_exception(State, Function, Exception) ->
    handle_error(State, Function, {exception_not_declared_as_thrown,
                                   Exception}).

handle_error(#thrift_processor{out_protocol = OProto}, Function, Error) ->
    Stack = erlang:get_stacktrace(),
    error_logger:error_msg("~p had an error: ~p~n", [Function, {Error, Stack}]),

    Message =
        case application:get_env(thrift, exceptions_include_traces) of
            {ok, true} ->
                lists:flatten(io_lib:format("An error occurred: ~p~n",
                                            [{Error, Stack}]));
            _ ->
                "An unknown handler error occurred."
        end,
    Reply = {?TApplicationException_Structure,
             #'TApplicationException'{
                message = Message,
                type = ?TApplicationException_UNKNOWN}},
    send_reply(OProto, Function, ?tMessageType_EXCEPTION, Reply).

send_reply(OProto, Function, ReplyMessageType, Reply) ->
    ok = thrift_protocol:write(OProto, #protocol_message_begin{
                                 name = atom_to_list(Function),
                                 type = ReplyMessageType,
                                 seqid = 0}),
    ok = thrift_protocol:write(OProto, Reply),
    ok = thrift_protocol:write(OProto, message_end),
    ok = thrift_protocol:flush_transport(OProto),
    ok.

after_reply(OProto) ->
    ok = thrift_protocol:flush_transport(OProto)
    %%     ok = thrift_protocol:close_transport(OProto)
    .