direct.c

A server that can be used to demonstrate direct (no broker) peer-to-peer communication It can accept an incoming connection from either the send.c or receive.c examples and will act as the directly-connected counterpart (receive or send)

/*
*
* 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 <proton/condition.h>
#include <proton/connection.h>
#include <proton/delivery.h>
#include <proton/link.h>
#include <proton/listener.h>
#include <proton/netaddr.h>
#include <proton/message.h>
#include <proton/proactor.h>
#include <proton/sasl.h>
#include <proton/session.h>
#include <proton/transport.h>
#include <stdio.h>
#include <stdlib.h>
typedef struct app_data_t {
const char *host, *port;
const char *amqp_address;
const char *container_id;
int message_count;
pn_proactor_t *proactor;
pn_listener_t *listener;
pn_rwbytes_t msgin, msgout; /* Buffers for incoming/outgoing messages */
/* Sender values */
int sent;
int acknowledged;
pn_link_t *sender;
/* Receiver values */
int received;
} app_data_t;
static const int BATCH = 1000; /* Batch size for unlimited receive */
static int exit_code = 0;
/* Close the connection and the listener so so we will get a
* PN_PROACTOR_INACTIVE event and exit, once all outstanding events
* are processed.
*/
static void close_all(pn_connection_t *c, app_data_t *app) {
if (c) pn_connection_close(c);
if (app->listener) pn_listener_close(app->listener);
}
static void check_condition(pn_event_t *e, pn_condition_t *cond, app_data_t *app) {
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));
close_all(pn_event_connection(e), app);
exit_code = 1;
}
}
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
static pn_bytes_t encode_message(app_data_t* app) {
/* Construct a message with the map { "sequence": app.sent } */
pn_message_t* message = pn_message();
pn_data_t* body = pn_message_body(message);
pn_data_put_int(pn_message_id(message), app->sent); /* Set the message_id also */
pn_data_put_map(body);
pn_data_enter(body);
pn_data_put_string(body, pn_bytes(sizeof("sequence")-1, "sequence"));
pn_data_put_int(body, app->sent); /* The sequence number */
pn_data_exit(body);
/* encode the message, expanding the encode buffer as needed */
if (app->msgout.start == NULL) {
static const size_t initial_size = 128;
app->msgout = pn_rwbytes(initial_size, (char*)malloc(initial_size));
}
/* app->msgout is the total buffer space available. */
/* mbuf wil point at just the portion used by the encoded message */
{
pn_rwbytes_t mbuf = pn_rwbytes(app->msgout.size, app->msgout.start);
int status = 0;
while ((status = pn_message_encode(message, mbuf.start, &mbuf.size)) == PN_OVERFLOW) {
app->msgout.size *= 2;
app->msgout.start = (char*)realloc(app->msgout.start, app->msgout.size);
mbuf.size = app->msgout.size;
}
if (status != 0) {
fprintf(stderr, "error encoding message: %s\n", pn_error_text(pn_message_error(message)));
exit(1);
}
pn_message_free(message);
return pn_bytes(mbuf.size, mbuf.start);
}
}
static void decode_message(pn_rwbytes_t data) {
pn_message_t *m = pn_message();
int err = pn_message_decode(m, data.start, data.size);
if (!err) {
/* Print the decoded message */
pn_string_t *s = pn_string(NULL);
pn_inspect(pn_message_body(m), s);
printf("%s\n", pn_string_get(s));
fflush(stdout);
pn_free(s);
pn_message_free(m);
free(data.start);
} else {
fprintf(stderr, "decode_message: %s\n", pn_code(err));
exit_code = 1;
}
}
/* This function handles events when we are acting as the receiver */
static void handle_receive(app_data_t *app, pn_event_t* event) {
switch (pn_event_type(event)) {
case PN_LINK_REMOTE_OPEN: {
pn_link_t *l = pn_event_link(event);
pn_link_open(l);
pn_link_flow(l, app->message_count ? app->message_count : BATCH);
} break;
case PN_DELIVERY: { /* Incoming message data */
pn_delivery_t *d = pn_event_delivery(event);
if (pn_delivery_readable(d)) {
pn_link_t *l = pn_delivery_link(d);
size_t size = pn_delivery_pending(d);
pn_rwbytes_t* m = &app->msgin; /* 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, 1); /* Replace credit for 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(recv));
pn_link_close(l); /* Unexpected error, close the link */
} else if (!pn_delivery_partial(d)) { /* Message is complete */
decode_message(*m);
*m = pn_rwbytes_null;
pn_delivery_update(d, PN_ACCEPTED);
pn_delivery_settle(d); /* settle and free d */
if (app->message_count == 0) {
/* receive forever - see if more credit is needed */
if (pn_link_credit(l) < BATCH/2) {
pn_link_flow(l, BATCH - pn_link_credit(l));
}
} else if (++app->received >= app->message_count) {
printf("%d messages received\n", app->received);
close_all(pn_event_connection(event), app);
}
}
}
break;
}
default:
break;
}
}
/* This function handles events when we are acting as the sender */
static void handle_send(app_data_t* app, pn_event_t* event) {
switch (pn_event_type(event)) {
case PN_LINK_REMOTE_OPEN: {
pn_link_t* l = pn_event_link(event);
pn_terminus_set_address(pn_link_target(l), app->amqp_address);
pn_link_open(l);
} break;
case PN_LINK_FLOW: {
/* The peer has given us some credit, now we can send messages */
pn_link_t *sender = pn_event_link(event);
while (pn_link_credit(sender) > 0 && app->sent < app->message_count) {
++app->sent;
/* Use sent counter as unique delivery tag. */
pn_delivery(sender, pn_dtag((const char *)&app->sent, sizeof(app->sent)));
{
pn_bytes_t msgbuf = encode_message(app);
pn_link_send(sender, msgbuf.start, msgbuf.size);
pn_link_advance(sender);
}
}
break;
}
case PN_DELIVERY: {
/* We received acknowledgement from the peer that a message was delivered. */
pn_delivery_t* d = pn_event_delivery(event);
if (pn_delivery_remote_state(d) == PN_ACCEPTED) {
if (++app->acknowledged == app->message_count) {
printf("%d messages sent and acknowledged\n", app->acknowledged);
close_all(pn_event_connection(event), app);
}
}
} break;
default:
break;
}
}
/* Handle all events, delegate to handle_send or handle_receive depending on link mode.
Return true to continue, false to exit
*/
static bool handle(app_data_t* app, pn_event_t* event) {
switch (pn_event_type(event)) {
case PN_LISTENER_OPEN: {
char port[256]; /* Get the listening port */
pn_netaddr_host_port(pn_listener_addr(pn_event_listener(event)), NULL, 0, port, sizeof(port));
printf("listening on %s\n", port);
fflush(stdout);
break;
}
case PN_LISTENER_ACCEPT:
pn_listener_accept2(pn_event_listener(event), NULL, NULL);
break;
case PN_CONNECTION_INIT:
pn_connection_set_container(pn_event_connection(event), app->container_id);
break;
case PN_CONNECTION_BOUND: {
/* Turn off security */
pn_transport_t *t = pn_event_transport(event);
pn_transport_require_auth(t, false);
pn_sasl_allowed_mechs(pn_sasl(t), "ANONYMOUS");
break;
}
case PN_CONNECTION_REMOTE_OPEN: {
pn_connection_open(pn_event_connection(event)); /* Complete the open */
break;
}
case PN_SESSION_REMOTE_OPEN: {
pn_session_open(pn_event_session(event));
break;
}
case PN_TRANSPORT_CLOSED:
check_condition(event, pn_transport_condition(pn_event_transport(event)), app);
break;
case PN_CONNECTION_REMOTE_CLOSE:
check_condition(event, pn_connection_remote_condition(pn_event_connection(event)), app);
pn_connection_close(pn_event_connection(event)); /* Return the close */
break;
case PN_SESSION_REMOTE_CLOSE:
check_condition(event, pn_session_remote_condition(pn_event_session(event)), app);
pn_session_close(pn_event_session(event)); /* Return the close */
pn_session_free(pn_event_session(event));
break;
case PN_LINK_REMOTE_CLOSE:
case PN_LINK_REMOTE_DETACH:
check_condition(event, pn_link_remote_condition(pn_event_link(event)), app);
pn_link_close(pn_event_link(event)); /* Return the close */
pn_link_free(pn_event_link(event));
break;
case PN_PROACTOR_TIMEOUT:
/* Wake the sender's connection */
pn_connection_wake(pn_session_connection(pn_link_session(app->sender)));
break;
case PN_LISTENER_CLOSE:
app->listener = NULL; /* Listener is closed */
check_condition(event, pn_listener_condition(pn_event_listener(event)), app);
break;
case PN_PROACTOR_INACTIVE:
return false;
break;
default: {
pn_link_t *l = pn_event_link(event);
if (l) { /* Only delegate link-related events */
if (pn_link_is_sender(l)) {
handle_send(app, event);
} else {
handle_receive(app, event);
}
}
}
}
return true;
}
void run(app_data_t *app) {
/* Loop and handle events */
do {
pn_event_batch_t *events = pn_proactor_wait(app->proactor);
pn_event_t *e;
for (e = pn_event_batch_next(events); e; e = pn_event_batch_next(events)) {
if (!handle(app, e)) {
return;
}
}
pn_proactor_done(app->proactor, events);
} while(true);
}
int main(int argc, char **argv) {
struct app_data_t app = {0};
char addr[PN_MAX_ADDR];
app.container_id = argv[0]; /* Should be unique */
app.host = (argc > 1) ? argv[1] : "";
app.port = (argc > 2) ? argv[2] : "amqp";
app.amqp_address = (argc > 3) ? argv[3] : "examples";
app.message_count = (argc > 4) ? atoi(argv[4]) : 10;
/* Create the proactor and connect */
app.proactor = pn_proactor();
app.listener = pn_listener();
pn_proactor_addr(addr, sizeof(addr), app.host, app.port);
pn_proactor_listen(app.proactor, app.listener, addr, 16);
run(&app);
pn_proactor_free(app.proactor);
free(app.msgout.start);
free(app.msgin.start);
return exit_code;
}
condition.h
An endpoint error state.
connection.h
A connection to a remote AMQP peer.
delivery.h
A message transfer.
pn_rwbytes
pn_rwbytes_t pn_rwbytes(size_t size, char *start)
Create a pn_rwbytes_t.
pn_bytes
pn_bytes_t pn_bytes(size_t size, const char *start)
Create a pn_bytes_t.
pn_bytes_t
A const byte buffer.
Definition: types.h:212
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_connection_open
void pn_connection_open(pn_connection_t *connection)
Open a connection.
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_data_enter
bool pn_data_enter(pn_data_t *data)
Sets the parent node to the current node and clears the current node.
pn_data_put_map
int pn_data_put_map(pn_data_t *data)
Puts an empty map value into a pn_data_t.
pn_data_t
struct pn_data_t pn_data_t
An AMQP Data object.
Definition: codec.h:364
pn_data_put_string
int pn_data_put_string(pn_data_t *data, pn_bytes_t string)
Puts a PN_STRING value.
pn_data_exit
bool pn_data_exit(pn_data_t *data)
Sets the current node to the parent node and the parent node to its own parent.
pn_data_put_int
int pn_data_put_int(pn_data_t *data, int32_t i)
Puts a PN_INT value.
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_delivery_remote_state
uint64_t pn_delivery_remote_state(pn_delivery_t *delivery)
Get the remote disposition state for 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_error_text
const char * pn_error_text(pn_error_t *error)
Get the error text.
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_OVERFLOW
#define PN_OVERFLOW
Overflow error.
Definition: error.h:49
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_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_PROACTOR_TIMEOUT
@ PN_PROACTOR_TIMEOUT
Timeout set by pn_proactor_set_timeout() time limit expired.
Definition: event.h:334
PN_LINK_FLOW
@ PN_LINK_FLOW
The flow control state for a link has changed.
Definition: event.h:241
PN_LINK_REMOTE_DETACH
@ PN_LINK_REMOTE_DETACH
The remote endpoint has detached the link.
Definition: event.h:235
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_CONNECTION_BOUND
@ PN_CONNECTION_BOUND
The connection has been bound to a transport.
Definition: event.h:119
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_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_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_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_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_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_flow
void pn_link_flow(pn_link_t *receiver, int credit)
Grant credit for incoming deliveries on a receiver.
pn_listener_close
PNP_EXTERN void pn_listener_close(pn_listener_t *l)
Close the listener.
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_t
struct pn_listener_t pn_listener_t
A listener for incoming connections.
Definition: types.h:420
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_message_body
pn_data_t * pn_message_body(pn_message_t *msg)
Get and set the body of a message.
pn_message_encode
int pn_message_encode(pn_message_t *msg, char *bytes, size_t *size)
Encode/save message content as AMQP formatted binary data.
pn_message_id
pn_data_t * pn_message_id(pn_message_t *msg)
Get/set the id for a message.
pn_message_free
void pn_message_free(pn_message_t *msg)
Free a previously constructed pn_message_t.
pn_message_error
pn_error_t * pn_message_error(pn_message_t *msg)
Access the error information for a message.
pn_message
pn_message_t * pn_message(void)
Construct a new pn_message_t.
pn_message_decode
int pn_message_decode(pn_message_t *msg, const char *bytes, size_t size)
Decode/load message content from AMQP formatted binary data.
pn_message_t
struct pn_message_t pn_message_t
An AMQP Message object.
Definition: message.h:51
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_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_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_terminus_set_address
int pn_terminus_set_address(pn_terminus_t *terminus, const char *address)
Set 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.
link.h
A channel for transferring messages.
listener.h
Unsettled API - A listener for incoming connections.
message.h
A mutable holder of application content.
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.
session.h
A container of links.
transport.h
A network channel supporting an AMQP connection.