xref: /third_party/ffmpeg/tools/zmqsend.c (revision cabdff1a) 
1/*
2 * Copyright (c) 2013 Stefano Sabatini
3 *
4 * This file is part of FFmpeg.
5 *
6 * FFmpeg is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
10 *
11 * FFmpeg is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with FFmpeg; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19 */
20
21#include "config.h"
22
23#include <zmq.h>
24
25#include "libavutil/mem.h"
26#include "libavutil/bprint.h"
27
28#if HAVE_UNISTD_H
29#include <unistd.h>             /* getopt */
30#endif
31
32#if !HAVE_GETOPT
33#include "compat/getopt.c"
34#endif
35
36/**
37 * @file
38 * zmq message sender example, meant to be used with the zmq filters
39 */
40
41static void usage(void)
42{
43    printf("send message to ZMQ recipient, to use with the zmq filters\n");
44    printf("usage: zmqsend [OPTIONS]\n");
45    printf("\n"
46           "Options:\n"
47           "-b ADDRESS        set bind address\n"
48           "-h                print this help\n"
49           "-i INFILE         set INFILE as input file, stdin if omitted\n");
50}
51
52int main(int argc, char **argv)
53{
54    AVBPrint src;
55    char *src_buf, *recv_buf;
56    int c;
57    int recv_buf_size, ret = 0;
58    void *zmq_ctx, *socket;
59    const char *bind_address = "tcp://localhost:5555";
60    const char *infilename = NULL;
61    FILE *infile = NULL;
62    zmq_msg_t msg;
63
64    while ((c = getopt(argc, argv, "b:hi:")) != -1) {
65        switch (c) {
66        case 'b':
67            bind_address = optarg;
68            break;
69        case 'h':
70            usage();
71            return 0;
72        case 'i':
73            infilename = optarg;
74            break;
75        case '?':
76            return 1;
77        }
78    }
79
80    if (!infilename || !strcmp(infilename, "-")) {
81        infilename = "stdin";
82        infile = stdin;
83    } else {
84        infile = fopen(infilename, "r");
85    }
86    if (!infile) {
87        av_log(NULL, AV_LOG_ERROR,
88               "Impossible to open input file '%s': %s\n", infilename, strerror(errno));
89        return 1;
90    }
91
92    zmq_ctx = zmq_ctx_new();
93    if (!zmq_ctx) {
94        av_log(NULL, AV_LOG_ERROR,
95               "Could not create ZMQ context: %s\n", zmq_strerror(errno));
96        return 1;
97    }
98
99    socket = zmq_socket(zmq_ctx, ZMQ_REQ);
100    if (!socket) {
101        av_log(NULL, AV_LOG_ERROR,
102               "Could not create ZMQ socket: %s\n", zmq_strerror(errno));
103        ret = 1;
104        goto end;
105    }
106
107    if (zmq_connect(socket, bind_address) == -1) {
108        av_log(NULL, AV_LOG_ERROR, "Could not bind ZMQ responder to address '%s': %s\n",
109               bind_address, zmq_strerror(errno));
110        ret = 1;
111        goto end;
112    }
113
114    /* grab the input and store it in src */
115    av_bprint_init(&src, 1, AV_BPRINT_SIZE_UNLIMITED);
116    while ((c = fgetc(infile)) != EOF)
117        av_bprint_chars(&src, c, 1);
118    av_bprint_chars(&src, 0, 1);
119
120    if (!av_bprint_is_complete(&src)) {
121        av_log(NULL, AV_LOG_ERROR, "Could not allocate a buffer for the source string\n");
122        av_bprint_finalize(&src, NULL);
123        ret = 1;
124        goto end;
125    }
126    av_bprint_finalize(&src, &src_buf);
127
128    if (zmq_send(socket, src_buf, strlen(src_buf), 0) == -1) {
129        av_log(NULL, AV_LOG_ERROR, "Could not send message: %s\n", zmq_strerror(errno));
130        ret = 1;
131        goto end;
132    }
133
134    if (zmq_msg_init(&msg) == -1) {
135        av_log(NULL, AV_LOG_ERROR,
136               "Could not initialize receiving message: %s\n", zmq_strerror(errno));
137        ret = 1;
138        goto end;
139    }
140
141    if (zmq_msg_recv(&msg, socket, 0) == -1) {
142        av_log(NULL, AV_LOG_ERROR,
143               "Could not receive message: %s\n", zmq_strerror(errno));
144        zmq_msg_close(&msg);
145        ret = 1;
146        goto end;
147    }
148
149    recv_buf_size = zmq_msg_size(&msg) + 1;
150    recv_buf = av_malloc(recv_buf_size);
151    if (!recv_buf) {
152        av_log(NULL, AV_LOG_ERROR,
153               "Could not allocate receiving message buffer\n");
154        zmq_msg_close(&msg);
155        ret = 1;
156        goto end;
157    }
158    memcpy(recv_buf, zmq_msg_data(&msg), recv_buf_size - 1);
159    recv_buf[recv_buf_size-1] = 0;
160    printf("%s\n", recv_buf);
161    zmq_msg_close(&msg);
162    av_free(recv_buf);
163
164end:
165    zmq_close(socket);
166    zmq_ctx_destroy(zmq_ctx);
167    return ret;
168}
169