1 /*
2  * ***** BEGIN LICENSE BLOCK *****
3  * Version: MIT
4  *
5  * Portions created by Alan Antonuk are Copyright (c) 2012-2013
6  * Alan Antonuk. All Rights Reserved.
7  *
8  * Portions created by VMware are Copyright (c) 2007-2012 VMware, Inc.
9  * All Rights Reserved.
10  *
11  * Portions created by Tony Garnock-Jones are Copyright (c) 2009-2010
12  * VMware, Inc. and Tony Garnock-Jones. All Rights Reserved.
13  *
14  * Permission is hereby granted, free of charge, to any person
15  * obtaining a copy of this software and associated documentation
16  * files (the "Software"), to deal in the Software without
17  * restriction, including without limitation the rights to use, copy,
18  * modify, merge, publish, distribute, sublicense, and/or sell copies
19  * of the Software, and to permit persons to whom the Software is
20  * furnished to do so, subject to the following conditions:
21  *
22  * The above copyright notice and this permission notice shall be
23  * included in all copies or substantial portions of the Software.
24  *
25  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
26  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
27  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
28  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
29  * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
30  * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
31  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
32  * SOFTWARE.
33  * ***** END LICENSE BLOCK *****
34  */
35 
36 #include <stdint.h>
37 #include <stdio.h>
38 #include <stdlib.h>
39 #include <string.h>
40 
41 #include <amqp.h>
42 #include <amqp_tcp_socket.h>
43 
44 #include <assert.h>
45 
46 #include "utils.h"
47 
main(int argc,char * argv[])48 int main(int argc, char *argv[]) {
49   char const *hostname;
50   int port, status;
51   char const *exchange;
52   char const *routingkey;
53   char const *messagebody;
54   amqp_socket_t *socket = NULL;
55   amqp_connection_state_t conn;
56   amqp_bytes_t reply_to_queue;
57 
58   if (argc < 6) { /* minimum number of mandatory arguments */
59     fprintf(stderr,
60             "usage:\namqp_rpc_sendstring_client host port exchange routingkey "
61             "messagebody\n");
62     return 1;
63   }
64 
65   hostname = argv[1];
66   port = atoi(argv[2]);
67   exchange = argv[3];
68   routingkey = argv[4];
69   messagebody = argv[5];
70 
71   /*
72      establish a channel that is used to connect RabbitMQ server
73   */
74 
75   conn = amqp_new_connection();
76 
77   socket = amqp_tcp_socket_new(conn);
78   if (!socket) {
79     die("creating TCP socket");
80   }
81 
82   status = amqp_socket_open(socket, hostname, port);
83   if (status) {
84     die("opening TCP socket");
85   }
86 
87   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
88                                "guest", "guest"),
89                     "Logging in");
90   amqp_channel_open(conn, 1);
91   die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
92 
93   /*
94      create private reply_to queue
95   */
96 
97   {
98     amqp_queue_declare_ok_t *r = amqp_queue_declare(
99         conn, 1, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table);
100     die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
101     reply_to_queue = amqp_bytes_malloc_dup(r->queue);
102     if (reply_to_queue.bytes == NULL) {
103       fprintf(stderr, "Out of memory while copying queue name");
104       return 1;
105     }
106   }
107 
108   /*
109      send the message
110   */
111 
112   {
113     /*
114       set properties
115     */
116     amqp_basic_properties_t props;
117     props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
118                    AMQP_BASIC_DELIVERY_MODE_FLAG | AMQP_BASIC_REPLY_TO_FLAG |
119                    AMQP_BASIC_CORRELATION_ID_FLAG;
120     props.content_type = amqp_cstring_bytes("text/plain");
121     props.delivery_mode = 2; /* persistent delivery mode */
122     props.reply_to = amqp_bytes_malloc_dup(reply_to_queue);
123     if (props.reply_to.bytes == NULL) {
124       fprintf(stderr, "Out of memory while copying queue name");
125       return 1;
126     }
127     props.correlation_id = amqp_cstring_bytes("1");
128 
129     /*
130       publish
131     */
132     die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes(exchange),
133                                     amqp_cstring_bytes(routingkey), 0, 0,
134                                     &props, amqp_cstring_bytes(messagebody)),
135                  "Publishing");
136 
137     amqp_bytes_free(props.reply_to);
138   }
139 
140   /*
141     wait an answer
142   */
143 
144   {
145     amqp_basic_consume(conn, 1, reply_to_queue, amqp_empty_bytes, 0, 1, 0,
146                        amqp_empty_table);
147     die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
148     amqp_bytes_free(reply_to_queue);
149 
150     {
151       amqp_frame_t frame;
152       int result;
153 
154       amqp_basic_deliver_t *d;
155       amqp_basic_properties_t *p;
156       size_t body_target;
157       size_t body_received;
158 
159       for (;;) {
160         amqp_maybe_release_buffers(conn);
161         result = amqp_simple_wait_frame(conn, &frame);
162         printf("Result: %d\n", result);
163         if (result < 0) {
164           break;
165         }
166 
167         printf("Frame type: %u channel: %u\n", frame.frame_type, frame.channel);
168         if (frame.frame_type != AMQP_FRAME_METHOD) {
169           continue;
170         }
171 
172         printf("Method: %s\n", amqp_method_name(frame.payload.method.id));
173         if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) {
174           continue;
175         }
176 
177         d = (amqp_basic_deliver_t *)frame.payload.method.decoded;
178         printf("Delivery: %u exchange: %.*s routingkey: %.*s\n",
179                (unsigned)d->delivery_tag, (int)d->exchange.len,
180                (char *)d->exchange.bytes, (int)d->routing_key.len,
181                (char *)d->routing_key.bytes);
182 
183         result = amqp_simple_wait_frame(conn, &frame);
184         if (result < 0) {
185           break;
186         }
187 
188         if (frame.frame_type != AMQP_FRAME_HEADER) {
189           fprintf(stderr, "Expected header!");
190           abort();
191         }
192         p = (amqp_basic_properties_t *)frame.payload.properties.decoded;
193         if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
194           printf("Content-type: %.*s\n", (int)p->content_type.len,
195                  (char *)p->content_type.bytes);
196         }
197         printf("----\n");
198 
199         body_target = (size_t)frame.payload.properties.body_size;
200         body_received = 0;
201 
202         while (body_received < body_target) {
203           result = amqp_simple_wait_frame(conn, &frame);
204           if (result < 0) {
205             break;
206           }
207 
208           if (frame.frame_type != AMQP_FRAME_BODY) {
209             fprintf(stderr, "Expected body!");
210             abort();
211           }
212 
213           body_received += frame.payload.body_fragment.len;
214           assert(body_received <= body_target);
215 
216           amqp_dump(frame.payload.body_fragment.bytes,
217                     frame.payload.body_fragment.len);
218         }
219 
220         if (body_received != body_target) {
221           /* Can only happen when amqp_simple_wait_frame returns <= 0 */
222           /* We break here to close the connection */
223           break;
224         }
225 
226         /* everything was fine, we can quit now because we received the reply */
227         break;
228       }
229     }
230   }
231 
232   /*
233      closing
234   */
235 
236   die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),
237                     "Closing channel");
238   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
239                     "Closing connection");
240   die_on_error(amqp_destroy_connection(conn), "Ending connection");
241 
242   return 0;
243 }
244