/* * Advanced Message Queuing Protocol (AMQP) 0-9-1 * Copyright (c) 2020 Andriy Gelman * * This file is part of FFmpeg. * * FFmpeg is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public * License as published by the Free Software Foundation; either * version 2.1 of the License, or (at your option) any later version. * * FFmpeg is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public * License along with FFmpeg; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA */ #include <amqp.h> #include <amqp_tcp_socket.h> #include <sys/time.h> #include "avformat.h" #include "libavutil/mem.h" #include "libavutil/opt.h" #include "network.h" #include "url.h" #include "urldecode.h" typedef struct AMQPContext { const AVClass *class; amqp_connection_state_t conn; amqp_socket_t *socket; const char *exchange; const char *routing_key; int pkt_size; int64_t connection_timeout; int pkt_size_overflow; int delivery_mode; } AMQPContext; #define STR_LEN 1024 #define DEFAULT_CHANNEL 1 #define OFFSET(x) offsetof(AMQPContext, x) #define D AV_OPT_FLAG_DECODING_PARAM #define E AV_OPT_FLAG_ENCODING_PARAM static const AVOption options[] = { { "pkt_size", "Maximum send/read packet size", OFFSET(pkt_size), AV_OPT_TYPE_INT, { .i64 = 131072 }, 4096, INT_MAX, .flags = D | E }, { "exchange", "Exchange to send/read packets", OFFSET(exchange), AV_OPT_TYPE_STRING, { .str = "amq.direct" }, 0, 0, .flags = D | E }, { "routing_key", "Key to filter streams", OFFSET(routing_key), AV_OPT_TYPE_STRING, { .str = "amqp" }, 0, 0, .flags = D | E }, { "connection_timeout", "Initial connection timeout", OFFSET(connection_timeout), AV_OPT_TYPE_DURATION, { .i64 = -1 }, -1, INT64_MAX, .flags = D | E}, { "delivery_mode", "Delivery mode", OFFSET(delivery_mode), AV_OPT_TYPE_INT, { .i64 = AMQP_DELIVERY_PERSISTENT }, 1, 2, .flags = E, .unit = "delivery_mode"}, { "persistent", "Persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_PERSISTENT }, 0, 0, E, .unit = "delivery_mode" }, { "non-persistent", "Non-persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E, .unit = "delivery_mode" }, { NULL } }; static int amqp_proto_open(URLContext *h, const char *uri, int flags) { int ret, server_msg; char hostname[STR_LEN], credentials[STR_LEN], path[STR_LEN]; int port; const char *user, *password = NULL, *vhost; const char *user_decoded, *password_decoded, *vhost_decoded; char *p; amqp_rpc_reply_t broker_reply; struct timeval tval = { 0 }; AMQPContext *s = h->priv_data; h->is_streamed = 1; h->max_packet_size = s->pkt_size; av_url_split(NULL, 0, credentials, sizeof(credentials), hostname, sizeof(hostname), &port, path, sizeof(path), uri); if (port < 0) port = 5672; if (hostname[0] == '\0' || port <= 0 || port > 65535 ) { av_log(h, AV_LOG_ERROR, "Invalid hostname/port\n"); return AVERROR(EINVAL); } p = strchr(credentials, ':'); if (p) { *p = '\0'; password = p + 1; } if (!password || *password == '\0') password = "guest"; password_decoded = ff_urldecode(password, 0); if (!password_decoded) return AVERROR(ENOMEM); user = credentials; if (*user == '\0') user = "guest"; user_decoded = ff_urldecode(user, 0); if (!user_decoded) { av_freep(&password_decoded); return AVERROR(ENOMEM); } /* skip query for now */ p = strchr(path, '?'); if (p) *p = '\0'; vhost = path; if (*vhost == '\0') vhost = "/"; else vhost++; /* skip leading '/' */ vhost_decoded = ff_urldecode(vhost, 0); if (!vhost_decoded) { av_freep(&user_decoded); av_freep(&password_decoded); return AVERROR(ENOMEM); } s->conn = amqp_new_connection(); if (!s->conn) { av_freep(&vhost_decoded); av_freep(&user_decoded); av_freep(&password_decoded); av_log(h, AV_LOG_ERROR, "Error creating connection\n"); return AVERROR_EXTERNAL; } s->socket = amqp_tcp_socket_new(s->conn); if (!s->socket) { av_log(h, AV_LOG_ERROR, "Error creating socket\n"); goto destroy_connection; } if (s->connection_timeout < 0) s->connection_timeout = (h->rw_timeout > 0 ? h->rw_timeout : 5000000); tval.tv_sec = s->connection_timeout / 1000000; tval.tv_usec = s->connection_timeout % 1000000; ret = amqp_socket_open_noblock(s->socket, hostname, port, &tval); if (ret) { av_log(h, AV_LOG_ERROR, "Error connecting to server: %s\n", amqp_error_string2(ret)); goto destroy_connection; } broker_reply = amqp_login(s->conn, vhost_decoded, 0, s->pkt_size, 0, AMQP_SASL_METHOD_PLAIN, user_decoded, password_decoded); if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) { av_log(h, AV_LOG_ERROR, "Error login\n"); server_msg = AMQP_ACCESS_REFUSED; goto close_connection; } amqp_channel_open(s->conn, DEFAULT_CHANNEL); broker_reply = amqp_get_rpc_reply(s->conn); if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) { av_log(h, AV_LOG_ERROR, "Error set channel\n"); server_msg = AMQP_CHANNEL_ERROR; goto close_connection; } if (h->flags & AVIO_FLAG_READ) { amqp_bytes_t queuename; char queuename_buff[STR_LEN]; amqp_queue_declare_ok_t *r; r = amqp_queue_declare(s->conn, DEFAULT_CHANNEL, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table); broker_reply = amqp_get_rpc_reply(s->conn); if (!r || broker_reply.reply_type != AMQP_RESPONSE_NORMAL) { av_log(h, AV_LOG_ERROR, "Error declare queue\n"); server_msg = AMQP_RESOURCE_ERROR; goto close_channel; } /* store queuename */ queuename.bytes = queuename_buff; queuename.len = FFMIN(r->queue.len, STR_LEN); memcpy(queuename.bytes, r->queue.bytes, queuename.len); amqp_queue_bind(s->conn, DEFAULT_CHANNEL, queuename, amqp_cstring_bytes(s->exchange), amqp_cstring_bytes(s->routing_key), amqp_empty_table); broker_reply = amqp_get_rpc_reply(s->conn); if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) { av_log(h, AV_LOG_ERROR, "Queue bind error\n"); server_msg = AMQP_INTERNAL_ERROR; goto close_channel; } amqp_basic_consume(s->conn, DEFAULT_CHANNEL, queuename, amqp_empty_bytes, 0, 1, 0, amqp_empty_table); broker_reply = amqp_get_rpc_reply(s->conn); if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) { av_log(h, AV_LOG_ERROR, "Set consume error\n"); server_msg = AMQP_INTERNAL_ERROR; goto close_channel; } } av_freep(&vhost_decoded); av_freep(&user_decoded); av_freep(&password_decoded); return 0; close_channel: amqp_channel_close(s->conn, DEFAULT_CHANNEL, server_msg); close_connection: amqp_connection_close(s->conn, server_msg); destroy_connection: amqp_destroy_connection(s->conn); av_freep(&vhost_decoded); av_freep(&user_decoded); av_freep(&password_decoded); return AVERROR_EXTERNAL; } static int amqp_proto_write(URLContext *h, const unsigned char *buf, int size) { int ret; AMQPContext *s = h->priv_data; int fd = amqp_socket_get_sockfd(s->socket); amqp_bytes_t message = { size, (void *)buf }; amqp_basic_properties_t props; ret = ff_network_wait_fd_timeout(fd, 1, h->rw_timeout, &h->interrupt_callback); if (ret) return ret; props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG; props.content_type = amqp_cstring_bytes("octet/stream"); props.delivery_mode = s->delivery_mode; ret = amqp_basic_publish(s->conn, DEFAULT_CHANNEL, amqp_cstring_bytes(s->exchange), amqp_cstring_bytes(s->routing_key), 0, 0, &props, message); if (ret) { av_log(h, AV_LOG_ERROR, "Error publish: %s\n", amqp_error_string2(ret)); return AVERROR_EXTERNAL; } return size; } static int amqp_proto_read(URLContext *h, unsigned char *buf, int size) { AMQPContext *s = h->priv_data; int fd = amqp_socket_get_sockfd(s->socket); int ret; amqp_rpc_reply_t broker_reply; amqp_envelope_t envelope; ret = ff_network_wait_fd_timeout(fd, 0, h->rw_timeout, &h->interrupt_callback); if (ret) return ret; amqp_maybe_release_buffers(s->conn); broker_reply = amqp_consume_message(s->conn, &envelope, NULL, 0); if (broker_reply.reply_type != AMQP_RESPONSE_NORMAL) return AVERROR_EXTERNAL; if (envelope.message.body.len > size) { s->pkt_size_overflow = FFMAX(s->pkt_size_overflow, envelope.message.body.len); av_log(h, AV_LOG_WARNING, "Message exceeds space in the buffer. " "Message will be truncated. Setting -pkt_size %d " "may resolve this issue.\n", s->pkt_size_overflow); } size = FFMIN(size, envelope.message.body.len); memcpy(buf, envelope.message.body.bytes, size); amqp_destroy_envelope(&envelope); return size; } static int amqp_proto_close(URLContext *h) { AMQPContext *s = h->priv_data; amqp_channel_close(s->conn, DEFAULT_CHANNEL, AMQP_REPLY_SUCCESS); amqp_connection_close(s->conn, AMQP_REPLY_SUCCESS); amqp_destroy_connection(s->conn); return 0; } static const AVClass amqp_context_class = { .class_name = "amqp", .item_name = av_default_item_name, .option = options, .version = LIBAVUTIL_VERSION_INT, }; const URLProtocol ff_libamqp_protocol = { .name = "amqp", .url_close = amqp_proto_close, .url_open = amqp_proto_open, .url_read = amqp_proto_read, .url_write = amqp_proto_write, .priv_data_size = sizeof(AMQPContext), .priv_data_class = &amqp_context_class, .flags = URL_PROTOCOL_FLAG_NETWORK, };