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

-export([new/4]).
-export([new_multiplexed/3, new_multiplexed/4]).

-type service_name() :: nonempty_string().
-type service_module() :: atom().
-type multiplexed_service_map() :: [
    {ServiceName :: service_name(), ServiceModule :: service_module()}
].

%%
%% Splits client options into client, protocol, and transport options
%%
%% split_options([Options...]) -> {ProtocolOptions, TransportOptions}
%%
split_options(Options) ->
    split_options(Options, [], []).

split_options([], ProtoIn, TransIn) ->
    {ProtoIn, TransIn};
split_options([Opt = {OptKey, _} | Rest], ProtoIn, TransIn) when
    OptKey =:= strict_read;
    OptKey =:= strict_write;
    OptKey =:= protocol
->
    split_options(Rest, [Opt | ProtoIn], TransIn);
split_options([Opt = {OptKey, _} | Rest], ProtoIn, TransIn) when
    OptKey =:= framed;
    OptKey =:= connect_timeout;
    OptKey =:= recv_timeout;
    OptKey =:= sockopts;
    OptKey =:= ssltransport;
    OptKey =:= ssloptions
->
    split_options(Rest, ProtoIn, [Opt | TransIn]).

%% Client constructor for the common-case of socket transports
new(Host, Port, Service, Options) when
    is_integer(Port), is_atom(Service), is_list(Options)
->
    {ProtoOpts, TransOpts0} = split_options(Options),

    {TransportModule, TransOpts2} =
        case lists:keytake(ssltransport, 1, TransOpts0) of
            {value, {_, true}, TransOpts1} -> {thrift_sslsocket_transport, TransOpts1};
            false -> {thrift_socket_transport, TransOpts0}
        end,

    {ProtocolModule, ProtoOpts1} =
        case lists:keytake(protocol, 1, ProtoOpts) of
            {value, {_, compact}, Opts} -> {thrift_compact_protocol, Opts};
            {value, {_, json}, Opts} -> {thrift_json_protocol, Opts};
            {value, {_, binary}, Opts} -> {thrift_binary_protocol, Opts};
            false -> {thrift_binary_protocol, ProtoOpts}
        end,
    {ok, TransportFactory} =
        TransportModule:new_transport_factory(Host, Port, TransOpts2),

    {ok, ProtocolFactory} = ProtocolModule:new_protocol_factory(
        TransportFactory, ProtoOpts1
    ),

    case ProtocolFactory() of
        {ok, Protocol} ->
            thrift_client:new(Protocol, Service);
        {error, Error} ->
            {error, Error}
    end.

-spec new_multiplexed(Host, Port, Services, Options) -> {ok, ServiceThriftClientList} when
    Host :: nonempty_string(),
    Port :: non_neg_integer(),
    Services :: multiplexed_service_map(),
    Options :: list(),
    ServiceThriftClientList :: [{ServiceName :: list(), ThriftClient :: term()}].
new_multiplexed(Host, Port, Services, Options) when
    is_integer(Port),
    is_list(Services),
    is_list(Options)
->
    new_multiplexed(
        thrift_socket_transport:new_transport_factory(Host, Port, Options), Services, Options
    ).

-spec new_multiplexed(TransportFactoryTuple, Services, Options) ->
    {ok, ServiceThriftClientList}
when
    TransportFactoryTuple :: {ok, TransportFactory :: term()},
    Services :: multiplexed_service_map(),
    Options :: list(),
    ServiceThriftClientList :: [{ServiceName :: service_name(), ThriftClient :: term()}].
new_multiplexed(TransportFactoryTuple, Services, Options) when
    is_list(Services),
    is_list(Options),
    is_tuple(TransportFactoryTuple)
->
    {ProtoOpts, _} = split_options(Options),

    {ok, TransportFactory} = TransportFactoryTuple,

    {ok, ProtocolFactory} = thrift_binary_protocol:new_protocol_factory(
        TransportFactory, ProtoOpts
    ),

    {ok, Protocol} = ProtocolFactory(),

    {ok, [
        {ServiceName,
            element(
                2,
                thrift_client:new(
                    element(2, thrift_multiplexed_protocol:new(Protocol, ServiceName)), Service
                )
            )}
     || {ServiceName, Service} <- Services
    ]}.