Loading...
Searching...
No Matches
send.c

Send a fixed number of messages to the "examples" node.

Send a fixed number of messages to the "examples" node.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;
pn_message_t *message;
pn_rwbytes_t message_buffer;
int sent;
int acknowledged;
} app_data_t;
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;
}
}
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
static void send_message(app_data_t* app, pn_link_t *sender) {
/* Construct a message with the map { "sequence": app.sent } */
pn_data_t* body;
pn_message_clear(app->message);
body = pn_message_body(app->message);
pn_message_set_id(app->message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
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);
if (pn_message_send(app->message, sender, &app->message_buffer) < 0) {
fprintf(stderr, "error sending message: %s\n", pn_error_text(pn_message_error(app->message)));
exit(1);
}
}
/* Returns true to continue, false if finished */
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(pn_event_connection(event));
pn_connection_set_container(c, app->container_id);
pn_connection_open(c);
pn_session_open(s);
{
pn_link_t* l = pn_sender(s, "my_sender");
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)));
send_message(app, 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);
pn_connection_close(pn_event_connection(event));
/* Continue handling events till we receive TRANSPORT_CLOSED */
}
} else {
fprintf(stderr, "unexpected delivery state %d\n", (int)pn_delivery_remote_state(d));
pn_connection_close(pn_event_connection(event));
exit_code=1;
}
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;
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)) {
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;
app.message = pn_message();
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);
free(app.message_buffer.start);
pn_message_free(app.message);
return exit_code;
}
condition.h
An endpoint error state.
connection.h
A connection to a remote AMQP peer.
delivery.h
A message transfer.
pn_bytes
pn_bytes_t pn_bytes(size_t size, const char *start)
Create a pn_bytes_t.
pn_atom_t
A discriminated union that holds any scalar AMQP value.
Definition codec.h:200
pn_rwbytes_t
A non-const byte buffer.
Definition types.h:235
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_get_description
const char * pn_condition_get_description(pn_condition_t *condition)
Gets the description associated with the exceptional condition.
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_connection_open
void pn_connection_open(pn_connection_t *connection)
Open a connection.
pn_connection_t
struct pn_connection_t pn_connection_t
An AMQP Connection object.
Definition types.h:285
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_remote_condition
pn_condition_t * pn_connection_remote_condition(pn_connection_t *connection)
Get the remote condition associated with the connection endpoint.
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:375
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
pn_delivery_t * pn_delivery(pn_link_t *link, pn_delivery_tag_t tag)
Create a delivery on a link.
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:405
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_event_delivery
pn_delivery_t * pn_event_delivery(pn_event_t *event)
Get the delivery associated with an event.
pn_event_session
pn_session_t * pn_event_session(pn_event_t *event)
Get the session 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_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_connection
pn_connection_t * pn_event_connection(pn_event_t *event)
Get the connection associated with an event.
pn_event_type
pn_event_type_t pn_event_type(pn_event_t *event)
Get the type of an event.
pn_event_transport
pn_transport_t * pn_event_transport(pn_event_t *event)
Get the transport 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_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_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_remote_condition
pn_condition_t * pn_link_remote_condition(pn_link_t *link)
Get the remote 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_t
struct pn_link_t pn_link_t
An AMQP Link object.
Definition types.h:315
pn_link_open
void pn_link_open(pn_link_t *link)
Open a link.
pn_sender
pn_link_t * pn_sender(pn_session_t *session, const char *name)
Construct a new sender on a session.
pn_link_target
pn_terminus_t * pn_link_target(pn_link_t *link)
Access the locally defined target definition for a link.
pn_message_set_id
int pn_message_set_id(pn_message_t *msg, pn_msgid_t id)
Set the id for a message.
pn_message
pn_message_t * pn_message(void)
Construct a new 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_free
void pn_message_free(pn_message_t *msg)
Free a previously constructed pn_message_t.
pn_message_send
ssize_t pn_message_send(pn_message_t *msg, pn_link_t *sender, pn_rwbytes_t *buf)
Unsettled API
pn_message_clear
void pn_message_clear(pn_message_t *msg)
Clears the content of a pn_message_t.
pn_message_body
pn_data_t * pn_message_body(pn_message_t *msg)
Get and set the body of a message.
pn_message_t
struct pn_message_t pn_message_t
An AMQP Message object.
Definition message.h:51
pn_proactor
PNP_EXTERN pn_proactor_t * pn_proactor(void)
Create a proactor.
pn_event_batch_next
PNP_EXTERN pn_event_t * pn_event_batch_next(pn_event_batch_t *batch)
Remove the next event from the batch and return it.
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_event_batch_t
struct pn_event_batch_t pn_event_batch_t
A batch of events that must be handled in sequence.
Definition types.h:462
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:442
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_connect() or pn_proactor_listen()
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_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:296
pn_session
pn_session_t * pn_session(pn_connection_t *connection)
Factory for creating a new session on a given connection object.
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.