Skip to content

Commit 1a3125d

Browse files
astitcherastitcher
committed
PROTON-2839: Update examples to use new APIs
1 parent 9ad0c4b commit 1a3125d

5 files changed

Lines changed: 39 additions & 76 deletions

File tree

‎c/examples/direct.c‎

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ typedef struct app_data_t {
4242
int message_count;
4343

4444
pn_proactor_t *proactor;
45+
pn_message_t *message;
4546
pn_listener_t *listener;
4647
pn_rwbytes_t msgin, msgout; /* Buffers for incoming/outgoing messages */
4748

@@ -79,30 +80,29 @@ static void check_condition(pn_event_t *e, pn_condition_t *cond, app_data_t *app
7980
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
8081
static void send_message(app_data_t *app, pn_link_t *sender) {
8182
/* Construct a message with the map { "sequence": app.sent } */
82-
pn_message_t* message = pn_message();
83-
pn_data_t* body = pn_message_body(message);
84-
pn_message_set_id(message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
85-
pn_data_put_map(body);
86-
pn_data_enter(body);
87-
pn_data_put_string(body, pn_bytes(sizeof("sequence")-1, "sequence"));
88-
pn_data_put_int(body, app->sent); /* The sequence number */
89-
pn_data_exit(body);
90-
if (pn_message_send(message, sender, &app->msgout) < 0) {
91-
fprintf(stderr, "send error: %s\n", pn_error_text(pn_message_error(message)));
83+
pn_bytes_t sequence = pn_bytes(sizeof("sequence")-1, "sequence");
84+
pn_message_clear(app->message);
85+
pn_message_set_id(app->message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
86+
pn_amqp_map_t *props = pn_message_properties_build(NULL, sequence, (pn_atom_t){.type=PN_INT, .u.as_int=app->sent}, pn_bytes_null);
87+
pn_message_set_body_value(app->message, (pn_amqp_value_t*)props);
88+
pn_amqp_map_free(props);
89+
if (pn_message_send(app->message, sender, &app->msgout) < 0) {
90+
fprintf(stderr, "send error: %s\n", pn_error_text(pn_message_error(app->message)));
9291
exit_code = 1;
9392
}
94-
pn_message_free(message);
9593
}
9694

9795
static void decode_message(pn_rwbytes_t data) {
9896
pn_message_t *m = pn_message();
9997
int err = pn_message_decode(m, data.start, data.size);
10098
if (!err) {
10199
/* Print the decoded message */
102-
char *s = pn_tostring(pn_message_body(m));
100+
pn_amqp_value_t *value = pn_message_get_body_value(m);
101+
char *s = pn_amqp_value_tostring(value);
103102
printf("%s\n", s);
104103
fflush(stdout);
105104
free(s);
105+
pn_amqp_value_free(value);
106106
pn_message_free(m);
107107
free(data.start);
108108
} else {
@@ -311,6 +311,7 @@ int main(int argc, char **argv) {
311311
app.port = (argc > 2) ? argv[2] : "amqp";
312312
app.amqp_address = (argc > 3) ? argv[3] : "examples";
313313
app.message_count = (argc > 4) ? atoi(argv[4]) : 10;
314+
app.message = pn_message();
314315

315316
/* Create the proactor and connect */
316317
app.proactor = pn_proactor();
@@ -321,5 +322,6 @@ int main(int argc, char **argv) {
321322
pn_proactor_free(app.proactor);
322323
free(app.msgout.start);
323324
free(app.msgin.start);
325+
pn_message_free(app.message);
324326
return exit_code;
325327
}

‎c/examples/receive.c‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,11 @@ static void decode_message(pn_rwbytes_t data) {
6262
int err = pn_message_decode(m, data.start, data.size);
6363
if (!err) {
6464
/* Print the decoded message */
65-
char *s = pn_tostring(pn_message_body(m));
65+
pn_amqp_value_t *value = pn_message_get_body_value(m);
66+
char *s = pn_amqp_value_tostring(value);
6667
printf("%s\n", s);
6768
free(s);
69+
pn_amqp_value_free(value);
6870
pn_message_free(m);
6971
free(data.start);
7072
} else {

‎c/examples/send-abort.c‎

Lines changed: 4 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -73,40 +73,24 @@ static void check_condition(pn_event_t *e, pn_condition_t *cond) {
7373
}
7474
}
7575

76-
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
7776
static pn_bytes_t encode_message(app_data_t* app) {
78-
/* Construct a message with the map { "sequence": app.sent } */
7977
pn_message_t* message = pn_message();
8078
char data[MSG_SIZE + 11] = {0};
81-
pn_data_t* body;
8279
pn_message_set_id(message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
83-
body = pn_message_body(message);
84-
pn_data_enter(body);
85-
pn_data_put_string(body, pn_bytes(MSG_SIZE, data));
86-
pn_data_exit(body);
80+
pn_message_set_body_data(message, pn_bytes(MSG_SIZE, data));
8781

8882
/* encode the message, expanding the encode buffer as needed */
8983
if (app->message_buffer.start == NULL) {
9084
static const size_t initial_size = MSG_SIZE + 1000;
9185
app->message_buffer = pn_rwbytes(initial_size, (char*)malloc(initial_size));
9286
}
93-
/* app->message_buffer is the total buffer space available. */
94-
/* mbuf wil point at just the portion used by the encoded message */
95-
{
96-
pn_rwbytes_t mbuf = pn_rwbytes(app->message_buffer.size, app->message_buffer.start);
97-
int status = 0;
98-
while ((status = pn_message_encode(message, mbuf.start, &mbuf.size)) == PN_OVERFLOW) {
99-
app->message_buffer.size *= 2;
100-
app->message_buffer.start = (char*)realloc(app->message_buffer.start, app->message_buffer.size);
101-
mbuf.size = app->message_buffer.size;
102-
}
103-
if (status != 0) {
87+
ssize_t size = pn_message_encode2(message, &app->message_buffer);
88+
if (size < 0) {
10489
fprintf(stderr, "error encoding message: %s\n", pn_error_text(pn_message_error(message)));
10590
exit(1);
10691
}
10792
pn_message_free(message);
108-
return pn_bytes(mbuf.size, mbuf.start);
109-
}
93+
return pn_bytes(size, app->message_buffer.start);
11094
}
11195

11296
/* Returns true to continue, false if finished */

‎c/examples/send-ssl.c‎

Lines changed: 13 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ typedef struct app_data_t {
4242
int message_count;
4343

4444
pn_proactor_t *proactor;
45+
pn_message_t *message;
4546
pn_rwbytes_t message_buffer;
4647
int sent;
4748
int acknowledged;
@@ -58,40 +59,18 @@ static void check_condition(pn_event_t *e, pn_condition_t *cond) {
5859
}
5960

6061
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
61-
static pn_bytes_t encode_message(app_data_t* app) {
62+
static void send_message(app_data_t* app, pn_link_t *sender) {
6263
/* Construct a message with the map { "sequence": app.sent } */
63-
pn_message_t* message = pn_message();
64-
pn_data_t* body = pn_message_body(message);
65-
pn_message_set_id(message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
66-
pn_data_put_map(body);
67-
pn_data_enter(body);
68-
pn_data_put_string(body, pn_bytes(sizeof("sequence")-1, "sequence"));
69-
pn_data_put_int(body, app->sent); /* The sequence number */
70-
pn_data_exit(body);
71-
72-
/* encode the message, expanding the encode buffer as needed */
73-
if (app->message_buffer.start == NULL) {
74-
static const size_t initial_size = 128;
75-
app->message_buffer = pn_rwbytes(initial_size, (char*)malloc(initial_size));
76-
}
77-
/* app->message_buffer is the total buffer space available. */
78-
/* mbuf wil point at just the portion used by the encoded message */
79-
{
80-
pn_rwbytes_t mbuf = pn_rwbytes(app->message_buffer.size, app->message_buffer.start);
81-
int status = 0;
82-
while ((status = pn_message_encode(message, mbuf.start, &mbuf.size)) == PN_OVERFLOW) {
83-
app->message_buffer.size *= 2;
84-
app->message_buffer.start = (char*)realloc(app->message_buffer.start, app->message_buffer.size);
85-
mbuf.size = app->message_buffer.size;
86-
mbuf.start = app->message_buffer.start;
87-
}
88-
if (status != 0) {
89-
fprintf(stderr, "error encoding message: %s\n", pn_error_text(pn_message_error(message)));
64+
pn_bytes_t sequence = pn_bytes(sizeof("sequence")-1, "sequence");
65+
pn_message_clear(app->message);
66+
pn_message_set_id(app->message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
67+
pn_amqp_map_t *props = pn_message_properties_build(NULL, sequence, (pn_atom_t){.type=PN_INT, .u.as_int=app->sent}, pn_bytes_null);
68+
pn_message_set_body_value(app->message, (pn_amqp_value_t*)props);
69+
pn_amqp_map_free(props);
70+
if (pn_message_send(app->message, sender, &app->message_buffer) < 0) {
71+
fprintf(stderr, "error sending message: %s\n", pn_error_text(pn_message_error(app->message)));
9072
exit(1);
9173
}
92-
pn_message_free(message);
93-
return pn_bytes(mbuf.size, mbuf.start);
94-
}
9574
}
9675

9776
/* Returns true to continue, false if finished */
@@ -140,11 +119,7 @@ static bool handle(app_data_t* app, pn_event_t* event) {
140119
++app->sent;
141120
/* Use sent counter as unique delivery tag. */
142121
pn_delivery(sender, pn_dtag((const char *)&app->sent, sizeof(app->sent)));
143-
{
144-
pn_bytes_t msgbuf = encode_message(app);
145-
pn_link_send(sender, msgbuf.start, msgbuf.size);
146-
}
147-
pn_link_advance(sender);
122+
send_message(app, sender);
148123
}
149124
break;
150125
}
@@ -220,6 +195,7 @@ int main(int argc, char **argv) {
220195
app.port = (argc > 2) ? argv[2] : "amqp";
221196
app.amqp_address = (argc > 3) ? argv[3] : "examples";
222197
app.message_count = (argc > 4) ? atoi(argv[4]) : 10;
198+
app.message = pn_message();
223199
app.user = (argc > 5) ? argv[5] : 0 ;
224200
app.pass = (argc > 6) ? argv[6] : 0 ;
225201

@@ -250,5 +226,6 @@ int main(int argc, char **argv) {
250226

251227
pn_proactor_free(app.proactor);
252228
free(app.message_buffer.start);
229+
pn_message_free(app.message);
253230
return exit_code;
254231
}

‎c/examples/send.c‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
*
2020
*/
2121

22+
#include <proton/amqp_value.h>
2223
#include <proton/connection.h>
2324
#include <proton/condition.h>
2425
#include <proton/delivery.h>
@@ -58,15 +59,12 @@ static void check_condition(pn_event_t *e, pn_condition_t *cond) {
5859
/* Create a message with a map { "sequence" : number } encode it and return the encoded buffer. */
5960
static void send_message(app_data_t* app, pn_link_t *sender) {
6061
/* Construct a message with the map { "sequence": app.sent } */
61-
pn_data_t* body;
62+
pn_bytes_t sequence = pn_bytes(sizeof("sequence")-1, "sequence");
6263
pn_message_clear(app->message);
63-
body = pn_message_body(app->message);
6464
pn_message_set_id(app->message, (pn_atom_t){.type=PN_ULONG, .u.as_ulong=app->sent});
65-
pn_data_put_map(body);
66-
pn_data_enter(body);
67-
pn_data_put_string(body, pn_bytes(sizeof("sequence")-1, "sequence"));
68-
pn_data_put_int(body, app->sent); /* The sequence number */
69-
pn_data_exit(body);
65+
pn_amqp_map_t *props = pn_message_properties_build(NULL, sequence, (pn_atom_t){.type=PN_INT, .u.as_int=app->sent}, pn_bytes_null);
66+
pn_message_set_body_value(app->message, (pn_amqp_value_t*)props);
67+
pn_amqp_map_free(props);
7068
if (pn_message_send(app->message, sender, &app->message_buffer) < 0) {
7169
fprintf(stderr, "error sending message: %s\n", pn_error_text(pn_message_error(app->message)));
7270
exit(1);

0 commit comments

Comments
 (0)