Connecting to Azure Service Bus

[stevenlivz]

Hi. Firstly, I'm very new to Erlang in general but keen to learn. I know Azure pretty well but AMQP is new on me. I have been working on this for a couple of days and learned a lot.

I have created a new Erlang application and used your test code to try and push a message and receive it. I am now able to connect and authenticate but no matter what I do, cannot get a message posted - i get no error but i also see no message on the bus. If i try from Python i can do it.

In addition, the receive section below throws a "no match of right hand side value {error,timeout}" and i can't get past that ... but i'd like to just get a message in to start with.

The code in the README didn't work so i tried the test code and given it is from May it's not that old so should be ok i'd have thought.

Anyone have any ideas? Is this my code error, a specific config i need to add or an issue with the library?

Appreciate all the work - and any advice.

%%%-------------------------------------------------------------------
%% @doc myapp public API
%% @end
%%%-------------------------------------------------------------------

-module(myapp_app).

-behaviour(application).

%% Application callbacks
-export([start/2, stop/1, connect/0]).

%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called whenever an application is started using
%% application:start/[1,2], and should start the processes of the
%% application. If the application is structured according to the OTP
%% design principles as a supervision tree, this means starting the
%% top supervisor of the tree.
%%
%% @spec start(StartType, StartArgs) -> {ok, Pid} |
%%                                      {ok, Pid, State} |
%%                                      {error, Reason}
%%      StartType = normal | {takeover, Node} | {failover, Node}
%%      StartArgs = term()
%% @end
%%--------------------------------------------------------------------
start(_StartType, _StartArgs) ->
  io:format("starting~n"),
  application:start(ssl),
  application:ensure_all_started(amqp10_client),
  myapp_sup:start_link().

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called whenever an application has stopped. It
%% is intended to be the opposite of Module:start/2 and should do
%% any necessary cleaning up. The return value is ignored.
%%
%% @spec stop(State) -> void()
%% @end
%%--------------------------------------------------------------------
stop(_State) ->
  ok.

connect() ->
  io:format("Connecting~n"),
  basic_roundtrip_service_bus(),
  ok.

basic_roundtrip_service_bus() ->
  Hostname = "myazurebroker.servicebus.windows.net",
  Port = 5671,
  User = "RootManageSharedAccessKey",
  Password = "XXXXXXXXXXXXXXXXXXXXXX=",
  OpnConf = #{
    address => Hostname,
    hostname => to_bin(Hostname),
    port => Port,
    notify => self(),
    tls_opts => {secure_port, [{versions, ['tlsv1.1']}]},
    container_id =>  <<"test">>,
    sasl => {plain, to_bin(User), to_bin(Password)}
  },
  roundtrip(OpnConf).

roundtrip(OpenConf) ->
  {ok, Connection} = amqp10_client:open_connection(OpenConf),
  {ok, Session} = amqp10_client:begin_session(Connection),
  {ok, Sender} = amqp10_client:attach_sender_link(Session,
    <<"test-sender">>,
    <<"test">>,
    settled),
  await_link(Sender, credited, link_credit_timeout),

  Now = os:system_time(millisecond),
  Props = #{creation_time => Now},
  Msg0 =  amqp10_msg:set_properties(Props,
    amqp10_msg:new(<<"my-tag">>, <<"banana">>,
      true)),
  Msg1 = amqp10_msg:set_application_properties(#{"a_key" => "a_value"}, Msg0),
  Msg = amqp10_msg:set_message_annotations(#{<<"x_key">> => "x_value"}, Msg1),
  % RabbitMQ AMQP 1.0 does not yet support delivery annotations
  % Msg = amqp10_msg:set_delivery_annotations(#{<<"x_key">> => "x_value"}, Msg2),
  ok = amqp10_client:send_msg(Sender, Msg),
  ok = amqp10_client:detach_link(Sender),
  await_link(Sender, {detached, normal}, link_detach_timeout),

  {error, link_not_found} = amqp10_client:detach_link(Sender),
  {ok, Receiver} = amqp10_client:attach_receiver_link(Session,
    <<"banana-receiver">>,
    <<"test">>,
    settled),
  {ok, OutMsg} = amqp10_client:get_msg(Receiver),
  ok = amqp10_client:end_session(Session),
  ok = amqp10_client:close_connection(Connection),
  #{creation_time := Now} = amqp10_msg:properties(OutMsg),
  #{<<"a_key">> := <<"a_value">>} = amqp10_msg:application_properties(OutMsg),
  #{<<"x_key">> := <<"x_value">>} = amqp10_msg:message_annotations(OutMsg),
  ok.

%%====================================================================
%% Internal functions
%====================================================================

await_link(Who, What, Err) ->
  receive
    {amqp10_event, {link, Who, What}} ->
      ok;
    {amqp10_event, {link, Who, {detached, Why}}} ->
      exit(Why)
  after 5000 -> exit(Err)
  end.

to_bin(X) when is_list(X) ->
  list_to_binary(X).

xxxx

[michaelklishin]

Thank you for your time.

Team RabbitMQ uses GitHub issues for specific actionable items engineers can work on. This assumes two things:

1. GitHub issues are not used for questions, investigations, root cause analysis, discussions of potential issues, etc (as defined by this team)
2. We have a certain amount of information to work with

We get at least a dozen of questions through various venues every single day, often quite light on details. At that rate GitHub issues can very quickly turn into a something impossible to navigate and make sense of even for our team. Because of that questions, investigations, root cause analysis, discussions of potential features are all considered to be mailing list material by our team. Please post this to rabbitmq-users.

Getting all the details necessary to reproduce an issue, make a conclusion or even form a hypothesis about what's happening can take a fair amount of time. Our team is multiple orders of magnitude smaller than the RabbitMQ community. Please help others help you by providing a way to reproduce the behavior you're observing, or at least sharing as much relevant information as possible on the list:

• Server, client library and plugin (if applicable) versions used
• Server logs
• A code example or terminal transcript that can be used to reproduce
• Full exception stack traces (not a single line message)
• rabbitmqctl status (and, if possible, rabbitmqctl environment output)
• Other relevant things about the environment and workload, e.g. a traffic capture

Feel free to edit out hostnames and other potentially sensitive information.

When/if we have enough details and evidence we'd be happy to file a new issue.

Thank you.

[michaelklishin]

Consider posting full stack traces when seeking help, for RabbitMQ projects and elsewhere.

{error, timeout} means that something timed out, it could be a server connection (likely) or something else (less likely). We cannot suggest much more with the amount of information provided.

Please take this to rabbitmq-users.

[stevenlivz]

Thanks. I tried a few places but as this is so new hard to find anyone. I figured the test cases would work but when they didn't i was at a a loss. Thanks. Will try that group.

[michaelklishin]

@stevenlivz the test suite of this plugin is fairly involved and has several dependencies.

It assumes it can start

• a RabbitMQ node with AMQP 1.0 plugin
• an ActiveMQ node (downloaded automatically).

It mean implicitly or explicitly require that RabbitMQ public umbrella is used.

rabbitmq-shovel is a good example of how this client can be used, or at least the best our team has to offer.