har0ke/FFmpeg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

avformat/libamqp: parse vhost in uri

Browse Source 
Reviewed-by: Marton Balint <cus@passwd.hu>
Signed-off-by: Andriy Gelman <andriy.gelman@gmail.com>
This commit is contained in:
Andriy Gelman 2020-12-19 22:32:04 -05:00
parent 6e96e3d970
commit cd97e1ff4d
2 changed files with 30 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
doc/protocols.texi
View File

@ -63,16 +63,17 @@ After starting the broker, an FFmpeg client may stream data to the broker using
the command:
@example
ffmpeg -re -i input -f mpegts amqp://[[user]:[password]@@]hostname[:port]
ffmpeg -re -i input -f mpegts amqp://[[user]:[password]@@]hostname[:port][/vhost]
@end example
Where hostname and port (default is 5672) is the address of the broker. The
client may also set a user/password for authentication. The default for both
fields is "guest".
fields is "guest". Name of virtual host on broker can be set with vhost. The
default value is "/".
Muliple subscribers may stream from the broker using the command:
@example
ffplay amqp://[[user]:[password]@@]hostname[:port]
ffplay amqp://[[user]:[password]@@]hostname[:port][/vhost]
@end example
In RabbitMQ all data published to the broker flows through a specific exchange,

31
libavformat/libamqp.c
View File

@ -62,10 +62,10 @@ static const AVOption options[] = {
static int amqp_proto_open(URLContext *h, const char *uri, int flags)
{
int ret, server_msg;
char hostname[STR_LEN], credentials[STR_LEN];
char hostname[STR_LEN], credentials[STR_LEN], path[STR_LEN];
int port;
const char *user, *password = NULL;
const char *user_decoded, *password_decoded;
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 };
@ -76,7 +76,7 @@ static int amqp_proto_open(URLContext *h, const char *uri, int flags)
h->max_packet_size = s->pkt_size;
av_url_split(NULL, 0, credentials, sizeof(credentials),
hostname, sizeof(hostname), &port, NULL, 0, uri);
hostname, sizeof(hostname), &port, path, sizeof(path), uri);
if (port < 0)
port = 5672;
@ -109,8 +109,27 @@ static int amqp_proto_open(URLContext *h, const char *uri, int flags)
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");
@ -136,7 +155,7 @@ static int amqp_proto_open(URLContext *h, const char *uri, int flags)
goto destroy_connection;
}
broker_reply = amqp_login(s->conn, "/", 0, s->pkt_size, 0,
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) {
@ -195,6 +214,7 @@ static int amqp_proto_open(URLContext *h, const char *uri, int flags)
}
}
av_freep(&vhost_decoded);
av_freep(&user_decoded);
av_freep(&password_decoded);
return 0;
@ -206,6 +226,7 @@ close_connection:
destroy_connection:
amqp_destroy_connection(s->conn);
av_freep(&vhost_decoded);
av_freep(&user_decoded);
av_freep(&password_decoded);
return AVERROR_EXTERNAL;