receive.c

Subscribes to the 'example' node and prints the message bodies received.Can be used with broker.c, direct.c or an external AMQP broker.

/*
*
* 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/connection.h>
#include <proton/condition.h>
#include <proton/delivery.h>
#include <proton/link.h>
#include <proton/message.h>
#include <proton/proactor.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;
int received;
bool finished;
pn_rwbytes_t msgin; /* Partially received message */
} app_data_t;
static const int BATCH = 1000; /* Batch size for unlimited receive */
static int exit_code = 0;
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));
pn_connection_close(pn_event_connection(e));
exit_code = 1;
}
}
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));
pn_free(s);
pn_message_free(m);
free(data.start);
} else {
fprintf(stderr, "decode_message: %s\n", pn_code(err));
exit_code = 1;
}
}
/* 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_CONNECTION_INIT: {
pn_connection_t* c = pn_event_connection(event);
pn_session_t* s = pn_session(c);
pn_connection_set_container(c, app->container_id);
pn_connection_open(c);
pn_session_open(s);
{
pn_link_t* l = pn_receiver(s, "my_receiver");
pn_terminus_set_address(pn_link_source(l), app->amqp_address);
pn_link_open(l);
/* cannot receive without granting credit: */
pn_link_flow(l, app->message_count ? app->message_count : BATCH);
}
} break;
case PN_DELIVERY: {
/* A message has been received */
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 */
int recv;
size_t oldsize = m->size;
m->size += size;
m->start = (char*)realloc(m->start, m->size);
recv = pn_link_recv(l, m->start + oldsize, m->size);
if (recv == PN_ABORTED) {
fprintf(stderr, "Message aborted\n");
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; /* Reset the buffer for the next message*/
/* Accept the delivery */
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) {
/* Grant enough credit to bring it up to BATCH: */
pn_link_flow(l, BATCH - pn_link_credit(l));
}
} else if (++app->received >= app->message_count) {
pn_session_t *ssn = pn_link_session(l);
printf("%d messages received\n", app->received);
pn_link_close(l);
pn_session_close(ssn);
pn_connection_close(pn_session_connection(ssn));
}
}
}
break;
}
case PN_TRANSPORT_CLOSED:
check_condition(event, pn_transport_condition(pn_event_transport(event)));
break;
case PN_CONNECTION_REMOTE_CLOSE:
check_condition(event, pn_connection_remote_condition(pn_event_connection(event)));
pn_connection_close(pn_event_connection(event));
break;
case PN_SESSION_REMOTE_CLOSE:
check_condition(event, pn_session_remote_condition(pn_event_session(event)));
pn_connection_close(pn_event_connection(event));
break;
case PN_LINK_REMOTE_CLOSE:
case PN_LINK_REMOTE_DETACH:
check_condition(event, pn_link_remote_condition(pn_event_link(event)));
pn_connection_close(pn_event_connection(event));
break;
case PN_PROACTOR_INACTIVE:
return false;
break;
default:
break;
}
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) || exit_code != 0) {
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();
pn_proactor_addr(addr, sizeof(addr), app.host, app.port);
pn_proactor_connect2(app.proactor, NULL, NULL, addr);
run(&app);
pn_proactor_free(app.proactor);
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_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_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_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_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_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_DETACH
@ PN_LINK_REMOTE_DETACH
The remote endpoint has detached the link.
Definition: event.h:235
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_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_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_receiver
pn_link_t * pn_receiver(pn_session_t *session, const char *name)
Construct a new receiver on a session.
pn_link_t
struct pn_link_t pn_link_t
An AMQP Link object.
Definition: types.h:311
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_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_flow
void pn_link_flow(pn_link_t *receiver, int credit)
Grant credit for incoming deliveries on a receiver.
pn_message_body
pn_data_t * pn_message_body(pn_message_t *msg)
Get and set the body of a message.
pn_message_free
void pn_message_free(pn_message_t *msg)
Free a previously constructed pn_message_t.
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_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_proactor_connect2
PNP_EXTERN void pn_proactor_connect2(pn_proactor_t *proactor, pn_connection_t *connection, pn_transport_t *transport, const char *addr)
Connect transport to addr and bind to connection.
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_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_session_close
void pn_session_close(pn_session_t *session)
Close a session.
pn_session
pn_session_t * pn_session(pn_connection_t *connection)
Factory for creating a new session on a given connection object.
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_terminus_set_address
int pn_terminus_set_address(pn_terminus_t *terminus, const char *address)
Set the address of a terminus object.
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.
message.h
A mutable holder of application content.
proactor.h
Unsettled API - An API for multithreaded IO.
session.h
A container of links.
transport.h
A network channel supporting an AMQP connection.