From patchwork Tue Apr 17 01:52:32 2018 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Stephan Holljes X-Patchwork-Id: 8473 Delivered-To: ffmpegpatchwork@gmail.com Received: by 10.2.1.85 with SMTP id c82csp1441881jad; Mon, 16 Apr 2018 19:00:50 -0700 (PDT) X-Google-Smtp-Source: AIpwx4/qRQPHIqCmVwsALjoKPkcvc4e+Xp6J7NIRofv5Dfdi7S+mjNjpLrSyTVrLa8e5EYyo5pRS X-Received: by 10.223.224.203 with SMTP id e11mr114686wri.2.1523930450716; Mon, 16 Apr 2018 19:00:50 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1523930450; cv=none; d=google.com; s=arc-20160816; b=x2LRUfTs79qrjc8pTC3q5v8+39oo5pLapwJBd6LNikdqGLl3bQ1ECPjKKity6lXXii UPtPgvymtLE85rsF9Ob1rttH4S7HslTHBeLBNqkcaYKqQ+oP42Z3x/IDi45YjIVHw3XI /Q4E6CiJ3WvpV7vzkZtCVRQ9bH0VsxpUYxHjkghR1foWY89NM8WU2sAHJgImCVrwOBT5 mKsMpJPMwnM4uaJqcFebUT5T22kNMAwxBN1YshftkuncLrpe/0b3WcZfKJ3fRci5M9CQ BtvRGvX+U9cswPDnWBZyDgA5vSL5P9hYAzO+D4Ho3Q+p4CwMMnjFHebkiSUo1CUf81kc edeA== ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=google.com; s=arc-20160816; h=sender:errors-to:content-transfer-encoding:mime-version:cc:reply-to :list-subscribe:list-help:list-post:list-archive:list-unsubscribe :list-id:precedence:subject:references:in-reply-to:message-id:date :to:from:dkim-signature:delivered-to:arc-authentication-results; bh=7ZRtPwtoErX6gdh9/xZlKWLtTGZPcqXAwYDjndhOOhM=; b=iG18fgjgEf5pple9O2jFwtlvtKt+V4hgdWQqx9/YSszjLeV5XFLo22IUHL6XmXJs1P IpT95axe2Y1ROg7Za6FtJIuLfWCo7MZzM9rgDt8ANK5BQABTSt/v/s5b9ZEV+TwrYHQX 64gsWDlfK4/sIVUCqGAzRGec80ZkThDWIy4ZuSK0WSEM2gX6vggaN0GtsHlu9/8mIw8o wOGGnDgjK5QtR9fumh+r4HT1iuIymqGxI0NHTbz2eMkO98cj9ryNZkS1BTfxS8UbRBi0 Yeqwc3X3rDBG/iBYhjpv79l8Gt4eTrgfes51v+3yG04rcGR36HPWUiGU41wnKfD8P8UA 94nA== ARC-Authentication-Results: i=1; mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=sDE7u2HD; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org; dmarc=fail (p=QUARANTINE sp=QUARANTINE dis=NONE) header.from=googlemail.com Return-Path: Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org. [79.124.17.100]) by mx.google.com with ESMTP id o20si9877439wro.121.2018.04.16.19.00.50; Mon, 16 Apr 2018 19:00:50 -0700 (PDT) Received-SPF: pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) client-ip=79.124.17.100; Authentication-Results: mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=sDE7u2HD; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org; dmarc=fail (p=QUARANTINE sp=QUARANTINE dis=NONE) header.from=googlemail.com Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id A1A96689E69; Tue, 17 Apr 2018 05:00:22 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail-wr0-f179.google.com (mail-wr0-f179.google.com [209.85.128.179]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id C9D59689BB5 for ; Tue, 17 Apr 2018 05:00:15 +0300 (EEST) Received: by mail-wr0-f179.google.com with SMTP id d1so30986800wrj.13 for ; Mon, 16 Apr 2018 19:00:42 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=googlemail.com; s=20161025; h=from:to:cc:subject:date:message-id:in-reply-to:references; bh=eIDmqbDAGZatxUSWSP7mfpgm/hvSgVS8hXNtGG7aC24=; b=sDE7u2HDkr3p0S7PwQTv3Dh58kvEMS7be+OK8HUbLr12PFicdVnhfKf5zMX5tV5rDi jIk2FvmVfPFpuQTkN8YOHSB9Slmmb3THg06i4W6wUgmTG/2t6d/wvsd9U7x/tHshPhJs xax2nKsaLD4m9/SQSzdQlE/PsbCHe8bsOgCSfsDAiB6Ke4Hf2M0v/phx/VaTGj/LaauM 1S2wsaeMZpgHPnGyM/LgRvgQTl3f2IpfMS5Fzqsw+sHXvorg9KNif2d3HtI1I9JXk4wA 1LAKLlWvLls10GwbJWLdgCfDGZiIiB7r+3x0//egTGAuD9qg+5Gg59AfBBZWFmKTV0PJ C3gQ== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:from:to:cc:subject:date:message-id:in-reply-to :references; bh=eIDmqbDAGZatxUSWSP7mfpgm/hvSgVS8hXNtGG7aC24=; b=euTgpMiZlZMr4C3Gf/Wgh+QMdidij25J/EJ1Uq0qHs37+xRXEcZuH28mUv9FWUkRta 5IKIkiLZK65VnLXGgclDSa/2FSBa8hshuUxKA35LprJ85bOy7wOucH5gv8dSPgipgGyM caOlRQ/s7jnkPqHS+EgiSR41TAGE4TIy6kiMyNoUxSyvlqg7Yu2Nwkek5TozYaBN/gt5 Do4VJGR54PHaOTyf+NwpMQjZ/tws1V88KVGMTWYIYB5fvkPbxgtuXu9rpwIRRej0YFfS kob8W0l0lqklb4imJPLJni18V2h1ueKi5835kEtdtDJ8awnyOPPFb43W+FQNBhUkhMZS /00w== X-Gm-Message-State: ALQs6tBFQN/9aPOspz7inolwBMyyJGfWiWTcgTvQe5nZZDqcWfqDyYCL op3FhxHU4XS8QYeLS11gFiSvvw== X-Received: by 10.28.241.18 with SMTP id p18mr354514wmh.40.1523929979472; Mon, 16 Apr 2018 18:52:59 -0700 (PDT) Received: from localhost.localdomain (HSI-KBW-046-005-002-245.hsi8.kabel-badenwuerttemberg.de. [46.5.2.245]) by smtp.gmail.com with ESMTPSA id b47sm11845123wrg.13.2018.04.16.18.52.58 (version=TLS1_2 cipher=ECDHE-RSA-AES128-GCM-SHA256 bits=128/128); Mon, 16 Apr 2018 18:52:58 -0700 (PDT) From: Stephan Holljes To: ffmpeg-devel@ffmpeg.org Date: Tue, 17 Apr 2018 03:52:32 +0200 Message-Id: <20180417015233.31983-4-klaxa1337@googlemail.com> X-Mailer: git-send-email 2.16.2 In-Reply-To: <20180417015233.31983-1-klaxa1337@googlemail.com> References: <20180412133549.19939-1-klaxa1337@googlemail.com> <20180417015233.31983-1-klaxa1337@googlemail.com> Subject: [FFmpeg-devel] [PATCH 3/4] ffserver: Implement ffserver and add Makefile X-BeenThere: ffmpeg-devel@ffmpeg.org X-Mailman-Version: 2.1.20 Precedence: list List-Id: FFmpeg development discussions and patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: FFmpeg development discussions and patches Cc: Stephan Holljes MIME-Version: 1.0 Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" --- Makefile | 15 ++ ffserver.c | 477 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 492 insertions(+) create mode 100644 Makefile create mode 100644 ffserver.c diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b077039 --- /dev/null +++ b/Makefile @@ -0,0 +1,15 @@ +all: ffserver +LAV_FLAGS = $(shell pkg-config --libs --cflags libavformat libavcodec libavutil) +# LAV_FLAGS = -L/usr/local/lib -lavcodec -lavformat -lavutil + +ffserver: segment.o publisher.o ffserver.c + cc -g -Wall $(LAV_FLAGS) -lpthread -o ffserver segment.o publisher.o ffserver.c + +segment.o: segment.c segment.h + cc -g -Wall $(LAV_FLAGS) -lpthread -c segment.c + +publisher.o: publisher.c publisher.h + cc -g -Wall $(LAV_FLAGS) -lpthread -c publisher.c + +clean: + rm *.o ffserver diff --git a/ffserver.c b/ffserver.c new file mode 100644 index 0000000..b0ff00e --- /dev/null +++ b/ffserver.c @@ -0,0 +1,477 @@ +/* + * 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 + */ + +/** + * @file + * multimedia server based on the FFmpeg libraries + */ + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "segment.h" +#include "publisher.h" + +#define BUFFER_SECS 30 +#define LISTEN_TIMEOUT_MSEC 1000 + +struct ReadInfo { + struct PublisherContext *pub; + AVFormatContext *ifmt_ctx; + char *in_filename; +}; + +struct WriteInfo { + struct PublisherContext *pub; + int thread_id; +}; + +struct AcceptInfo { + struct PublisherContext *pub; + AVFormatContext *ifmt_ctx; + const char *out_uri; +}; + + +void *read_thread(void *arg) +{ + struct ReadInfo *info = (struct ReadInfo*) arg; + AVFormatContext *ifmt_ctx = info->ifmt_ctx; + int ret, i; + int video_idx = -1; + int id = 0; + int64_t pts, now, start; + int64_t *ts; + struct Segment *seg = NULL; + AVPacket pkt; + AVStream *in_stream; + AVRational tb; + tb.num = 1; + tb.den = AV_TIME_BASE; + AVStream *stream; + AVCodecParameters *params; + enum AVMediaType type; + + if ((ret = avformat_find_stream_info(ifmt_ctx, NULL)) < 0) { + av_log(ifmt_ctx, AV_LOG_ERROR, "Could not get input stream info.\n"); + goto end; + } + + av_log(ifmt_ctx, AV_LOG_INFO, "Finding video stream.\n"); + for (i = 0; i < ifmt_ctx->nb_streams; i++) { + av_log(ifmt_ctx, AV_LOG_DEBUG, "Checking stream %d\n", i); + stream = ifmt_ctx->streams[i]; + params = stream->codecpar; + type = params->codec_type; + if (type == AVMEDIA_TYPE_VIDEO) { + video_idx = i; + break; + } + } + if (video_idx == -1) { + av_log(ifmt_ctx, AV_LOG_ERROR, "No video stream found.\n"); + goto end; + } + + + // All information needed to start segmenting the file is gathered now. + // start BUFFER_SECS seconds "in the past" to "catch up" to real-time. Has no effect on streamed sources. + start = av_gettime_relative() - BUFFER_SECS * AV_TIME_BASE; + + // segmenting main-loop + + for (;;) { + ret = av_read_frame(ifmt_ctx, &pkt); + if (ret < 0) + break; + + in_stream = ifmt_ctx->streams[pkt.stream_index]; + if (pkt.pts == AV_NOPTS_VALUE) { + pkt.pts = 0; + } + if (pkt.dts == AV_NOPTS_VALUE) { + pkt.dts = 0; + } + + // current pts + pts = av_rescale_q(pkt.pts, in_stream->time_base, tb); + + // current stream "uptime" + now = av_gettime_relative() - start; + + // simulate real-time reading + while (pts > now) { + usleep(1000); + now = av_gettime_relative() - start; + } + + // keyframe or first Segment + if ((pkt.flags & AV_PKT_FLAG_KEY && pkt.stream_index == video_idx) || !seg) { + if (seg) { + segment_close(seg); + publisher_push_segment(info->pub, seg); + av_log(NULL, AV_LOG_DEBUG, "New segment pushed.\n"); + publish(info->pub); + av_log(NULL, AV_LOG_DEBUG, "Published new segment.\n"); + } + segment_init(&seg, ifmt_ctx); + seg->id = id++; + av_log(NULL, AV_LOG_DEBUG, "Starting new segment, id: %d\n", seg->id); + } + + ts = av_dynarray2_add((void **)&seg->ts, &seg->ts_len, sizeof(int64_t), + (const void *)&pkt.dts); + if (!ts) { + av_log(seg->fmt_ctx, AV_LOG_ERROR, "could not write dts\n."); + goto end; + } + + ts = av_dynarray2_add((void **)&seg->ts, &seg->ts_len, sizeof(int64_t), + (const void *)&pkt.pts); + if (!ts) { + av_log(seg->fmt_ctx, AV_LOG_ERROR, "could not write pts\n."); + goto end; + } + ret = av_write_frame(seg->fmt_ctx, &pkt); + av_packet_unref(&pkt); + if (ret < 0) { + av_log(seg->fmt_ctx, AV_LOG_ERROR, "av_write_frame() failed.\n"); + goto end; + } + } + + if (ret < 0 && ret != AVERROR_EOF) { + av_log(seg->fmt_ctx, AV_LOG_ERROR, "Error occurred during read: %s\n", av_err2str(ret)); + goto end; + } + + segment_close(seg); + publisher_push_segment(info->pub, seg); + publish(info->pub); + + +end: + avformat_close_input(&ifmt_ctx); + info->pub->shutdown = 1; + return NULL; +} + +void write_segment(struct Client *c) +{ + struct Segment *seg; + int ret; + int pkt_count = 0; + if (av_fifo_size(c->buffer) > 0) { + AVFormatContext *fmt_ctx; + AVIOContext *avio_ctx; + AVPacket pkt; + struct SegmentReadInfo info; + unsigned char *avio_buffer; + + av_fifo_generic_peek(c->buffer, &seg, sizeof(struct Segment*), NULL); + client_set_state(c, BUSY); + c->current_segment_id = seg->id; + info.buf = seg->buf; + info.left = seg->size; + + if (!(fmt_ctx = avformat_alloc_context())) { + av_log(NULL, AV_LOG_ERROR, "Could not allocate format context\n"); + client_disconnect(c, 0); + return; + } + + avio_buffer = (unsigned char*) av_malloc(AV_BUFSIZE); + avio_ctx = avio_alloc_context(avio_buffer, AV_BUFSIZE, 0, &info, &segment_read, NULL, NULL); + + fmt_ctx->pb = avio_ctx; + ret = avformat_open_input(&fmt_ctx, NULL, seg->ifmt, NULL); + if (ret < 0) { + av_log(avio_ctx, AV_LOG_ERROR, "Could not open input\n"); + av_free(avio_ctx->buffer); + avio_context_free(&avio_ctx); + client_disconnect(c, 0); + return; + } + + ret = avformat_find_stream_info(fmt_ctx, NULL); + if (ret < 0) { + av_log(fmt_ctx, AV_LOG_ERROR, "Could not find stream information\n"); + av_free(avio_ctx->buffer); + avio_context_free(&avio_ctx); + client_disconnect(c, 0); + return; + } + + av_log(fmt_ctx, AV_LOG_DEBUG, "Client: %d, Segment: %d\n", c->id, seg->id); + + for (;;) { + ret = av_read_frame(fmt_ctx, &pkt); + if (ret < 0) + break; + + pkt.dts = seg->ts[pkt_count]; + pkt.pts = seg->ts[pkt_count+1]; + pkt_count += 2; + ret = av_write_frame(c->ofmt_ctx, &pkt); + av_packet_unref(&pkt); + if (ret < 0) { + av_log(fmt_ctx, AV_LOG_ERROR, "write_frame failed, disconnecting client: %d\n", c->id); + avformat_close_input(&fmt_ctx); + av_free(avio_ctx->buffer); + avio_context_free(&avio_ctx); + client_disconnect(c, 0); + return; + } + } + avformat_close_input(&fmt_ctx); + av_free(avio_ctx->buffer); + avformat_free_context(fmt_ctx); + avio_context_free(&avio_ctx); + av_fifo_drain(c->buffer, sizeof(struct Segment*)); + segment_unref(seg); + client_set_state(c, WRITABLE); + } else { + client_set_state(c, WAIT); + } +} + +void *accept_thread(void *arg) +{ + struct AcceptInfo *info = (struct AcceptInfo*) arg; + const char *out_uri = info->out_uri; + char *method, *resource; + char status[4096]; + AVIOContext *client; + AVIOContext *server = NULL; + AVFormatContext *ofmt_ctx = NULL; + AVOutputFormat *ofmt; + AVDictionary *opts = NULL; + AVDictionary *mkvopts = NULL; + AVStream *in_stream, *out_stream; + AVCodecContext *codec_ctx; + int ret, i, reply_code, handshake; + + if ((ret = av_dict_set(&opts, "listen", "2", 0)) < 0) { + av_log(opts, AV_LOG_ERROR, "Failed to set listen mode for server: %s\n", av_err2str(ret)); + return NULL; + } + + if ((ret = av_dict_set_int(&opts, "listen_timeout", LISTEN_TIMEOUT_MSEC, 0)) < 0) { + av_log(opts, AV_LOG_ERROR, "Failed to set listen_timeout for server: %s\n", av_err2str(ret)); + return NULL; + } + + if ((ret = avio_open2(&server, out_uri, AVIO_FLAG_WRITE, NULL, &opts)) < 0) { + av_log(server, AV_LOG_ERROR, "Failed to open server: %s\n", av_err2str(ret)); + return NULL; + } + + for (;;) { + if (info->pub->shutdown) + break; + publisher_gen_status_json(info->pub, status); + av_log(server, AV_LOG_INFO, status); + reply_code = 200; + client = NULL; + av_log(server, AV_LOG_DEBUG, "Accepting new clients.\n"); + + if ((ret = avio_accept(server, &client)) < 0) { + av_log(server, AV_LOG_DEBUG, "Timeout or error, retrying to accept.\n"); + continue; + } + + client->seekable = 0; + if ((ret = av_dict_set(&mkvopts, "live", "1", 0)) < 0) { + av_log(client, AV_LOG_ERROR, "Failed to set live mode for matroska: %s\n", av_err2str(ret)); + continue; + } + + if (publisher_reserve_client(info->pub)) { + av_log(client, AV_LOG_WARNING, "No more client slots free, Returning 503.\n"); + reply_code = 503; + } + + while ((handshake = avio_handshake(client)) > 0) { + av_opt_get(client, "method", AV_OPT_SEARCH_CHILDREN, &method); + av_opt_get(client, "resource", AV_OPT_SEARCH_CHILDREN, &resource); + av_log(client, AV_LOG_DEBUG, "method: %s resource: %s\n", method, resource); + if (method && strlen(method) && strncmp("GET", method, 3)) { + reply_code = 400; + } + av_free(method); + av_free(resource); + } + + if (handshake < 0) + reply_code = 400; + + if ((ret = av_opt_set_int(client, "reply_code", reply_code, AV_OPT_SEARCH_CHILDREN)) < 0) { + av_log(client, AV_LOG_ERROR, "Failed to set reply_code: %s.\n", av_err2str(ret)); + continue; + } + + if (reply_code != 200) { + publisher_cancel_reserve(info->pub); + avio_close(client); + continue; + } + + avformat_alloc_output_context2(&ofmt_ctx, NULL, "matroska", NULL); + if (!ofmt_ctx) { + av_log(client, AV_LOG_ERROR, "Could not allocate output format context.\n"); + continue; + } + + ofmt_ctx->flags |= AVFMT_FLAG_GENPTS; + ofmt = ofmt_ctx->oformat; + ofmt->flags &= AVFMT_NOFILE; + + for (i = 0; i < info->ifmt_ctx->nb_streams; i++) { + in_stream = info->ifmt_ctx->streams[i]; + codec_ctx = avcodec_alloc_context3(NULL); + avcodec_parameters_to_context(codec_ctx, in_stream->codecpar); + out_stream = avformat_new_stream(ofmt_ctx, codec_ctx->codec); + avcodec_free_context(&codec_ctx); + + if (!out_stream) { + av_log(client, AV_LOG_ERROR, "Could not allocate output stream.\n"); + continue; + } + + ret = avcodec_parameters_copy(out_stream->codecpar, in_stream->codecpar); + if (ret < 0) { + av_log(client, AV_LOG_ERROR, "Failed to copy context from input to output stream codec context: %s.\n", av_err2str(ret)); + continue; + } + av_dict_copy(&out_stream->metadata, in_stream->metadata, 0); + } + + ofmt_ctx->pb = client; + ret = avformat_write_header(ofmt_ctx, &mkvopts); + if (ret < 0) { + av_log(client, AV_LOG_ERROR, "Could not write header to client: %s.\n", av_err2str(ret)); + continue; + } + publisher_add_client(info->pub, ofmt_ctx); + ofmt_ctx = NULL; + + } + av_log(server, AV_LOG_INFO, "Shutting down http server.\n"); + ret = avio_close(server); + av_log(NULL, AV_LOG_INFO, "Shut down http server: %d\n", ret); + return NULL; +} + +void *write_thread(void *arg) +{ + struct WriteInfo *info = (struct WriteInfo*) arg; + int i, nb_free; + struct Client *c; + for(;;) { + nb_free = 0; + usleep(500000); + av_log(NULL, AV_LOG_DEBUG, "Checking clients, thread: %d\n", info->thread_id); + for (i = 0; i < MAX_CLIENTS; i++) { + c = &info->pub->clients[i]; + switch(c->state) { + case WRITABLE: + write_segment(c); + if (info->pub->shutdown && info->pub->current_segment_id == c->current_segment_id) { + client_disconnect(c, 1); + } + continue; + case FREE: + nb_free++; + default: + continue; + } + } + if (info->pub->shutdown && nb_free == MAX_CLIENTS) + break; + } + + return NULL; +} + + +int main(int argc, char *argv[]) +{ + struct ReadInfo rinfo; + struct AcceptInfo ainfo; + struct WriteInfo *winfos; + struct PublisherContext *pub; + int ret, i; + pthread_t r_thread, a_thread; + pthread_t *w_threads; + + AVFormatContext *ifmt_ctx = NULL; + + rinfo.in_filename = "pipe:0"; + ainfo.out_uri = "http://0:8080"; + if (argc > 1) + rinfo.in_filename = argv[1]; + + avformat_network_init(); + + av_log_set_level(AV_LOG_INFO); + + if ((ret = avformat_open_input(&ifmt_ctx, rinfo.in_filename, NULL, NULL))) { + av_log(NULL, AV_LOG_ERROR, "main: Could not open input\n"); + return 1; + } + + publisher_init(&pub); + + rinfo.ifmt_ctx = ifmt_ctx; + rinfo.pub = pub; + ainfo.ifmt_ctx = ifmt_ctx; + ainfo.pub = pub; + + w_threads = (pthread_t*) av_malloc(sizeof(pthread_t) * pub->nb_threads); + winfos = (struct WriteInfo*) av_malloc(sizeof(struct WriteInfo) * pub->nb_threads); + + for (i = 0; i < pub->nb_threads; i++) { + winfos[i].pub = pub; + winfos[i].thread_id = i; + pthread_create(&w_threads[i], NULL, write_thread, &winfos[i]); + } + + pthread_create(&r_thread, NULL, read_thread, &rinfo); + + accept_thread(&ainfo); + + pthread_join(r_thread, NULL); + + for (i = 0; i < pub->nb_threads; i++) { + pthread_join(w_threads[i], NULL); + } + av_free(w_threads); + av_free(winfos); + + publisher_freep(&pub); + return 0; +}