1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <stdint.h>
#include <amqp.h>
#include <amqp_framing.h>
#include <unistd.h>
#include <assert.h>
#include "example_utils.h"
/* Private: compiled out in NDEBUG mode */
extern void amqp_dump(void const *buffer, size_t len);
int main(int argc, char const * const *argv) {
char const *hostname;
int port;
char const *queuename;
int sockfd;
amqp_connection_state_t conn;
if (argc < 4) {
fprintf(stderr, "Usage: amqp_listenq host port queuename\n");
return 1;
}
hostname = argv[1];
port = atoi(argv[2]);
queuename = argv[3];
conn = amqp_new_connection();
die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
amqp_set_sockfd(conn, sockfd);
die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
"Logging in");
amqp_channel_open(conn, 1);
die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
amqp_basic_consume(conn, 1, amqp_cstring_bytes(queuename), AMQP_EMPTY_BYTES, 0, 0, 0);
die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
{
amqp_frame_t frame;
int result;
amqp_basic_deliver_t *d;
amqp_basic_properties_t *p;
size_t body_target;
size_t body_received;
while (1) {
amqp_maybe_release_buffers(conn);
result = amqp_simple_wait_frame(conn, &frame);
printf("Result %d\n", result);
if (result <= 0)
break;
printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel);
if (frame.frame_type != AMQP_FRAME_METHOD)
continue;
printf("Method %s\n", amqp_method_name(frame.payload.method.id));
if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
continue;
d = (amqp_basic_deliver_t *) frame.payload.method.decoded;
printf("Delivery %u, exchange %.*s routingkey %.*s\n",
(unsigned) d->delivery_tag,
(int) d->exchange.len, (char *) d->exchange.bytes,
(int) d->routing_key.len, (char *) d->routing_key.bytes);
result = amqp_simple_wait_frame(conn, &frame);
if (result <= 0)
break;
if (frame.frame_type != AMQP_FRAME_HEADER) {
fprintf(stderr, "Expected header!");
abort();
}
p = (amqp_basic_properties_t *) frame.payload.properties.decoded;
if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
printf("Content-type: %.*s\n",
(int) p->content_type.len, (char *) p->content_type.bytes);
}
printf("----\n");
body_target = frame.payload.properties.body_size;
body_received = 0;
while (body_received < body_target) {
result = amqp_simple_wait_frame(conn, &frame);
if (result <= 0)
break;
if (frame.frame_type != AMQP_FRAME_BODY) {
fprintf(stderr, "Expected body!");
abort();
}
body_received += frame.payload.body_fragment.len;
assert(body_received <= body_target);
amqp_dump(frame.payload.body_fragment.bytes,
frame.payload.body_fragment.len);
}
if (body_received != body_target) {
/* Can only happen when amqp_simple_wait_frame returns <= 0 */
/* We break here to close the connection */
break;
}
amqp_basic_ack(conn, 1, d->delivery_tag, 0);
}
}
die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
amqp_destroy_connection(conn);
die_on_error(close(sockfd), "Closing socket");
return 0;
}
|
