git.sesse.net Git - ffmpeg/blob - libavformat/libzmq.c

   1 /*
   2  * ZeroMQ Protocol
   3  * Copyright (c) 2019 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 <zmq.h>
  23 #include "url.h"
  24 #include "network.h"
  25 #include "libavutil/avstring.h"
  26 #include "libavutil/opt.h"
  27 #include "libavutil/time.h"
  28
  29 #define ZMQ_STRERROR zmq_strerror(zmq_errno())
  30
  31 typedef struct ZMQContext {
  32     const AVClass *class;
  33     void *context;
  34     void *socket;
  35     int   pkt_size;
  36     int   pkt_size_overflow; /*keep track of the largest packet during overflow*/
  37 } ZMQContext;
  38
  39 #define OFFSET(x) offsetof(ZMQContext, x)
  40 #define D AV_OPT_FLAG_DECODING_PARAM
  41 #define E AV_OPT_FLAG_ENCODING_PARAM
  42 static const AVOption options[] = {
  43     { "pkt_size", "Maximum send/read packet size", OFFSET(pkt_size), AV_OPT_TYPE_INT, { .i64 = 131072 }, -1, INT_MAX, .flags = D | E },
  44     { NULL }
  45 };
  46
  47 static int zmq_proto_wait(URLContext *h, void *socket, int write)
  48 {
  49     int ret;
  50     int ev = write ? ZMQ_POLLOUT : ZMQ_POLLIN;
  51     zmq_pollitem_t items = { .socket = socket, .fd = 0, .events = ev, .revents = 0 };
  52     ret = zmq_poll(&items, 1, POLLING_TIME);
  53     if (ret == -1) {
  54         av_log(h, AV_LOG_ERROR, "Error occured during zmq_poll(): %s\n", ZMQ_STRERROR);
  55         return AVERROR_EXTERNAL;
  56     }
  57     return items.revents & ev ? 0 : AVERROR(EAGAIN);
  58 }
  59
  60 static int zmq_proto_wait_timeout(URLContext *h, void *socket, int write, int64_t timeout, AVIOInterruptCB *int_cb)
  61 {
  62     int ret;
  63     int64_t wait_start = 0;
  64
  65     while (1) {
  66         if (ff_check_interrupt(int_cb))
  67             return AVERROR_EXIT;
  68         ret = zmq_proto_wait(h, socket, write);
  69         if (ret != AVERROR(EAGAIN))
  70             return ret;
  71         if (timeout > 0) {
  72             if (!wait_start)
  73                 wait_start = av_gettime_relative();
  74             else if (av_gettime_relative() - wait_start > timeout)
  75                 return AVERROR(ETIMEDOUT);
  76         }
  77     }
  78 }
  79
  80 static int zmq_proto_open(URLContext *h, const char *uri, int flags)
  81 {
  82     int ret;
  83     ZMQContext *s        = h->priv_data;
  84     s->pkt_size_overflow = 0;
  85     h->is_streamed       = 1;
  86
  87     if (s->pkt_size > 0)
  88         h->max_packet_size = s->pkt_size;
  89
  90     s->context = zmq_ctx_new();
  91     if (!s->context) {
  92         /*errno not set on failure during zmq_ctx_new()*/
  93         av_log(h, AV_LOG_ERROR, "Error occured during zmq_ctx_new()\n");
  94         return AVERROR_EXTERNAL;
  95     }
  96
  97     av_strstart(uri, "zmq:", &uri);
  98
  99     /*publish during write*/
 100     if (h->flags & AVIO_FLAG_WRITE) {
 101         s->socket = zmq_socket(s->context, ZMQ_PUB);
 102         if (!s->socket) {
 103             av_log(h, AV_LOG_ERROR, "Error occured during zmq_socket(): %s\n", ZMQ_STRERROR);
 104             goto fail_term;
 105         }
 106
 107         ret = zmq_bind(s->socket, uri);
 108         if (ret == -1) {
 109             av_log(h, AV_LOG_ERROR, "Error occured during zmq_bind(): %s\n", ZMQ_STRERROR);
 110             goto fail_close;
 111         }
 112     }
 113
 114     /*subscribe for read*/
 115     if (h->flags & AVIO_FLAG_READ) {
 116         s->socket = zmq_socket(s->context, ZMQ_SUB);
 117         if (!s->socket) {
 118             av_log(h, AV_LOG_ERROR, "Error occured during zmq_socket(): %s\n", ZMQ_STRERROR);
 119             goto fail_term;
 120         }
 121
 122         ret = zmq_setsockopt(s->socket, ZMQ_SUBSCRIBE, "", 0);
 123         if (ret == -1) {
 124             av_log(h, AV_LOG_ERROR, "Error occured during zmq_setsockopt(): %s\n", ZMQ_STRERROR);
 125             goto fail_close;
 126         }
 127
 128         ret = zmq_connect(s->socket, uri);
 129         if (ret == -1) {
 130             av_log(h, AV_LOG_ERROR, "Error occured during zmq_connect(): %s\n", ZMQ_STRERROR);
 131             goto fail_close;
 132         }
 133     }
 134     return 0;
 135
 136 fail_close:
 137     zmq_close(s->socket);
 138 fail_term:
 139     zmq_ctx_term(s->context);
 140     return AVERROR_EXTERNAL;
 141 }
 142
 143 static int zmq_proto_write(URLContext *h, const unsigned char *buf, int size)
 144 {
 145     int ret;
 146     ZMQContext *s = h->priv_data;
 147
 148     ret = zmq_proto_wait_timeout(h, s->socket, 1, h->rw_timeout, &h->interrupt_callback);
 149     if (ret)
 150         return ret;
 151     ret = zmq_send(s->socket, buf, size, 0);
 152     if (ret == -1) {
 153         av_log(h, AV_LOG_ERROR, "Error occured during zmq_send(): %s\n", ZMQ_STRERROR);
 154         return AVERROR_EXTERNAL;
 155     }
 156     return ret; /*number of bytes sent*/
 157 }
 158
 159 static int zmq_proto_read(URLContext *h, unsigned char *buf, int size)
 160 {
 161     int ret;
 162     ZMQContext *s = h->priv_data;
 163
 164     ret = zmq_proto_wait_timeout(h, s->socket, 0, h->rw_timeout, &h->interrupt_callback);
 165     if (ret)
 166         return ret;
 167     ret = zmq_recv(s->socket, buf, size, 0);
 168     if (ret == -1) {
 169         av_log(h, AV_LOG_ERROR, "Error occured during zmq_recv(): %s\n", ZMQ_STRERROR);
 170         return AVERROR_EXTERNAL;
 171     }
 172     if (ret > size) {
 173         s->pkt_size_overflow = FFMAX(s->pkt_size_overflow, ret);
 174         av_log(h, AV_LOG_WARNING, "Message exceeds available space in the buffer. Message will be truncated. Setting -pkt_size %d may resolve the issue.\n", s->pkt_size_overflow);
 175         ret = size;
 176     }
 177     return ret; /*number of bytes read*/
 178 }
 179
 180 static int zmq_proto_close(URLContext *h)
 181 {
 182     ZMQContext *s = h->priv_data;
 183     zmq_close(s->socket);
 184     zmq_ctx_term(s->context);
 185     return 0;
 186 }
 187
 188 static const AVClass zmq_context_class = {
 189     .class_name = "zmq",
 190     .item_name  = av_default_item_name,
 191     .option     = options,
 192     .version    = LIBAVUTIL_VERSION_INT,
 193 };
 194
 195 const URLProtocol ff_libzmq_protocol = {
 196     .name            = "zmq",
 197     .url_close       = zmq_proto_close,
 198     .url_open        = zmq_proto_open,
 199     .url_read        = zmq_proto_read,
 200     .url_write       = zmq_proto_write,
 201     .priv_data_size  = sizeof(ZMQContext),
 202     .priv_data_class = &zmq_context_class,
 203     .flags           = URL_PROTOCOL_FLAG_NETWORK,
 204 };