Subversion Repositories SmartDukaan

%%

%% 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_client).

-behaviour(gen_server).

%% API

-export([start_link/2, start_link/3, start_link/4,

         start/3, start/4,

         call/3, send_call/3, close/1]).

%% gen_server callbacks

-export([init/1, handle_call/3, handle_cast/2, handle_info/2,

         terminate/2, code_change/3]).

-include("thrift_constants.hrl").

-include("thrift_protocol.hrl").

-record(state, {service, protocol, seqid}).

%%====================================================================

%% API

%%====================================================================

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

%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}

%% Description: Starts the server as a linked process.

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

start_link(Host, Port, Service) when is_integer(Port), is_atom(Service) ->

    start_link(Host, Port, Service, []).

start_link(Host, Port, Service, Options) ->

    start(Host, Port, Service, [{monitor, link} | Options]).

start_link(ProtocolFactory, Service) ->

    start(ProtocolFactory, Service, [{monitor, link}]).

%%

%% Splits client options into protocol options and transport options

%%

%% split_options([Options...]) -> {ProtocolOptions, TransportOptions}

%%

split_options(Options) ->

    split_options(Options, [], [], []).

split_options([], ClientIn, ProtoIn, TransIn) ->

    {ClientIn, ProtoIn, TransIn};

split_options([Opt = {OptKey, _} | Rest], ClientIn, ProtoIn, TransIn)

  when OptKey =:= monitor ->

    split_options(Rest, [Opt | ClientIn], ProtoIn, TransIn);

split_options([Opt = {OptKey, _} | Rest], ClientIn, ProtoIn, TransIn)

  when OptKey =:= strict_read;

       OptKey =:= strict_write ->

    split_options(Rest, ClientIn, [Opt | ProtoIn], TransIn);

split_options([Opt = {OptKey, _} | Rest], ClientIn, ProtoIn, TransIn)

  when OptKey =:= framed;

       OptKey =:= connect_timeout;

       OptKey =:= sockopts ->

    split_options(Rest, ClientIn, ProtoIn, [Opt | TransIn]).

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

%% Function: start() -> {ok,Pid} | ignore | {error,Error}

%% Description: Starts the server as an unlinked process.

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

%% Backwards-compatible starter for the common-case of socket transports

start(Host, Port, Service, Options)

  when is_integer(Port), is_atom(Service), is_list(Options) ->

    {ClientOpts, ProtoOpts, TransOpts} = split_options(Options),

    {ok, TransportFactory} =

        thrift_socket_transport:new_transport_factory(Host, Port, TransOpts),

    {ok, ProtocolFactory} = thrift_binary_protocol:new_protocol_factory(

                              TransportFactory, ProtoOpts),

    start(ProtocolFactory, Service, ClientOpts).

%% ProtocolFactory :: fun() -> thrift_protocol()

start(ProtocolFactory, Service, ClientOpts)

  when is_function(ProtocolFactory), is_atom(Service) ->

    {Starter, Opts} =

        case lists:keysearch(monitor, 1, ClientOpts) of

            {value, {monitor, link}} ->

                {start_link, []};

            {value, {monitor, tether}} ->

                {start, [{tether, self()}]};

            _ ->

                {start, []}

        end,

    Connect =

        case lists:keysearch(connect, 1, ClientOpts) of

            {value, {connect, Choice}} ->

                Choice;

            _ ->

                %% By default, connect at creation-time.

                true

        end,

    Started = gen_server:Starter(?MODULE, [Service, Opts], []),

    if

        Connect ->

            case Started of

                {ok, Pid} ->

                    case gen_server:call(Pid, {connect, ProtocolFactory}) of

                        ok ->

                            {ok, Pid};

                        Error ->

                            Error

                    end;

                Else ->

                    Else

            end;

        true ->

            Started

    end.

call(Client, Function, Args)

  when is_pid(Client), is_atom(Function), is_list(Args) ->

    case gen_server:call(Client, {call, Function, Args}) of

        R = {ok, _} -> R;

        R = {error, _} -> R;

        {exception, Exception} -> throw(Exception)

    end.

cast(Client, Function, Args)

  when is_pid(Client), is_atom(Function), is_list(Args) ->

    gen_server:cast(Client, {call, Function, Args}).

%% Sends a function call but does not read the result. This is useful

%% if you're trying to log non-oneway function calls to write-only

%% transports like thrift_disk_log_transport.

send_call(Client, Function, Args)

  when is_pid(Client), is_atom(Function), is_list(Args) ->

    gen_server:call(Client, {send_call, Function, Args}).

close(Client) when is_pid(Client) ->

    gen_server:cast(Client, close).

%%====================================================================

%% gen_server callbacks

%%====================================================================

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

%% Function: init(Args) -> {ok, State} |

%%                         {ok, State, Timeout} |

%%                         ignore               |

%%                         {stop, Reason}

%% Description: Initiates the server

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

init([Service, Opts]) ->

    case lists:keysearch(tether, 1, Opts) of

        {value, {tether, Pid}} ->

            erlang:monitor(process, Pid);

        _Else ->

            ok

    end,

    {ok, #state{service = Service}}.

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

%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |

%%                                      {reply, Reply, State, Timeout} |

%%                                      {noreply, State} |

%%                                      {noreply, State, Timeout} |

%%                                      {stop, Reason, Reply, State} |

%%                                      {stop, Reason, State}

%% Description: Handling call messages

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

handle_call({connect, ProtocolFactory}, _From,

            State = #state{service = Service}) ->

    case ProtocolFactory() of

        {ok, Protocol} ->

            {reply, ok, State#state{protocol = Protocol,

                                    seqid = 0}};

        Error ->

            {stop, normal, Error, State}

    end;

handle_call({call, Function, Args}, _From, State = #state{service = Service}) ->

    Result = catch_function_exceptions(

               fun() ->

                       ok = send_function_call(State, Function, Args),

                       receive_function_result(State, Function)

               end,

               Service),

    {reply, Result, State};

handle_call({send_call, Function, Args}, _From, State = #state{service = Service}) ->

    Result = catch_function_exceptions(

               fun() ->

                       send_function_call(State, Function, Args)

               end,

               Service),

    {reply, Result, State}.

%% Helper function that catches exceptions thrown by sending or receiving

%% a function and returns the correct response for call or send_only above.

catch_function_exceptions(Fun, Service) ->

    try

        Fun()

    catch

        throw:{return, Return} ->

            Return;

          error:function_clause ->

            ST = erlang:get_stacktrace(),

            case hd(ST) of

                {Service, function_info, [Function, _]} ->

                    {error, {no_function, Function}};

                _ -> throw({error, {function_clause, ST}})

            end

    end.

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

%% Function: handle_cast(Msg, State) -> {noreply, State} |

%%                                      {noreply, State, Timeout} |

%%                                      {stop, Reason, State}

%% Description: Handling cast messages

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

handle_cast({call, Function, Args}, State = #state{service = Service,

                                                   protocol = Protocol,

                                                   seqid = SeqId}) ->

    _Result =

        try

            ok = send_function_call(State, Function, Args),

            receive_function_result(State, Function)

        catch

            Class:Reason ->

                error_logger:error_msg("error ignored in handle_cast({cast,...},...): ~p:~p~n", [Class, Reason])

        end,

    {noreply, State};

handle_cast(close, State=#state{protocol = Protocol}) ->

%%     error_logger:info_msg("thrift_client ~p received close", [self()]),

    {stop,normal,State};

handle_cast(_Msg, State) ->

    {noreply, State}.

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

%% Function: handle_info(Info, State) -> {noreply, State} |

%%                                       {noreply, State, Timeout} |

%%                                       {stop, Reason, State}

%% Description: Handling all non call/cast messages

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

handle_info({'DOWN', MonitorRef, process, Pid, _Info}, State)

  when is_reference(MonitorRef), is_pid(Pid) ->

    %% We don't actually verify the correctness of the DOWN message.

    {stop, parent_died, State};

handle_info(_Info, State) ->

    {noreply, State}.

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

%% Function: terminate(Reason, State) -> void()

%% Description: This function is called by a gen_server when it is about to

%% terminate. It should be the opposite of Module:init/1 and do any necessary

%% cleaning up. When it returns, the gen_server terminates with Reason.

%% The return value is ignored.

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

terminate(Reason, State = #state{protocol=undefined}) ->

    ok;

terminate(Reason, State = #state{protocol=Protocol}) ->

    thrift_protocol:close_transport(Protocol),

    ok.

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

%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}

%% Description: Convert process state when code is changed

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

code_change(_OldVsn, State, _Extra) ->

    {ok, State}.

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

%%% Internal functions

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

send_function_call(#state{protocol = Proto,

                          service  = Service,

                          seqid    = SeqId},

                   Function,

                   Args) ->

    Params = Service:function_info(Function, params_type),

    {struct, PList} = Params,

    if

        length(PList) =/= length(Args) ->

            throw({return, {error, {bad_args, Function, Args}}});

        true -> ok

    end,

    Begin = #protocol_message_begin{name = atom_to_list(Function),

                                    type = ?tMessageType_CALL,

                                    seqid = SeqId},

    ok = thrift_protocol:write(Proto, Begin),

    ok = thrift_protocol:write(Proto, {Params, list_to_tuple([Function | Args])}),

    ok = thrift_protocol:write(Proto, message_end),

    thrift_protocol:flush_transport(Proto),

    ok.

receive_function_result(State = #state{protocol = Proto,

                                       service = Service},

                        Function) ->

    ResultType = Service:function_info(Function, reply_type),

    read_result(State, Function, ResultType).

read_result(_State,

            _Function,

            oneway_void) ->

    {ok, ok};

read_result(State = #state{protocol = Proto,

                           seqid    = SeqId},

            Function,

            ReplyType) ->

    case thrift_protocol:read(Proto, message_begin) of

        #protocol_message_begin{seqid = RetSeqId} when RetSeqId =/= SeqId ->

            {error, {bad_seq_id, SeqId}};

        #protocol_message_begin{type = ?tMessageType_EXCEPTION} ->

            handle_application_exception(State);

        #protocol_message_begin{type = ?tMessageType_REPLY} ->

            handle_reply(State, Function, ReplyType)

    end.

handle_reply(State = #state{protocol = Proto,

                            service = Service},

             Function,

             ReplyType) ->

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

    ReplyStructDef = {struct, [{0, ReplyType}] ++ ExceptionFields},

    {ok, Reply} = thrift_protocol:read(Proto, ReplyStructDef),

    ReplyList = tuple_to_list(Reply),

    true = length(ReplyList) == length(ExceptionFields) + 1,

    ExceptionVals = tl(ReplyList),

    Thrown = [X || X <- ExceptionVals,

                   X =/= undefined],

    Result =

        case Thrown of

            [] when ReplyType == {struct, []} ->

                {ok, ok};

            [] ->

                {ok, hd(ReplyList)};

            [Exception] ->

                {exception, Exception}

        end,

    ok = thrift_protocol:read(Proto, message_end),

    Result.

handle_application_exception(State = #state{protocol = Proto}) ->

    {ok, Exception} = thrift_protocol:read(Proto,

                                           ?TApplicationException_Structure),

    ok = thrift_protocol:read(Proto, message_end),

    XRecord = list_to_tuple(

                ['TApplicationException' | tuple_to_list(Exception)]),

    error_logger:error_msg("X: ~p~n", [XRecord]),

    true = is_record(XRecord, 'TApplicationException'),

    {exception, XRecord}.