1 /*
2 * Advanced Message Queuing Protocol (AMQP) 0-9-1
3 * Copyright (c) 2020 Andriy Gelman
4 *
5 * This file is part of FFmpeg.
6 *
7 * FFmpeg is free software; you can redistribute it and/or
8 * modify it under the terms of the GNU Lesser General Public
9 * License as published by the Free Software Foundation; either
10 * version 2.1 of the License, or (at your option) any later version.
11 *
12 * FFmpeg is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with FFmpeg; if not, write to the Free Software
19 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20 */
21
22 #include <amqp.h>
23 #include <amqp_tcp_socket.h>
24 #include <sys/time.h>
25 #include "avformat.h"
26 #include "libavutil/avstring.h"
27 #include "libavutil/opt.h"
28 #include "libavutil/time.h"
29 #include "network.h"
30 #include "url.h"
31 #include "urldecode.h"
32
33 typedef struct AMQPContext {
34 const AVClass *class;
35 amqp_connection_state_t conn;
36 amqp_socket_t *socket;
37 const char *exchange;
38 const char *routing_key;
39 int pkt_size;
40 int64_t connection_timeout;
41 int pkt_size_overflow;
42 int delivery_mode;
43 } AMQPContext;
44
45 #define STR_LEN 1024
46 #define DEFAULT_CHANNEL 1
47
48 #define OFFSET(x) offsetof(AMQPContext, x)
49 #define D AV_OPT_FLAG_DECODING_PARAM
50 #define E AV_OPT_FLAG_ENCODING_PARAM
51 static const AVOption options[] = {
52 { "pkt_size", "Maximum send/read packet size", OFFSET(pkt_size), AV_OPT_TYPE_INT, { .i64 = 131072 }, 4096, INT_MAX, .flags = D | E },
53 { "exchange", "Exchange to send/read packets", OFFSET(exchange), AV_OPT_TYPE_STRING, { .str = "amq.direct" }, 0, 0, .flags = D | E },
54 { "routing_key", "Key to filter streams", OFFSET(routing_key), AV_OPT_TYPE_STRING, { .str = "amqp" }, 0, 0, .flags = D | E },
55 { "connection_timeout", "Initial connection timeout", OFFSET(connection_timeout), AV_OPT_TYPE_DURATION, { .i64 = -1 }, -1, INT64_MAX, .flags = D | E},
56 { "delivery_mode", "Delivery mode", OFFSET(delivery_mode), AV_OPT_TYPE_INT, { .i64 = AMQP_DELIVERY_PERSISTENT }, 1, 2, .flags = E, "delivery_mode"},
57 { "persistent", "Persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_PERSISTENT }, 0, 0, E, "delivery_mode" },
58 { "non-persistent", "Non-persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E, "delivery_mode" },
59 { NULL }
60 };
61
amqp_proto_open(URLContext * h,const char * uri,int flags)62 static int amqp_proto_open(URLContext *h, const char *uri, int flags)
63 {
64 int ret, server_msg;
65 char hostname[STR_LEN], credentials[STR_LEN], path[STR_LEN];
66 int port;
67 const char *user, *password = NULL, *vhost;
68 const char *user_decoded, *password_decoded, *vhost_decoded;
69 char *p;
70 amqp_rpc_reply_t broker_reply;
71 struct timeval tval = { 0 };
72
73 AMQPContext *s = h->priv_data;
74
75 h->is_streamed = 1;
76 h->max_packet_size = s->pkt_size;
77
78 av_url_split(NULL, 0, credentials, sizeof(credentials),
79 hostname, sizeof(hostname), &port, path, sizeof(path), uri);
80
81 if (port < 0)
82 port = 5672;
83
84 if (hostname[0] == '\0' || port <= 0 || port > 65535 ) {
85 av_log(h, AV_LOG_ERROR, "Invalid hostname/port\n");
86 return AVERROR(EINVAL);
87 }
88
89 p = strchr(credentials, ':');
90 if (p) {
91 *p = '\0';
92 password = p + 1;
93 }
94
95 if (!password || *password == '\0')
96 password = "guest";
97
98 password_decoded = ff_urldecode(password, 0);
99 if (!password_decoded)
100 return AVERROR(ENOMEM);
101
102 user = credentials;
103 if (*user == '\0')
104 user = "guest";
105
106 user_decoded = ff_urldecode(user, 0);
107 if (!user_decoded) {
108 av_freep(&password_decoded);
109 return AVERROR(ENOMEM);
110 }
111
112 /* skip query for now */
113 p = strchr(path, '?');
114 if (p)
115 *p = '\0';
116
117 vhost = path;
118 if (*vhost == '\0')
119 vhost = "/";
120 else
121 vhost++; /* skip leading '/' */
122
123 vhost_decoded = ff_urldecode(vhost, 0);
124 if (!vhost_decoded) {
125 av_freep(&user_decoded);
126 av_freep(&password_decoded);
127 return AVERROR(ENOMEM);
128 }
129
130 s->conn = amqp_new_connection();
131 if (!s->conn) {
132 av_freep(&vhost_decoded);
133 av_freep(&user_decoded);
134 av_freep(&password_decoded);
135 av_log(h, AV_LOG_ERROR, "Error creating connection\n");
136 return AVERROR_EXTERNAL;
137 }
138
139 s->socket = amqp_tcp_socket_new(s->conn);
140 if (!s->socket) {
141 av_log(h, AV_LOG_ERROR, "Error creating socket\n");
142 goto destroy_connection;
143 }
144
145 if (s->connection_timeout < 0)
146 s->connection_timeout = (h->rw_timeout > 0 ? h->rw_timeout : 5000000);
147
148 tval.tv_sec = s->connection_timeout / 1000000;
149 tval.tv_usec = s->connection_timeout % 1000000;
150 ret = amqp_socket_open_noblock(s->socket, hostname, port, &tval);
151
152 if (ret) {
153 av_log(h, AV_LOG_ERROR, "Error connecting to server: %s\n",
154 amqp_error_string2(ret));
155 goto destroy_connection;
156 }
157
158 broker_reply = amqp_login(s->conn, vhost_decoded, 0, s->pkt_size, 0,
159 AMQP_SASL_METHOD_PLAIN, user_decoded, password_decoded);
160
161 if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) {
162 av_log(h, AV_LOG_ERROR, "Error login\n");
163 server_msg = AMQP_ACCESS_REFUSED;
164 goto close_connection;
165 }
166
167 amqp_channel_open(s->conn, DEFAULT_CHANNEL);
168 broker_reply = amqp_get_rpc_reply(s->conn);
169
170 if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) {
171 av_log(h, AV_LOG_ERROR, "Error set channel\n");
172 server_msg = AMQP_CHANNEL_ERROR;
173 goto close_connection;
174 }
175
176 if (h->flags & AVIO_FLAG_READ) {
177 amqp_bytes_t queuename;
178 char queuename_buff[STR_LEN];
179 amqp_queue_declare_ok_t *r;
180
181 r = amqp_queue_declare(s->conn, DEFAULT_CHANNEL, amqp_empty_bytes,
182 0, 0, 0, 1, amqp_empty_table);
183 broker_reply = amqp_get_rpc_reply(s->conn);
184 if (!r || broker_reply.reply_type != AMQP_RESPONSE_NORMAL) {
185 av_log(h, AV_LOG_ERROR, "Error declare queue\n");
186 server_msg = AMQP_RESOURCE_ERROR;
187 goto close_channel;
188 }
189
190 /* store queuename */
191 queuename.bytes = queuename_buff;
192 queuename.len = FFMIN(r->queue.len, STR_LEN);
193 memcpy(queuename.bytes, r->queue.bytes, queuename.len);
194
195 amqp_queue_bind(s->conn, DEFAULT_CHANNEL, queuename,
196 amqp_cstring_bytes(s->exchange),
197 amqp_cstring_bytes(s->routing_key), amqp_empty_table);
198
199 broker_reply = amqp_get_rpc_reply(s->conn);
200 if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) {
201 av_log(h, AV_LOG_ERROR, "Queue bind error\n");
202 server_msg = AMQP_INTERNAL_ERROR;
203 goto close_channel;
204 }
205
206 amqp_basic_consume(s->conn, DEFAULT_CHANNEL, queuename, amqp_empty_bytes,
207 0, 1, 0, amqp_empty_table);
208
209 broker_reply = amqp_get_rpc_reply(s->conn);
210 if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) {
211 av_log(h, AV_LOG_ERROR, "Set consume error\n");
212 server_msg = AMQP_INTERNAL_ERROR;
213 goto close_channel;
214 }
215 }
216
217 av_freep(&vhost_decoded);
218 av_freep(&user_decoded);
219 av_freep(&password_decoded);
220 return 0;
221
222 close_channel:
223 amqp_channel_close(s->conn, DEFAULT_CHANNEL, server_msg);
224 close_connection:
225 amqp_connection_close(s->conn, server_msg);
226 destroy_connection:
227 amqp_destroy_connection(s->conn);
228
229 av_freep(&vhost_decoded);
230 av_freep(&user_decoded);
231 av_freep(&password_decoded);
232 return AVERROR_EXTERNAL;
233 }
234
amqp_proto_write(URLContext * h,const unsigned char * buf,int size)235 static int amqp_proto_write(URLContext *h, const unsigned char *buf, int size)
236 {
237 int ret;
238 AMQPContext *s = h->priv_data;
239 int fd = amqp_socket_get_sockfd(s->socket);
240
241 amqp_bytes_t message = { size, (void *)buf };
242 amqp_basic_properties_t props;
243
244 ret = ff_network_wait_fd_timeout(fd, 1, h->rw_timeout, &h->interrupt_callback);
245 if (ret)
246 return ret;
247
248 props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
249 props.content_type = amqp_cstring_bytes("octet/stream");
250 props.delivery_mode = s->delivery_mode;
251
252 ret = amqp_basic_publish(s->conn, DEFAULT_CHANNEL, amqp_cstring_bytes(s->exchange),
253 amqp_cstring_bytes(s->routing_key), 0, 0,
254 &props, message);
255
256 if (ret) {
257 av_log(h, AV_LOG_ERROR, "Error publish: %s\n", amqp_error_string2(ret));
258 return AVERROR_EXTERNAL;
259 }
260
261 return size;
262 }
263
amqp_proto_read(URLContext * h,unsigned char * buf,int size)264 static int amqp_proto_read(URLContext *h, unsigned char *buf, int size)
265 {
266 AMQPContext *s = h->priv_data;
267 int fd = amqp_socket_get_sockfd(s->socket);
268 int ret;
269
270 amqp_rpc_reply_t broker_reply;
271 amqp_envelope_t envelope;
272
273 ret = ff_network_wait_fd_timeout(fd, 0, h->rw_timeout, &h->interrupt_callback);
274 if (ret)
275 return ret;
276
277 amqp_maybe_release_buffers(s->conn);
278 broker_reply = amqp_consume_message(s->conn, &envelope, NULL, 0);
279
280 if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL)
281 return AVERROR_EXTERNAL;
282
283 if (envelope.message.body.len > size) {
284 s->pkt_size_overflow = FFMAX(s->pkt_size_overflow, envelope.message.body.len);
285 av_log(h, AV_LOG_WARNING, "Message exceeds space in the buffer. "
286 "Message will be truncated. Setting -pkt_size %d "
287 "may resolve this issue.\n", s->pkt_size_overflow);
288 }
289 size = FFMIN(size, envelope.message.body.len);
290
291 memcpy(buf, envelope.message.body.bytes, size);
292 amqp_destroy_envelope(&envelope);
293
294 return size;
295 }
296
amqp_proto_close(URLContext * h)297 static int amqp_proto_close(URLContext *h)
298 {
299 AMQPContext *s = h->priv_data;
300 amqp_channel_close(s->conn, DEFAULT_CHANNEL, AMQP_REPLY_SUCCESS);
301 amqp_connection_close(s->conn, AMQP_REPLY_SUCCESS);
302 amqp_destroy_connection(s->conn);
303
304 return 0;
305 }
306
307 static const AVClass amqp_context_class = {
308 .class_name = "amqp",
309 .item_name = av_default_item_name,
310 .option = options,
311 .version = LIBAVUTIL_VERSION_INT,
312 };
313
314 const URLProtocol ff_libamqp_protocol = {
315 .name = "amqp",
316 .url_close = amqp_proto_close,
317 .url_open = amqp_proto_open,
318 .url_read = amqp_proto_read,
319 .url_write = amqp_proto_write,
320 .priv_data_size = sizeof(AMQPContext),
321 .priv_data_class = &amqp_context_class,
322 .flags = URL_PROTOCOL_FLAG_NETWORK,
323 };
324