A working example for accessing Service Bus session-enabled queues.Also provides some general notes on Service Bus usage.
#include "options.hpp"
#include <iostream>
#include <sstream>
#include "fake_cpp11.hpp"
void do_next_sequence();
namespace {
void check_arg(const std::string &value, const std::string &name) {
if (value.empty())
throw std::runtime_error("missing argument for \"" + name + "\"");
}
}
private:
const std::string &connection_url;
const std::string &entity;
int message_count;
bool closed;
public:
session_receiver(const std::string &c, const std::string &e,
const char *sid) : connection_url(c), entity(e), message_count(0), closed(false), read_timeout(5000), last_read(0), container(0) {
if (sid)
session_identifier = std::string(sid);
}
message_count = 0;
closed = false;
c.
connect(connection_url, connection_options().handler(*
this));
container = &c;
}
sb_filter_map.
put(key, session_identifier);
receiver = connection.open_receiver(entity, receiver_options().source(source_options().filters(sb_filter_map)));
container->
schedule(read_timeout, [
this]() { this->process_timeout(); });
}
if (closed) return;
proton::value actual_session_id = r.source().filters().
get(
"com.microsoft:session-filter");
std::cout << "receiving messages with session identifier \"" << actual_session_id
<< "\" from queue " << entity << std::endl;
}
message_count++;
std::cout << " received message: " << m.body() << std::endl;
}
void process_timeout() {
if (now >= deadline) {
closed = true;
if (message_count)
do_next_sequence();
else
std::cout << "Done. No more messages." << std::endl;
} else {
container->
schedule(next, [
this]() { this->process_timeout(); });
}
}
};
private:
const std::string &connection_url;
const std::string &entity;
int msg_count;
int total;
int accepts;
public:
session_sender(const std::string &c, const std::string &e) : connection_url(c), entity(e),
msg_count(0), total(7), accepts(0) {}
c.
open_sender(connection_url +
"/" + entity, sender_options(), connection_options().handler(*
this));
}
std::string gid;
for (; msg_count < total && s.
credit() > 0; msg_count++) {
switch (msg_count) {
case 0: gid = "red"; break;
case 1: gid = "green"; break;
case 2: gid = "blue"; break;
case 3: gid = "red"; break;
case 4: gid = "black"; break;
case 5: gid = "blue"; break;
case 6: gid = "yellow"; break;
}
std::ostringstream mbody;
mbody << "message " << msg_count << " in service bus session \"" << gid << "\"";
m.group_id(gid);
std::cout << " sent message: " << m.body() << std::endl;
}
}
send_remaining_messages(s);
}
accepts++;
if (accepts == total) {
t.sender().close();
t.sender().connection().close();
do_next_sequence();
}
}
};
private:
int sequence_no;
session_sender snd;
session_receiver rcv_red, rcv_green, rcv_null;
public:
static sequence *the_sequence;
sequence (const std::string &c, const std::string &e) :
container(0), sequence_no(0),
snd(c, e), rcv_red(c, e, "red"), rcv_green(c, e, "green"), rcv_null(c, e, NULL) {
the_sequence = this;
}
container = &c;
next_sequence();
}
void next_sequence() {
switch (sequence_no++) {
case 0: snd.run(*container); break;
case 1: rcv_green.run(*container); break;
case 2: rcv_red.run(*container); break;
default: rcv_null.run(*container); break;
}
}
};
sequence *sequence::the_sequence = NULL;
void do_next_sequence() { sequence::the_sequence->next_sequence(); }
int main(int argc, char **argv) {
std::string sb_namespace;
std::string sb_key_name;
std::string sb_key;
std::string sb_entity;
example::options opts(argc, argv);
opts.add_value(sb_namespace, 'n', "namespace", "Service Bus full namespace", "NAMESPACE");
opts.add_value(sb_key_name, 'p', "policy", "policy name that specifies access rights (key name)", "POLICY");
opts.add_value(sb_key, 'k', "key", "secret key for the policy", "key");
opts.add_value(sb_entity, 'e', "entity", "entity path (queue name)", "ENTITY");
try {
opts.parse();
check_arg(sb_namespace, "namespace");
check_arg(sb_key_name, "policy");
check_arg(sb_key, "key");
check_arg(sb_entity, "entity");
std::string connection_string("amqps://" + sb_key_name + ":" + sb_key + "@" + sb_namespace);
sequence seq(connection_string, sb_entity);
return 0;
} catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
}
return 1;
}
Options for creating a connection.
Definition: connection_options.hpp:66
A connection to a remote AMQP peer.
Definition: connection.hpp:44
void close()
Close the endpoint.
A top-level container of connections, sessions, and links.
Definition: container.hpp:50
void run()
Run the container in the current thread.
void schedule(duration dur, work fn)
Schedule fn for execution after a duration.
returned< connection > connect(const std::string &conn_url, const connection_options &conn_opts)
Connect to conn_url and send an open request to the remote peer.
returned< sender > open_sender(const std::string &addr_url)
Open a connection and sender for addr_url.
A received message.
Definition: delivery.hpp:39
A span of time in milliseconds.
Definition: duration.hpp:39
void close()
Close the endpoint.
int credit() const
Credit available on the link.
class connection connection() const
The connection that owns this link.
void put(const K &k, const T &v)
Put a map entry for key k.
An AMQP message.
Definition: message.hpp:50
A handler for Proton messaging events.
Definition: messaging_handler.hpp:63
virtual void on_connection_open(connection &)
The remote peer opened the connection.
virtual void on_tracker_accept(tracker &)
The receiving peer accepted a transfer.
virtual void on_message(delivery &, message &)
A message is received.
virtual void on_receiver_open(receiver &)
The remote peer opened the link.
virtual void on_sendable(sender &)
A message can be sent.
Options for creating a receiver.
Definition: receiver_options.hpp:56
A channel for receiving messages.
Definition: receiver.hpp:41
Options for creating a sender.
Definition: sender_options.hpp:57
A channel for sending messages.
Definition: sender.hpp:40
tracker send(const message &m)
Send a message on the sender.
Options for creating a source node for a sender or receiver.
Definition: source_options.hpp:44
A string that represents the AMQP symbol type.
Definition: symbol.hpp:35
A 64-bit timestamp in milliseconds since the Unix epoch.
Definition: timestamp.hpp:35
static timestamp now()
The current wall-clock time.
A tracker for a sent message.
Definition: tracker.hpp:40
A holder for any AMQP value, simple or complex.
Definition: value.hpp:57
T get(const value &v)
Get a contained value of type T.
Definition: value.hpp:126
A connection to a remote AMQP peer.
Options for creating a connection.
A top-level container of connections, sessions, and links.
A handler for Proton messaging events.
Options for creating a receiver.
A channel for sending messages.
Options for creating a sender.
Options for creating a source node for a sender or receiver.
A tracker for a sent message.
Unsettled API - A context for thread-safe execution of work.