mirror of https://github.com/FFmpeg/FFmpeg.git
Supports connecting to a RabbitMQ broker via AMQP version 0-9-1. Signed-off-by: Andriy Gelman <andriy.gelman@gmail.com> Signed-off-by: Marton Balint <cus@passwd.hu>pull/334/head
Andriy Gelman
5 years ago
committed by
Marton Balint
parent
d99f3dc6b2
commit
1676d0fb25
8 changed files with 368 additions and 1 deletions
@ -0,0 +1,298 @@ |
||||
/*
|
||||
* 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/avstring.h" |
||||
#include "libavutil/opt.h" |
||||
#include "libavutil/time.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; |
||||
} 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}, |
||||
{ NULL } |
||||
}; |
||||
|
||||
static int amqp_proto_open(URLContext *h, const char *uri, int flags) |
||||
{ |
||||
int ret, server_msg; |
||||
char hostname[STR_LEN], credentials[STR_LEN]; |
||||
int port; |
||||
const char *user, *password = NULL; |
||||
const char *user_decoded, *password_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, NULL, 0, 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); |
||||
} |
||||
|
||||
s->conn = amqp_new_connection(); |
||||
if (!s->conn) { |
||||
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, "/", 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(&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(&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 = 2; /* persistent 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, |
||||
}; |
||||
Loading…
Reference in new issue