mirror of
https://github.com/pineappleEA/pineapple-src.git
synced 2024-12-05 13:58:23 -05:00
303 lines
10 KiB
C
Executable File
303 lines
10 KiB
C
Executable File
/*
|
|
* 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;
|
|
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, "delivery_mode"},
|
|
{ "persistent", "Persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_PERSISTENT }, 0, 0, E, "delivery_mode" },
|
|
{ "non-persistent", "Non-persistent delivery mode", 0, AV_OPT_TYPE_CONST, { .i64 = AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E, "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];
|
|
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 = 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,
|
|
};
|