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 #ifdef HAVE_CONFIG_H
37 #include "config.h"
38 #endif
39 
40 #include <stdio.h>
41 #include <stdlib.h>
42 #include <string.h>
43 
44 #include "common.h"
45 
46 #define MAX_LINE_LENGTH 1024 * 32
47 
do_publish(amqp_connection_state_t conn,char * exchange,char * routing_key,amqp_basic_properties_t * props,amqp_bytes_t body)48 static void do_publish(amqp_connection_state_t conn, char *exchange,
49                        char *routing_key, amqp_basic_properties_t *props,
50                        amqp_bytes_t body) {
51   int res = amqp_basic_publish(conn, 1, cstring_bytes(exchange),
52                                cstring_bytes(routing_key), 0, 0, props, body);
53   die_amqp_error(res, "basic.publish");
54 }
55 
main(int argc,const char ** argv)56 int main(int argc, const char **argv) {
57   amqp_connection_state_t conn;
58   static char *exchange = NULL;
59   static char *routing_key = NULL;
60   static char *content_type = NULL;
61   static char *content_encoding = NULL;
62   static char **headers = NULL;
63   static char *reply_to = NULL;
64   static char *body = NULL;
65   amqp_basic_properties_t props;
66   amqp_bytes_t body_bytes;
67   static int delivery = 1; /* non-persistent by default */
68   static int line_buffered = 0;
69   static char **pos;
70 
71   struct poptOption options[] = {
72       INCLUDE_OPTIONS(connect_options),
73       {"exchange", 'e', POPT_ARG_STRING, &exchange, 0,
74        "the exchange to publish to", "exchange"},
75       {"routing-key", 'r', POPT_ARG_STRING, &routing_key, 0,
76        "the routing key to publish with", "routing key"},
77       {"persistent", 'p', POPT_ARG_VAL, &delivery, 2,
78        "use the persistent delivery mode", NULL},
79       {"content-type", 'C', POPT_ARG_STRING, &content_type, 0,
80        "the content-type for the message", "content type"},
81       {"reply-to", 't', POPT_ARG_STRING, &reply_to, 0,
82        "the replyTo to use for the message", "reply to"},
83       {"line-buffered", 'l', POPT_ARG_VAL, &line_buffered, 2,
84        "treat each line from standard in as a separate message", NULL},
85       {"content-encoding", 'E', POPT_ARG_STRING, &content_encoding, 0,
86        "the content-encoding for the message", "content encoding"},
87       {"header", 'H', POPT_ARG_ARGV, &headers, 0,
88        "set a message header (may be specified multiple times)",
89        "\"key: value\""},
90       {"body", 'b', POPT_ARG_STRING, &body, 0, "specify the message body",
91        "body"},
92       POPT_AUTOHELP{NULL, '\0', 0, NULL, 0, NULL, NULL}};
93 
94   process_all_options(argc, argv, options);
95 
96   if (!exchange && !routing_key) {
97     fprintf(stderr, "neither exchange nor routing key specified\n");
98     return 1;
99   }
100 
101   memset(&props, 0, sizeof props);
102   props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
103   props.delivery_mode = delivery;
104 
105   if (content_type) {
106     props._flags |= AMQP_BASIC_CONTENT_TYPE_FLAG;
107     props.content_type = amqp_cstring_bytes(content_type);
108   }
109 
110   if (content_encoding) {
111     props._flags |= AMQP_BASIC_CONTENT_ENCODING_FLAG;
112     props.content_encoding = amqp_cstring_bytes(content_encoding);
113   }
114 
115   if (reply_to) {
116     props._flags |= AMQP_BASIC_REPLY_TO_FLAG;
117     props.reply_to = amqp_cstring_bytes(reply_to);
118   }
119 
120   if (headers) {
121     int num = 0;
122     for (pos = headers; *pos; pos++) {
123       num++;
124     }
125 
126     if (num > 0) {
127       amqp_table_t *table = &props.headers;
128       table->num_entries = num;
129       table->entries = calloc(num, sizeof(amqp_table_entry_t));
130       int i = 0;
131       for (pos = headers; *pos; pos++) {
132         char *colon = strchr(*pos, ':');
133         if (colon) {
134           *colon++ = '\0';
135           while (*colon == ' ') colon++;
136           table->entries[i].key = amqp_cstring_bytes(*pos);
137           table->entries[i].value.kind = AMQP_FIELD_KIND_UTF8;
138           table->entries[i].value.value.bytes = amqp_cstring_bytes(colon);
139           i++;
140         } else {
141           fprintf(stderr,
142                   "Ignored header definition missing ':' delimiter in \"%s\"\n",
143                   *pos);
144         }
145       }
146       props._flags |= AMQP_BASIC_HEADERS_FLAG;
147     }
148   }
149 
150   conn = make_connection();
151 
152   if (body) {
153     body_bytes = amqp_cstring_bytes(body);
154   } else {
155     if (line_buffered) {
156       body_bytes.bytes = (char *)malloc(MAX_LINE_LENGTH);
157       while (fgets(body_bytes.bytes, MAX_LINE_LENGTH, stdin)) {
158         body_bytes.len = strlen(body_bytes.bytes);
159         do_publish(conn, exchange, routing_key, &props, body_bytes);
160       }
161     } else {
162       body_bytes = read_all(0);
163     }
164   }
165 
166   if (!line_buffered) {
167     do_publish(conn, exchange, routing_key, &props, body_bytes);
168   }
169 
170   if (props.headers.num_entries > 0) {
171     free(props.headers.entries);
172   }
173 
174   if (!body) {
175     free(body_bytes.bytes);
176   }
177 
178   close_connection(conn);
179   return 0;
180 }
181