amqp_client

%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (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.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
%% License for the specific language governing rights and limitations
%% under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developer of the Original Code is VMware, Inc.
%% Copyright (c) 2007-2012 VMware, Inc.  All rights reserved.
%%

-module(amqp_uri).

-include("amqp_client.hrl").

-export([parse/1, parse/2]).

%%---------------------------------------------------------------------------
%% AMQP URI Parsing
%%---------------------------------------------------------------------------

%% @spec (Uri) -> {ok, #amqp_params_network{} | #amqp_params_direct{}} |
%%                {error, {Info, Uri}}
%% where
%%      Uri  = string()
%%      Info = any()
%%
%% @doc Parses an AMQP URI.  If any of the URI parts are missing, the
%% default values are used.  If the hostname is zero-length, an
%% #amqp_params_direct{} record is returned; otherwise, an
%% #amqp_params_network{} record is returned.  Extra parameters may be
%% specified via the query string (e.g. "?heartbeat=5"). In case of
%% failure, an {error, {Info, Uri}} tuple is returned.
%%
%% The extra parameters that may be specified are channel_max,
%% frame_max, and heartbeat.  The extra parameters that may be
%% specified for an SSL connection are cacertfile, certfile, keyfile,
%% verify, and fail_if_no_peer_cert.
parse(Uri) -> parse(Uri, <<"/">>).

parse(Uri, DefaultVHost) ->
    try case parse1(Uri, DefaultVHost) of
            {ok, #amqp_params_network{host         = undefined,
                                      username     = User,
                                      virtual_host = Vhost}} ->
                return({ok, #amqp_params_direct{username     = User,
                                                virtual_host = Vhost}});
            {ok, Params} ->
                return({ok, Params})
        end
    catch throw:Err -> {error, {Err, Uri}};
          error:Err -> {error, {Err, Uri}}
    end.

parse1(Uri, DefaultVHost) when is_list(Uri) ->
    case uri_parser:parse(Uri, [{host, undefined}, {path, undefined},
                                {port, undefined}, {'query', []}]) of
        {error, Err} ->
            throw({unable_to_parse_uri, Err});
        Parsed ->
            Endpoint =
                case string:to_lower(proplists:get_value(scheme, Parsed)) of
                    "amqp"  -> build_broker(Parsed, DefaultVHost);
                    "amqps" -> build_ssl_broker(Parsed, DefaultVHost);
                    Scheme  -> fail({unexpected_uri_scheme, Scheme})
                end,
            return({ok, broker_add_query(Endpoint, Parsed)})
    end;
parse1(_, _DefaultVHost) ->
    fail(expected_string_uri).

unescape_string(Atom) when is_atom(Atom) ->
    Atom;
unescape_string([]) ->
    [];
unescape_string([$%, N1, N2 | Rest]) ->
    try
        [erlang:list_to_integer([N1, N2], 16) | unescape_string(Rest)]
    catch
        error:badarg -> throw({invalid_entitiy, ['%', N1, N2]})
    end;
unescape_string([$% | Rest]) ->
    fail({unterminated_entity, ['%' | Rest]});
unescape_string([C | Rest]) ->
    [C | unescape_string(Rest)].

build_broker(ParsedUri, DefaultVHost) ->
    [Host, Port, Path] =
        [proplists:get_value(F, ParsedUri) || F <- [host, port, path]],
    case Port =:= undefined orelse (0 < Port andalso Port =< 65535) of
        true  -> ok;
        false -> fail({port_out_of_range, Port})
    end,
    VHost = case Path of
                undefined -> DefaultVHost;
                [$/|Rest] -> case string:chr(Rest, $/) of
                                 0 -> list_to_binary(unescape_string(Rest));
                                 _ -> fail({invalid_vhost, Rest})
                             end
            end,
    UserInfo = proplists:get_value(userinfo, ParsedUri),
    Ps = #amqp_params_network{host            = unescape_string(Host),
                              port            = Port,
                              virtual_host    = VHost,
                              auth_mechanisms = mechanisms(ParsedUri)},
    case UserInfo of
        [U, P | _] -> Ps#amqp_params_network{
                        username = list_to_binary(unescape_string(U)),
                        password = list_to_binary(unescape_string(P))};
        [U | _]    -> Ps#amqp_params_network{
                        username = list_to_binary(unescape_string(U))};
        _          -> Ps
    end.

build_ssl_broker(ParsedUri, DefaultVHost) ->
    Params = build_broker(ParsedUri, DefaultVHost),
    Query = proplists:get_value('query', ParsedUri),
    SSLOptions =
        run_state_monad(
          [fun (L) -> KeyString = atom_to_list(Key),
                      case lists:keysearch(KeyString, 1, Query) of
                          {value, {_, Value}} ->
                              try return([{Key, unescape_string(Fun(Value))} | L])
                              catch throw:Reason ->
                                      fail({invalid_ssl_parameter,
                                            Key, Value, Query, Reason})
                              end;
                          false ->
                              L
                      end
           end || {Fun, Key} <-
                      [{fun find_path_parameter/1,    cacertfile},
                       {fun find_path_parameter/1,    certfile},
                       {fun find_path_parameter/1,    keyfile},
                       {fun find_atom_parameter/1,    verify},
                       {fun find_boolean_parameter/1, fail_if_no_peer_cert}]],
          []),
    Params#amqp_params_network{ssl_options = SSLOptions}.

broker_add_query(Params = #amqp_params_network{}, Uri) ->
    broker_add_query(Params, Uri, record_info(fields, amqp_params_network)).

broker_add_query(Params, ParsedUri, Fields) ->
    Query = proplists:get_value('query', ParsedUri),
    {Params1, _Pos} =
        run_state_monad(
          [fun ({ParamsN, Pos}) ->
                   Pos1 = Pos + 1,
                   KeyString = atom_to_list(Field),
                   case proplists:get_value(KeyString, Query) of
                       undefined ->
                           return({ParamsN, Pos1});
                       true -> %% proplists short form, not permitted
                           return({ParamsN, Pos1});
                       Value ->
                           try
                               ValueParsed = parse_amqp_param(Field, Value),
                               return(
                                 {setelement(Pos, ParamsN, ValueParsed), Pos1})
                           catch throw:Reason ->
                                   fail({invalid_amqp_params_parameter,
                                         Field, Value, Query, Reason})
                           end
                   end
           end || Field <- Fields], {Params, 2}),
    Params1.

parse_amqp_param(Field, String) when Field =:= channel_max orelse
                                     Field =:= frame_max   orelse
                                     Field =:= heartbeat   ->
    try return(list_to_integer(String))
    catch error:badarg -> fail({not_an_integer, String})
    end;
parse_amqp_param(Field, String) ->
    fail({parameter_unconfigurable_in_query, Field, String}).

find_path_parameter(Value) -> return(Value).

find_boolean_parameter(Value) ->
    Bool = list_to_atom(Value),
    case is_boolean(Bool) of
        true  -> return(Bool);
        false -> fail({require_boolean, Bool})
    end.

find_atom_parameter(Value) -> return(list_to_atom(Value)).

mechanisms(ParsedUri) ->
    Query = proplists:get_value('query', ParsedUri),
    Mechanisms = case proplists:get_all_values("auth_mechanism", Query) of
                     []    -> ["plain", "amqplain"];
                     Mechs -> Mechs
                 end,
    [case [list_to_atom(T) || T <- string:tokens(Mech, ":")] of
         [F]    -> fun (R, P, S) -> amqp_auth_mechanisms:F(R, P, S) end;
         [M, F] -> fun (R, P, S) -> M:F(R, P, S) end;
         L      -> throw({not_mechanism, L})
     end || Mech <- Mechanisms].

%% --=: Plain state monad implementation start :=--
run_state_monad(FunList, State) ->
    lists:foldl(fun (Fun, StateN) -> Fun(StateN) end, State, FunList).

return(V) -> V.

fail(Reason) -> throw(Reason).
%% --=: end :=--