broker.c

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/listener.h>
#include <proton/netaddr.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
#include <proton/ssl.h>
#include <proton/transport.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. */
pn_connection_t *c = pn_session_connection(pn_link_session(s));
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_link_advance(s);
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) {
if (!pn_link_get_context(l)) {
pn_link_set_context(l, calloc(1, sizeof(pn_rwbytes_t)));
}
return (pn_rwbytes_t*)pn_link_get_context(l);
}
/* 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;
bool finished;
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) {
if (pn_link_is_sender(s)) {
const char *qname = pn_terminus_get_address(pn_link_source(s));
if (qname) {
queue_t *q = queues_get(&b->queues, qname);
queue_unsub(q, pn_session_connection(pn_link_session(s)));
}
}
}
/* Called in connection's event loop when a connection is woken for messages.*/
static void connection_unsub(broker_t *b, pn_connection_t *c) {
pn_link_t *l;
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) {
pn_connection_t *c = pn_session_connection(ssn);
pn_link_t *l;
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)) {
fprintf(stderr, "%s: %s: %s\n", pn_event_type_name(pn_event_type(e)),
pn_condition_get_name(cond), pn_condition_get_description(cond));
}
}
const int WINDOW=5; /* Very small incoming credit window, to show flow control in action */
static void handle(broker_t* b, pn_event_t* e) {
pn_connection_t *c = pn_event_connection(e);
switch (pn_event_type(e)) {
case PN_LISTENER_OPEN: {
char port[256]; /* 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;
}
case PN_LISTENER_ACCEPT: {
/* Configure a transport to allow SSL and SASL connections. See ssl_domain setup in main() */
pn_transport_t *t = pn_transport();
pn_transport_require_auth(t, false);
pn_sasl_allowed_mechs(pn_sasl(t), "ANONYMOUS");
if (b->ssl_domain) {
pn_ssl_init(pn_ssl(t), b->ssl_domain, NULL);
}
pn_listener_accept2(pn_event_listener(e), NULL, t);
break;
}
case PN_CONNECTION_INIT:
pn_connection_set_container(c, b->container_id);
break;
case PN_CONNECTION_REMOTE_OPEN: {
pn_connection_open(pn_event_connection(e)); /* Complete the open */
break;
}
case PN_CONNECTION_WAKE: {
if (get_check_queues(c)) {
int flags = PN_LOCAL_ACTIVE&PN_REMOTE_ACTIVE;
pn_link_t *l;
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;
}
case PN_SESSION_REMOTE_OPEN: {
pn_session_open(pn_event_session(e));
break;
}
case PN_LINK_REMOTE_OPEN: {
pn_link_t *l = pn_event_link(e);
if (pn_link_is_sender(l)) {
const char *source = pn_terminus_get_address(pn_link_remote_source(l));
pn_terminus_set_address(pn_link_source(l), source);
} else {
const char* target = pn_terminus_get_address(pn_link_remote_target(l));
pn_terminus_set_address(pn_link_target(l), target);
pn_link_flow(l, WINDOW);
}
pn_link_open(l);
break;
}
case PN_LINK_FLOW: {
link_send(b, pn_event_link(e));
break;
}
case PN_LINK_FINAL: {
pn_rwbytes_t *buf = (pn_rwbytes_t*)pn_link_get_context(pn_event_link(e));
if (buf) {
free(buf->start);
free(buf);
}
break;
}
case PN_DELIVERY: { /* Incoming message data */
pn_delivery_t *d = pn_event_delivery(e);
if (pn_delivery_readable(d)) {
pn_link_t *l = pn_delivery_link(d);
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) { /* */
fprintf(stderr, "Message aborted\n");
fflush(stderr);
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_delivery_update(d, PN_ACCEPTED);
pn_delivery_settle(d);
pn_link_flow(l, WINDOW - pn_link_credit(l));
}
}
break;
}
case PN_TRANSPORT_CLOSED:
check_condition(e, pn_transport_condition(pn_event_transport(e)));
connection_unsub(b, pn_event_connection(e));
break;
case PN_CONNECTION_REMOTE_CLOSE:
check_condition(e, pn_connection_remote_condition(pn_event_connection(e)));
pn_connection_close(pn_event_connection(e));
break;
case PN_SESSION_REMOTE_CLOSE:
check_condition(e, pn_session_remote_condition(pn_event_session(e)));
session_unsub(b, pn_event_session(e));
pn_session_close(pn_event_session(e));
pn_session_free(pn_event_session(e));
break;
case PN_LINK_REMOTE_CLOSE:
check_condition(e, pn_link_remote_condition(pn_event_link(e)));
link_unsub(b, pn_event_link(e));
pn_link_close(pn_event_link(e));
pn_link_free(pn_event_link(e));
break;
case PN_LISTENER_CLOSE:
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;
case PN_PROACTOR_INTERRUPT:
b->finished = true;
pn_proactor_interrupt(b->proactor); /* Pass along the interrupt to the other threads */
break;
default:
break;
}
}
static void* broker_thread(void *void_broker) {
broker_t *b = (broker_t*)void_broker;
do {
pn_event_batch_t *events = pn_proactor_wait(b->proactor);
pn_event_t *e;
while ((e = pn_event_batch_next(events))) {
handle(b, e);
}
pn_proactor_done(b->proactor, events);
} while(!b->finished);
return NULL;
}
int main(int argc, char **argv) {
const char *host = (argc > 1) ? argv[1] : "";
const char *port = (argc > 2) ? argv[2] : "amqp";
broker_t b = {0};
b.proactor = pn_proactor();
queues_init(&b.queues);
b.container_id = argv[0];
b.threads = 4;
b.ssl_domain = pn_ssl_domain(PN_SSL_MODE_SERVER);
SET_CREDENTIALS(b.ssl_domain, "tserver");
pn_ssl_domain_allow_unsecured_client(b.ssl_domain); /* Allow SSL and plain connections */
{
/* 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;
}
}
pn_rwbytes_t
A non-const byte buffer.
Definition: types.h:231
pn_condition_get_name
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...
pn_condition_is_set
bool pn_condition_is_set(pn_condition_t *condition)
Returns true if the condition object is holding some information, i.e.
pn_condition_t
struct pn_condition_t pn_condition_t
An AMQP Condition object.
Definition: condition.h:64
pn_condition_format
int pn_condition_format(pn_condition_t *, const char *name, const char *fmt,...)
Set the name and printf-style formatted description.
pn_condition_get_description
const char * pn_condition_get_description(pn_condition_t *condition)
Gets the description associated with the exceptional condition.
PN_LOCAL_ACTIVE
#define PN_LOCAL_ACTIVE
The local endpoint state is active.
Definition: connection.h:55
pn_connection_open
void pn_connection_open(pn_connection_t *connection)
Open a connection.
pn_connection_get_context
void * pn_connection_get_context(pn_connection_t *connection)
Deprecated - Use pn_connection_attachments().
pn_connection_remote_condition
pn_condition_t * pn_connection_remote_condition(pn_connection_t *connection)
Get the remote condition associated with the connection endpoint.
pn_connection_t
struct pn_connection_t pn_connection_t
An AMQP Connection object.
Definition: types.h:281
pn_connection_close
void pn_connection_close(pn_connection_t *connection)
Close a connection.
pn_connection_set_container
void pn_connection_set_container(pn_connection_t *connection, const char *container)
Set the AMQP Container name advertised by a connection object.
pn_connection_set_context
void pn_connection_set_context(pn_connection_t *connection, void *context)
Deprecated - Use pn_connection_attachments().
PN_REMOTE_ACTIVE
#define PN_REMOTE_ACTIVE
The remote endpoint state is active.
Definition: connection.h:70
pn_delivery_readable
bool pn_delivery_readable(pn_delivery_t *delivery)
Check if a delivery is readable.
pn_delivery_pending
size_t pn_delivery_pending(pn_delivery_t *delivery)
Get the amount of pending message data for a delivery.
pn_delivery_partial
bool pn_delivery_partial(pn_delivery_t *delivery)
Check if a delivery only has partial message data.
pn_delivery_update
void pn_delivery_update(pn_delivery_t *delivery, uint64_t state)
Update the disposition of a delivery.
pn_delivery
pn_delivery_t * pn_delivery(pn_link_t *link, pn_delivery_tag_t tag)
Create a delivery on a link.
pn_delivery_settle
void pn_delivery_settle(pn_delivery_t *delivery)
Settle a delivery.
PN_ACCEPTED
#define PN_ACCEPTED
The PN_ACCEPTED delivery state is a terminal state indicating that the delivery was successfully proc...
Definition: disposition.h:66
pn_delivery_t
struct pn_delivery_t pn_delivery_t
An AMQP Delivery object.
Definition: types.h:401
pn_delivery_link
pn_link_t * pn_delivery_link(pn_delivery_t *delivery)
Get the parent link for a delivery object.
pn_dtag
pn_delivery_tag_t pn_dtag(const char *bytes, size_t size)
Construct a delivery tag.
PN_ABORTED
#define PN_ABORTED
Delivery aborted error.
Definition: error.h:57
PN_EOS
#define PN_EOS
End of stream.
Definition: error.h:47
pn_code
const char * pn_code(int code)
Get the name of the error code.
pn_event_delivery
pn_delivery_t * pn_event_delivery(pn_event_t *event)
Get the delivery associated with an event.
pn_event_transport
pn_transport_t * pn_event_transport(pn_event_t *event)
Get the transport associated with an event.
pn_event_type_name
const char * pn_event_type_name(pn_event_type_t type)
Get a human readable name for an event type.
pn_event_connection
pn_connection_t * pn_event_connection(pn_event_t *event)
Get the connection associated with an event.
pn_event_batch_t
struct pn_event_batch_t pn_event_batch_t
Unsettled API - A batch of events that must be handled in sequence.
Definition: event.h:547
pn_event_link
pn_link_t * pn_event_link(pn_event_t *event)
Get the link associated with an event.
pn_event_t
struct pn_event_t pn_event_t
Notification of a state change in the protocol engine.
Definition: event.h:75
pn_event_batch_next
pn_event_t * pn_event_batch_next(pn_event_batch_t *batch)
Unsettled API - Remove the next event from the batch and return it.
pn_event_type
pn_event_type_t pn_event_type(pn_event_t *event)
Get the type of an event.
pn_event_session
pn_session_t * pn_event_session(pn_event_t *event)
Get the session associated with an event.
PN_LISTENER_ACCEPT
@ 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
@ PN_LINK_REMOTE_CLOSE
The remote endpoint has closed the link.
Definition: event.h:223
PN_CONNECTION_REMOTE_CLOSE
@ PN_CONNECTION_REMOTE_CLOSE
The remote endpoint has closed the connection.
Definition: event.h:149
PN_CONNECTION_INIT
@ PN_CONNECTION_INIT
The connection has been created.
Definition: event.h:113
PN_TRANSPORT_CLOSED
@ 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
pn_connection_wake() was called.
Definition: event.h:309
PN_SESSION_REMOTE_CLOSE
@ PN_SESSION_REMOTE_CLOSE
The remote endpoint has closed the session.
Definition: event.h:186
PN_LINK_REMOTE_OPEN
@ PN_LINK_REMOTE_OPEN
The remote endpoint has opened the link.
Definition: event.h:211
PN_LINK_FLOW
@ PN_LINK_FLOW
The flow control state for a link has changed.
Definition: event.h:241
PN_LISTENER_OPEN
@ PN_LISTENER_OPEN
The listener is listening.
Definition: event.h:350
PN_DELIVERY
@ PN_DELIVERY
A delivery has been created or updated.
Definition: event.h:254
PN_PROACTOR_INACTIVE
@ 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
@ PN_CONNECTION_REMOTE_OPEN
The remote endpoint has opened the connection.
Definition: event.h:137
PN_LISTENER_CLOSE
@ PN_LISTENER_CLOSE
Indicates the listener has closed.
Definition: event.h:322
PN_LINK_FINAL
@ PN_LINK_FINAL
The link has been freed and any outstanding processing has been completed.
Definition: event.h:248
PN_PROACTOR_INTERRUPT
@ PN_PROACTOR_INTERRUPT
Indicates pn_proactor_interrupt() was called to interrupt a proactor thread.
Definition: event.h:328
PN_SESSION_REMOTE_OPEN
@ PN_SESSION_REMOTE_OPEN
The remote endpoint has opened the session.
Definition: event.h:174
pn_link_recv
ssize_t pn_link_recv(pn_link_t *receiver, char *bytes, size_t n)
Receive message data for the current delivery on a link.
pn_link_source
pn_terminus_t * pn_link_source(pn_link_t *link)
Access the locally defined source definition for a link.
pn_link_set_context
void pn_link_set_context(pn_link_t *link, void *context)
Deprecated - Use pn_link_attachments().
pn_link_close
void pn_link_close(pn_link_t *link)
Close a link.
pn_link_condition
pn_condition_t * pn_link_condition(pn_link_t *link)
Get the local condition associated with a link endpoint.
pn_link_credit
int pn_link_credit(pn_link_t *link)
Get the credit balance for a link.
pn_link_is_sender
bool pn_link_is_sender(pn_link_t *link)
Test if a link is a sender.
pn_link_head
pn_link_t * pn_link_head(pn_connection_t *connection, pn_state_t state)
Retrieve the first link that matches the given state mask.
pn_link_t
struct pn_link_t pn_link_t
An AMQP Link object.
Definition: types.h:311
pn_link_advance
bool pn_link_advance(pn_link_t *link)
Advance the current delivery of a link to the next delivery on the link.
pn_link_get_context
void * pn_link_get_context(pn_link_t *link)
Deprecated - Use pn_link_attachments().
pn_link_remote_condition
pn_condition_t * pn_link_remote_condition(pn_link_t *link)
Get the remote condition associated with a link endpoint.
pn_link_target
pn_terminus_t * pn_link_target(pn_link_t *link)
Access the locally defined target definition for a link.
pn_link_next
pn_link_t * pn_link_next(pn_link_t *link, pn_state_t state)
Retrieve the next link that matches the given state mask.
pn_link_send
ssize_t pn_link_send(pn_link_t *sender, const char *bytes, size_t n)
Send message data for the current delivery on a link.
pn_link_open
void pn_link_open(pn_link_t *link)
Open a link.
pn_link_remote_target
pn_terminus_t * pn_link_remote_target(pn_link_t *link)
Access the remotely defined target definition for a link.
pn_link_session
pn_session_t * pn_link_session(pn_link_t *link)
Get the parent session for a link object.
pn_link_free
void pn_link_free(pn_link_t *link)
Free a link object.
pn_link_remote_source
pn_terminus_t * pn_link_remote_source(pn_link_t *link)
Access the remotely defined source definition for a link.
pn_link_flow
void pn_link_flow(pn_link_t *receiver, int credit)
Grant credit for incoming deliveries on a receiver.
pn_listener_accept2
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 ...
pn_event_listener
PNP_EXTERN pn_listener_t * pn_event_listener(pn_event_t *event)
Return the listener associated with an event.
pn_listener_condition
PNP_EXTERN pn_condition_t * pn_listener_condition(pn_listener_t *l)
Get the error condition for a listener.
pn_listener
PNP_EXTERN pn_listener_t * pn_listener(void)
Create a listener to pass to pn_proactor_listen()
pn_connection_wake
PNP_EXTERN void pn_connection_wake(pn_connection_t *connection)
Return a PN_CONNECTION_WAKE event for connection as soon as possible.
PN_MAX_ADDR
#define PN_MAX_ADDR
Size of buffer that can hold the largest connection or listening address.
Definition: proactor.h:74
pn_proactor_wait
PNP_EXTERN pn_event_batch_t * pn_proactor_wait(pn_proactor_t *proactor)
Wait until there are Proactor events to handle.
pn_proactor_free
PNP_EXTERN void pn_proactor_free(pn_proactor_t *proactor)
Free the proactor.
pn_netaddr_host_port
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.
pn_proactor_interrupt
PNP_EXTERN void pn_proactor_interrupt(pn_proactor_t *proactor)
Return a PN_PROACTOR_INTERRUPT event as soon as possible.
pn_proactor_t
struct pn_proactor_t pn_proactor_t
A harness for multithreaded IO.
Definition: types.h:438
pn_proactor
PNP_EXTERN pn_proactor_t * pn_proactor(void)
Create a proactor.
pn_proactor_listen
PNP_EXTERN void pn_proactor_listen(pn_proactor_t *proactor, pn_listener_t *listener, const char *addr, int backlog)
Start listening for incoming connections.
pn_listener_addr
PNP_EXTERN const pn_netaddr_t * pn_listener_addr(pn_listener_t *l)
Get the listening addresses of a listener.
pn_proactor_done
PNP_EXTERN void pn_proactor_done(pn_proactor_t *proactor, pn_event_batch_t *events)
Call when finished handling a batch of events.
pn_proactor_addr
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_connect2() or pn_proactor_listen2()
pn_sasl_allowed_mechs
void pn_sasl_allowed_mechs(pn_sasl_t *sasl, const char *mechs)
SASL mechanisms that are to be considered for authentication.
pn_sasl
pn_sasl_t * pn_sasl(pn_transport_t *transport)
Construct an Authentication and Security Layer object.
pn_session_free
void pn_session_free(pn_session_t *session)
Free a session object.
pn_session_close
void pn_session_close(pn_session_t *session)
Close a session.
pn_session_open
void pn_session_open(pn_session_t *session)
Open a session.
pn_session_t
struct pn_session_t pn_session_t
An AMQP Session object.
Definition: types.h:292
pn_session_connection
pn_connection_t * pn_session_connection(pn_session_t *session)
Get the parent connection for a session object.
pn_session_remote_condition
pn_condition_t * pn_session_remote_condition(pn_session_t *session)
Get the remote condition associated with the session endpoint.
pn_ssl_domain_allow_unsecured_client
int pn_ssl_domain_allow_unsecured_client(pn_ssl_domain_t *domain)
Permit a server to accept connection requests from non-SSL clients.
pn_ssl_init
int pn_ssl_init(pn_ssl_t *ssl, pn_ssl_domain_t *domain, const char *session_id)
Initialize an SSL session.
pn_ssl_domain_free
void pn_ssl_domain_free(pn_ssl_domain_t *domain)
Release an SSL configuration domain.
pn_ssl_domain
pn_ssl_domain_t * pn_ssl_domain(pn_ssl_mode_t mode)
Create an SSL configuration domain.
pn_ssl
pn_ssl_t * pn_ssl(pn_transport_t *transport)
Create a new SSL session object associated with a transport.
pn_ssl_domain_t
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
@ PN_SSL_MODE_SERVER
Local connection endpoint is an SSL server.
Definition: ssl.h:92
pn_terminus_set_address
int pn_terminus_set_address(pn_terminus_t *terminus, const char *address)
Set the address of a terminus object.
pn_terminus_get_address
const char * pn_terminus_get_address(pn_terminus_t *terminus)
Get the address of a terminus object.
pn_transport_require_auth
void pn_transport_require_auth(pn_transport_t *transport, bool required)
Set whether a non-authenticated transport connection is allowed.
pn_transport_t
struct pn_transport_t pn_transport_t
A network channel supporting an AMQP connection.
Definition: types.h:431
pn_transport_condition
pn_condition_t * pn_transport_condition(pn_transport_t *transport)
Get additional information about the condition of the transport.
pn_transport
pn_transport_t * pn_transport(void)
Factory for creating a transport.
listener.h
Unsettled API - A listener for incoming connections.
netaddr.h
Unsettled API - The network address of a proactor transport.
proactor.h
Unsettled API - An API for multithreaded IO.
sasl.h
SASL secure transport layer.
ssl.h
SSL secure transport layer.
transport.h
A network channel supporting an AMQP connection.