Qpid Proton C API 0.39.0
 
Loading...
Searching...
No Matches
broker.c

A simple multithreaded broker that works with the send.c and receive.c examples.

A simple multithreaded broker that works with the send.c and receive.c examples.

/*
* 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.
*/
#include "thread.h"
#include <proton/engine.h>
#include <proton/netaddr.h>
#include <proton/object.h>
#include <proton/sasl.h>
#include <proton/ssl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
/* The ssl-certs subdir must be in the current directory for an ssl-enabled broker */
#define SSL_FILE(NAME) "ssl-certs/" NAME
#define SSL_PW "tserverpw"
/* Windows vs. OpenSSL certificates */
#if defined(_WIN32)
# define CERTIFICATE(NAME) SSL_FILE(NAME "-certificate.p12")
# define SET_CREDENTIALS(DOMAIN, NAME) \
pn_ssl_domain_set_credentials(DOMAIN, SSL_FILE(NAME "-full.p12"), "", SSL_PW)
#else
# define CERTIFICATE(NAME) SSL_FILE(NAME "-certificate.pem")
# define SET_CREDENTIALS(DOMAIN, NAME) \
pn_ssl_domain_set_credentials(DOMAIN, CERTIFICATE(NAME), SSL_FILE(NAME "-private-key.pem"), SSL_PW)
#endif
/* Simple re-sizable vector that acts as a queue */
#define VEC(T) struct { T* data; size_t len, cap; }
#define VEC_INIT(V) \
do { \
void **vp = (void**)&V.data; \
V.len = 0; \
V.cap = 16; \
*vp = malloc(V.cap * sizeof(*V.data)); \
} while(0)
#define VEC_FINAL(V) free(V.data)
#define VEC_PUSH(V, X) \
do { \
if (V.len == V.cap) { \
void **vp = (void**)&V.data; \
V.cap *= 2; \
*vp = realloc(V.data, V.cap * sizeof(*V.data)); \
} \
V.data[V.len++] = X; \
} while(0) \
#define VEC_POP(V) \
do { \
if (V.len > 0) \
memmove(V.data, V.data+1, (--V.len)*sizeof(*V.data)); \
} while(0)
/* Simple thread-safe queue implementation */
typedef struct queue_t {
pthread_mutex_t lock;
char *name;
VEC(pn_rwbytes_t) messages; /* Messages on the queue_t */
VEC(pn_connection_t*) waiting; /* Connections waiting to send messages from this queue */
struct queue_t *next; /* Next queue in chain */
size_t sent; /* Count of messages sent, used as delivery tag */
} queue_t;
static void queue_init(queue_t *q, const char* name, queue_t *next) {
pthread_mutex_init(&q->lock, NULL);
q->name = (char*)malloc(strlen(name)+1);
memcpy(q->name, name, strlen(name)+1);
VEC_INIT(q->messages);
VEC_INIT(q->waiting);
q->next = next;
q->sent = 0;
}
static void queue_destroy(queue_t *q) {
size_t i;
pthread_mutex_destroy(&q->lock);
for (i = 0; i < q->messages.len; ++i)
free(q->messages.data[i].start);
VEC_FINAL(q->messages);
for (i = 0; i < q->waiting.len; ++i)
pn_decref(q->waiting.data[i]);
VEC_FINAL(q->waiting);
free(q->name);
}
/* Send a message on s, or record s as waiting if there are no messages to send.
Called in s dispatch loop, assumes s has credit.
*/
static void queue_send(queue_t *q, pn_link_t *s) {
pn_rwbytes_t m = { 0 };
size_t tag = 0;
pthread_mutex_lock(&q->lock);
if (q->messages.len == 0) { /* Empty, record connection as waiting */
/* Record connection for wake-up if not already on the list. */
size_t i = 0;
for (; i < q->waiting.len && q->waiting.data[i] != c; ++i)
;
if (i == q->waiting.len) {
VEC_PUSH(q->waiting, c);
}
} else {
m = q->messages.data[0];
VEC_POP(q->messages);
tag = ++q->sent;
}
pthread_mutex_unlock(&q->lock);
if (m.start) {
pn_delivery_t *d = pn_delivery(s, pn_dtag((char*)&tag, sizeof(tag)));
pn_link_send(s, m.start, m.size);
pn_delivery_settle(d); /* Pre-settled: unreliable, there will be no ack/ */
free(m.start);
}
}
/* Use the connection context pointer as a boolean flag to indicate we need to check queues */
void set_check_queues(pn_connection_t *c, bool check) {
pn_connection_set_context(c, (void*)check);
}
bool get_check_queues(pn_connection_t *c) {
return (bool)pn_connection_get_context(c);
}
/* Use a buffer per link to accumulate message data - message can arrive in multiple deliveries,
and the broker can receive messages on many concurrently. */
pn_rwbytes_t *message_buffer(pn_link_t *l) {
pn_link_set_context(l, calloc(1, sizeof(pn_rwbytes_t)));
}
}
/* Put a message on the queue, called in receiver dispatch loop.
If the queue was previously empty, notify waiting senders.
*/
static void queue_receive(pn_proactor_t *d, queue_t *q, pn_rwbytes_t m) {
pthread_mutex_lock(&q->lock);
VEC_PUSH(q->messages, m);
if (q->messages.len == 1) { /* Was empty, notify waiting connections */
size_t i;
for (i = 0; i < q->waiting.len; ++i) {
pn_connection_t *c = q->waiting.data[i];
set_check_queues(c, true);
pn_connection_wake(c); /* Wake the connection */
}
q->waiting.len = 0;
}
pthread_mutex_unlock(&q->lock);
}
/* Thread safe set of queues */
typedef struct queues_t {
pthread_mutex_t lock;
queue_t *queues;
size_t sent;
} queues_t;
void queues_init(queues_t *qs) {
pthread_mutex_init(&qs->lock, NULL);
qs->queues = NULL;
}
void queues_destroy(queues_t *qs) {
while (qs->queues) {
queue_t *q = qs->queues;
qs->queues = qs->queues->next;
queue_destroy(q);
free(q);
}
pthread_mutex_destroy(&qs->lock);
}
queue_t* queues_get(queues_t *qs, const char* name) {
queue_t *q;
pthread_mutex_lock(&qs->lock);
for (q = qs->queues; q && strcmp(q->name, name) != 0; q = q->next)
;
if (!q) {
q = (queue_t*)malloc(sizeof(queue_t));
queue_init(q, name, qs->queues);
qs->queues = q;
}
pthread_mutex_unlock(&qs->lock);
return q;
}
/* The broker implementation */
typedef struct broker_t {
pn_proactor_t *proactor;
size_t threads;
const char *container_id; /* AMQP container-id */
queues_t queues;
pn_ssl_domain_t *ssl_domain;
} broker_t;
void broker_stop(broker_t *b) {
/* Interrupt the proactor to stop the working threads. */
pn_proactor_interrupt(b->proactor);
}
/* Try to send if link is sender and has credit */
static void link_send(broker_t *b, pn_link_t *s) {
if (pn_link_is_sender(s) && pn_link_credit(s) > 0) {
const char *qname = pn_terminus_get_address(pn_link_source(s));
queue_t *q = queues_get(&b->queues, qname);
queue_send(q, s);
}
}
static void queue_unsub(queue_t *q, pn_connection_t *c) {
size_t i;
pthread_mutex_lock(&q->lock);
for (i = 0; i < q->waiting.len; ++i) {
if (q->waiting.data[i] == c){
q->waiting.data[i] = q->waiting.data[0]; /* save old [0] */
VEC_POP(q->waiting);
break;
}
}
pthread_mutex_unlock(&q->lock);
}
/* Unsubscribe from the queue of interest to this link. */
static void link_unsub(broker_t *b, pn_link_t *s) {
const char *qname = pn_terminus_get_address(pn_link_source(s));
if (qname) {
queue_t *q = queues_get(&b->queues, qname);
}
}
}
/* Called in connection's event loop when a connection is woken for messages.*/
static void connection_unsub(broker_t *b, pn_connection_t *c) {
for (l = pn_link_head(c, 0); l != NULL; l = pn_link_next(l, 0))
link_unsub(b, l);
}
static void session_unsub(broker_t *b, pn_session_t *ssn) {
for (l = pn_link_head(c, 0); l != NULL; l = pn_link_next(l, 0)) {
if (pn_link_session(l) == ssn)
link_unsub(b, l);
}
}
static void check_condition(pn_event_t *e, pn_condition_t *cond) {
if (pn_condition_is_set(cond)) {
printf("%s: %s: %s\n", pn_event_type_name(pn_event_type(e)),
}
}
const int WINDOW=5; /* Very small incoming credit window, to show flow control in action */
static bool handle(broker_t* b, pn_event_t* e, pn_connection_t *c) {
switch (pn_event_type(e)) {
char port[PN_MAX_ADDR]; /* Get the listening port */
pn_netaddr_host_port(pn_listener_addr(pn_event_listener(e)), NULL, 0, port, sizeof(port));
printf("listening on %s\n", port);
fflush(stdout);
break;
}
/* Configure a transport to allow SSL and SASL connections. See ssl_domain setup in main() */
pn_transport_set_server(t); /* Must call before pn_sasl() */
pn_sasl_allowed_mechs(pn_sasl(t), "ANONYMOUS");
if (b->ssl_domain) {
pn_ssl_init(pn_ssl(t), b->ssl_domain, NULL);
pn_transport_require_encryption(t, false); /* Must call this after pn_ssl_init */
}
break;
}
pn_connection_set_container(c, b->container_id);
break;
pn_connection_open(pn_event_connection(e)); /* Complete the open */
break;
}
if (get_check_queues(c)) {
set_check_queues(c, false);
for (l = pn_link_head(c, flags); l != NULL; l = pn_link_next(l, flags))
link_send(b, l);
}
break;
}
break;
}
} else {
pn_link_flow(l, WINDOW);
}
break;
}
case PN_LINK_FLOW: {
link_send(b, pn_event_link(e));
break;
}
case PN_LINK_FINAL: {
if (buf) {
free(buf->start);
free(buf);
}
break;
}
case PN_DELIVERY: { /* Incoming message data */
size_t size = pn_delivery_pending(d);
pn_rwbytes_t* m = message_buffer(l); /* Append data to incoming message buffer */
ssize_t recv;
m->size += size;
m->start = (char*)realloc(m->start, m->size);
recv = pn_link_recv(l, m->start, m->size);
if (recv == PN_ABORTED) { /* */
printf("Message aborted\n");
fflush(stdout);
m->size = 0; /* Forget the data we accumulated */
pn_delivery_settle(d); /* Free the delivery so we can receive the next message */
pn_link_flow(l, WINDOW - pn_link_credit(l)); /* Replace credit for the aborted message */
} else if (recv < 0 && recv != PN_EOS) { /* Unexpected error */
pn_condition_format(pn_link_condition(l), "broker", "PN_DELIVERY error: %s", pn_code((int)recv));
pn_link_close(l); /* Unexpected error, close the link */
} else if (!pn_delivery_partial(d)) { /* Message is complete */
const char *qname = pn_terminus_get_address(pn_link_target(l));
queue_receive(b->proactor, queues_get(&b->queues, qname), *m);
*m = pn_rwbytes_null; /* Reset the buffer for the next message*/
pn_link_flow(l, WINDOW - pn_link_credit(l));
}
}
break;
}
check_condition(e, pn_transport_condition(pn_event_transport(e)));
connection_unsub(b, pn_event_connection(e));
break;
break;
session_unsub(b, pn_event_session(e));
break;
check_condition(e, pn_link_remote_condition(pn_event_link(e)));
link_unsub(b, pn_event_link(e));
break;
check_condition(e, pn_listener_condition(pn_event_listener(e)));
broker_stop(b);
break;
case PN_PROACTOR_INACTIVE: /* listener and all connections closed */
broker_stop(b);
break;
pn_proactor_interrupt(b->proactor); /* Pass along the interrupt to the other threads */
return false;
default:
break;
}
return true;
}
static void* broker_thread(void *void_broker) {
broker_t *b = (broker_t*)void_broker;
bool finished = false;
do {
pn_event_batch_t *events = pn_proactor_wait(b->proactor);
while ((e = pn_event_batch_next(events))) {
if (!handle(b, e, connection)) finished = true;
}
pn_proactor_done(b->proactor, events);
} while(!finished);
return NULL;
}
int main(int argc, char **argv) {
const char *host = (argc > 1) ? argv[1] : "";
const char *port = (argc > 2) ? argv[2] : "amqp";
int err;
broker_t b = {0};
b.proactor = pn_proactor();
queues_init(&b.queues);
b.container_id = argv[0];
b.threads = 4;
err = SET_CREDENTIALS(b.ssl_domain, "tserver");
if (err) {
printf("Failed to set up server certificate: %s, private key: %s\n", CERTIFICATE("tserver"), SSL_FILE("tserver-private-key.pem"));
}
{
/* Listen on addr */
char addr[PN_MAX_ADDR];
pn_proactor_addr(addr, sizeof(addr), host, port);
pn_proactor_listen(b.proactor, pn_listener(), addr, 16);
}
{
/* Start n-1 threads */
pthread_t* threads = (pthread_t*)calloc(sizeof(pthread_t), b.threads);
size_t i;
for (i = 0; i < b.threads-1; ++i) {
pthread_create(&threads[i], NULL, broker_thread, &b);
}
broker_thread(&b); /* Use the main thread too. */
/* Join the other threads */
for (i = 0; i < b.threads-1; ++i) {
pthread_join(threads[i], NULL);
}
pn_proactor_free(b.proactor);
free(threads);
pn_ssl_domain_free(b.ssl_domain);
return 0;
}
}
A non-const byte buffer.
Definition: types.h:235
bool pn_condition_is_set(pn_condition_t *condition)
Returns true if the condition object is holding some information, i.e.
struct pn_condition_t pn_condition_t
An AMQP Condition object.
Definition: condition.h:64
int pn_condition_format(pn_condition_t *, const char *name, const char *fmt,...)
Set the name and printf-style formatted description.
const char * pn_condition_get_description(pn_condition_t *condition)
Gets the description associated with the exceptional condition.
const char * pn_condition_get_name(pn_condition_t *condition)
Returns the name associated with the exceptional condition, or NULL if there is no conditional inform...
#define PN_LOCAL_ACTIVE
The local endpoint state is active.
Definition: connection.h:55
void pn_connection_open(pn_connection_t *connection)
Open a connection.
struct pn_connection_t pn_connection_t
An AMQP Connection object.
Definition: types.h:285
void pn_connection_close(pn_connection_t *connection)
Close a connection.
void pn_connection_set_container(pn_connection_t *connection, const char *container)
Set the AMQP Container name advertised by a connection object.
pn_condition_t * pn_connection_remote_condition(pn_connection_t *connection)
Get the remote condition associated with the connection endpoint.
void * pn_connection_get_context(pn_connection_t *connection)
Get the application context that is associated with a connection object.
void pn_connection_set_context(pn_connection_t *connection, void *context)
Set a new application context for a connection object.
#define PN_REMOTE_ACTIVE
The remote endpoint state is active.
Definition: connection.h:70
bool pn_delivery_readable(pn_delivery_t *delivery)
Check if a delivery is readable.
size_t pn_delivery_pending(pn_delivery_t *delivery)
Get the amount of pending message data for a delivery.
bool pn_delivery_partial(pn_delivery_t *delivery)
Check if a delivery only has partial message data.
void pn_delivery_update(pn_delivery_t *delivery, uint64_t state)
Update the disposition of a delivery.
pn_link_t * pn_delivery_link(pn_delivery_t *delivery)
Get the parent link for a delivery object.
pn_delivery_t * pn_delivery(pn_link_t *link, pn_delivery_tag_t tag)
Create a delivery on a link.
void pn_delivery_settle(pn_delivery_t *delivery)
Settle a delivery.
#define PN_ACCEPTED
The PN_ACCEPTED delivery state is a terminal state indicating that the delivery was successfully proc...
Definition: disposition.h:66
struct pn_delivery_t pn_delivery_t
An AMQP Delivery object.
Definition: types.h:405
pn_delivery_tag_t pn_dtag(const char *bytes, size_t size)
Construct a delivery tag.
#define PN_ABORTED
Delivery aborted error.
Definition: error.h:57
#define PN_EOS
End of stream.
Definition: error.h:47
const char * pn_code(int code)
Get the name of the error code.
pn_delivery_t * pn_event_delivery(pn_event_t *event)
Get the delivery associated with an event.
pn_session_t * pn_event_session(pn_event_t *event)
Get the session associated with an event.
const char * pn_event_type_name(pn_event_type_t type)
Get a human readable name for an event type.
pn_link_t * pn_event_link(pn_event_t *event)
Get the link associated with an event.
struct pn_event_t pn_event_t
Notification of a state change in the protocol engine.
Definition: event.h:75
pn_connection_t * pn_event_connection(pn_event_t *event)
Get the connection associated with an event.
pn_event_type_t pn_event_type(pn_event_t *event)
Get the type of an event.
pn_transport_t * pn_event_transport(pn_event_t *event)
Get the transport associated with an event.
@ PN_LISTENER_ACCEPT
Indicates the listener has an incoming connection, call pn_listener_accept2() to accept it.
Definition: event.h:316
@ PN_LINK_REMOTE_CLOSE
The remote endpoint has closed the link.
Definition: event.h:223
@ PN_CONNECTION_REMOTE_CLOSE
The remote endpoint has closed the connection.
Definition: event.h:149
@ PN_CONNECTION_INIT
The connection has been created.
Definition: event.h:113
@ PN_TRANSPORT_CLOSED
Indicates that the both the head and tail of the transport are closed.
Definition: event.h:295
@ PN_CONNECTION_WAKE
pn_connection_wake() was called.
Definition: event.h:309
@ PN_SESSION_REMOTE_CLOSE
The remote endpoint has closed the session.
Definition: event.h:186
@ PN_LINK_REMOTE_OPEN
The remote endpoint has opened the link.
Definition: event.h:211
@ PN_LINK_FLOW
The flow control state for a link has changed.
Definition: event.h:241
@ PN_LISTENER_OPEN
The listener is listening.
Definition: event.h:350
@ PN_DELIVERY
A delivery has been created or updated.
Definition: event.h:254
@ PN_PROACTOR_INACTIVE
The proactor has become inactive: all listeners and connections were closed and the timeout (if set) ...
Definition: event.h:344
@ PN_CONNECTION_REMOTE_OPEN
The remote endpoint has opened the connection.
Definition: event.h:137
@ PN_LISTENER_CLOSE
Indicates the listener has closed.
Definition: event.h:322
@ PN_LINK_FINAL
The link has been freed and any outstanding processing has been completed.
Definition: event.h:248
@ PN_PROACTOR_INTERRUPT
Indicates pn_proactor_interrupt() was called to interrupt a proactor thread.
Definition: event.h:328
@ PN_SESSION_REMOTE_OPEN
The remote endpoint has opened the session.
Definition: event.h:174
PNP_EXTERN void pn_listener_accept2(pn_listener_t *listener, pn_connection_t *connection, pn_transport_t *transport)
Accept an incoming connection request using transport and connection, which can be configured before ...
PNP_EXTERN pn_listener_t * pn_listener(void)
Create a listener to pass to pn_proactor_listen()
PNP_EXTERN pn_listener_t * pn_event_listener(pn_event_t *event)
Return the listener associated with an event.
PNP_EXTERN pn_condition_t * pn_listener_condition(pn_listener_t *l)
Get the error condition for a listener.
PNP_EXTERN pn_proactor_t * pn_proactor(void)
Create a proactor.
PNP_EXTERN pn_event_t * pn_event_batch_next(pn_event_batch_t *batch)
Remove the next event from the batch and return it.
PNP_EXTERN void pn_connection_wake(pn_connection_t *connection)
Return a PN_CONNECTION_WAKE event for connection as soon as possible.
#define PN_MAX_ADDR
Size of buffer that can hold the largest connection or listening address.
Definition: proactor.h:74
PNP_EXTERN pn_event_batch_t * pn_proactor_wait(pn_proactor_t *proactor)
Wait until there are Proactor events to handle.
PNP_EXTERN void pn_proactor_free(pn_proactor_t *proactor)
Free the proactor.
struct pn_event_batch_t pn_event_batch_t
A batch of events that must be handled in sequence.
Definition: types.h:462
PNP_EXTERN pn_connection_t * pn_event_batch_connection(pn_event_batch_t *batch)
Query the batch for the subject of the batch.
PNP_EXTERN int pn_netaddr_host_port(const pn_netaddr_t *na, char *host, size_t hlen, char *port, size_t plen)
Get the host and port name from na as separate strings.
PNP_EXTERN void pn_proactor_interrupt(pn_proactor_t *proactor)
Return a PN_PROACTOR_INTERRUPT event as soon as possible.
struct pn_proactor_t pn_proactor_t
A harness for multithreaded IO.
Definition: types.h:442
PNP_EXTERN void pn_proactor_listen(pn_proactor_t *proactor, pn_listener_t *listener, const char *addr, int backlog)
Start listening for incoming connections.
PNP_EXTERN void pn_proactor_done(pn_proactor_t *proactor, pn_event_batch_t *events)
Call when finished handling a batch of events.
PNP_EXTERN const pn_netaddr_t * pn_listener_addr(pn_listener_t *l)
Get the listening addresses of a listener.
PNP_EXTERN int pn_proactor_addr(char *addr, size_t size, const char *host, const char *port)
Format a host:port address string for pn_proactor_connect() or pn_proactor_listen()
pn_sasl_t * pn_sasl(pn_transport_t *transport)
Construct an Authentication and Security Layer object.
void pn_sasl_allowed_mechs(pn_sasl_t *sasl, const char *mechs)
SASL mechanisms that are to be considered for authentication.
pn_condition_t * pn_session_remote_condition(pn_session_t *session)
Get the remote condition associated with the session endpoint.
void pn_session_free(pn_session_t *session)
Free a session object.
void pn_session_close(pn_session_t *session)
Close a session.
pn_connection_t * pn_session_connection(pn_session_t *session)
Get the parent connection for a session object.
void pn_session_open(pn_session_t *session)
Open a session.
struct pn_session_t pn_session_t
An AMQP Session object.
Definition: types.h:296
int pn_ssl_init(pn_ssl_t *ssl, pn_ssl_domain_t *domain, const char *session_id)
Initialize an SSL session.
void pn_ssl_domain_free(pn_ssl_domain_t *domain)
Release an SSL configuration domain.
pn_ssl_domain_t * pn_ssl_domain(pn_ssl_mode_t mode)
Create an SSL configuration domain.
pn_ssl_t * pn_ssl(pn_transport_t *transport)
Create a new SSL session object associated with a transport.
struct pn_ssl_domain_t pn_ssl_domain_t
API for using SSL with the Transport Layer.
Definition: ssl.h:80
@ PN_SSL_MODE_SERVER
Local connection endpoint is an SSL server.
Definition: ssl.h:92
int pn_terminus_set_address(pn_terminus_t *terminus, const char *address)
Set the address of a terminus object.
const char * pn_terminus_get_address(pn_terminus_t *terminus)
Get the address of a terminus object.
pn_transport_t * pn_transport(void)
Factory for creating a transport.
struct pn_transport_t pn_transport_t
A network channel supporting an AMQP connection.
Definition: types.h:435
void pn_transport_set_server(pn_transport_t *transport)
Configure a transport as a server.
void pn_transport_require_encryption(pn_transport_t *transport, bool required)
Set whether a non encrypted transport connection is allowed.
pn_condition_t * pn_transport_condition(pn_transport_t *transport)
Get additional information about the condition of the transport.
Unsettled API - A listener for incoming connections.
Unsettled API - The network address of a proactor transport.
Unsettled API - An API for multithreaded IO.
SASL secure transport layer.
SSL secure transport layer.
A network channel supporting an AMQP connection.