ffserver.c

/*
 * Copyright (c) 2018 Stephan Holljes
 *
 * 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 <stdio.h>
#include <unistd.h>
#include <inttypes.h>
#include <pthread.h>
#include <sys/stat.h>
#include <errno.h>
#include <string.h>

#include <libavutil/log.h>
#include <libavutil/timestamp.h>
#include <libavutil/time.h>
#include <libavutil/opt.h>
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
#include <libavutil/file.h>

#include "segment.h"
#include "publisher.h"
#include "fileserver.h"
#include "httpd.h"
#include "configreader.h"

#define BUFFER_SECS 10
#define LISTEN_TIMEOUT_MSEC 1000
#define AUDIO_ONLY_SEGMENT_SECONDS 2


/** Information needed for a read thread */
struct ReadInfo {
    struct PublisherContext *pub;
    struct StreamConfig *config;
    AVFormatContext *ifmt_ctx;
    struct FileserverContext *fs;
    char *input_uri;
    char *server_name;
};

/** Information needed for a write thread */
struct WriteInfo {
    struct PublisherContext *pub;
    int thread_id;
};

/** Information needed for an accept thread, currently this runs on the main thread */
struct AcceptInfo {
    struct PublisherContext **pubs;
    struct HTTPDInterface *httpd;
    AVFormatContext **ifmt_ctxs;
    struct HTTPDConfig *config;
    struct FileserverContext *fs;
    int nb_pub; /** number of publishers (streams) equal to number of ifmt_ctx */
};


/** Map avio write callback to httpd write callback */
int ffserver_write(void *opaque, unsigned char *buf, int buf_size)
{
    struct FFServerInfo *info = (struct FFServerInfo*) opaque;
    return info->httpd->write(info->server, info->client, buf, buf_size);
}


/** Read a file and feed it into publishers and other muxers */
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 audio_only = 0;
    int id = 0;
    int64_t pts, pts_tmp, dts_tmp, now, start, last_cut = 0;
    int64_t *ts;
    char playlist_dirname[1024];
    char playlist_filename[1024];
    AVFormatContext *ofmt_ctx[FMT_NB] = { 0 }; // some may be left unused
    struct Segment *seg = NULL;
    AVPacket pkt;
    AVStream *in_stream, *out_stream;
    AVRational tb = {1, AV_TIME_BASE};
    AVStream *stream;
    int stream_formats[FMT_NB] = { 0 };

    for (i = 0; i < info->config->nb_formats; i++)
        stream_formats[info->config->formats[i]] = 1;

    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];
        if (stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
            video_idx = i;
            break;
        }
    }
    if (video_idx == -1)
        audio_only = 1;

    if (stream_formats[FMT_HLS] || stream_formats[FMT_DASH]) {
        snprintf(playlist_dirname, 1024, "%s/%s", info->server_name, info->config->stream_name);
        ret = mkdir(playlist_dirname, 0755);
        if (ret < 0 && errno != EEXIST) {
            av_log(NULL, AV_LOG_WARNING, "Could not create stream directory (%s) dropping hls/dash\n", strerror(errno));
            stream_formats[FMT_HLS] = 0;
            stream_formats[FMT_DASH] = 0;
        }
    }

    if (stream_formats[FMT_HLS]) {
        snprintf(playlist_filename, 1024, "%s/%s/%s_hls.m3u8", info->server_name, info->config->stream_name,
                                                                                  info->config->stream_name);
        avformat_alloc_output_context2(&ofmt_ctx[FMT_HLS], NULL, "hls", playlist_filename);

        if (!ofmt_ctx[FMT_HLS]) {
            av_log(NULL, AV_LOG_ERROR, "Could not allocate hls output context.\n");
            goto end;
        }

        for (i = 0; i < ifmt_ctx->nb_streams; i++) {
            in_stream = ifmt_ctx->streams[i];
            out_stream = avformat_new_stream(ofmt_ctx[FMT_HLS], NULL);
            if (!out_stream) {
                av_log(ofmt_ctx[FMT_HLS], AV_LOG_WARNING, "Failed allocating output stream\n");
                continue;
            }
            ret = avcodec_parameters_copy(out_stream->codecpar, in_stream->codecpar);
            if (ret < 0) {
                av_log(ofmt_ctx[FMT_HLS], AV_LOG_WARNING, "Failed to copy context from input to output stream codec context\n");
                continue;
            }
            out_stream->codecpar->codec_tag = 0;
            if (out_stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
                if (in_stream->sample_aspect_ratio.num)
                    out_stream->sample_aspect_ratio = in_stream->sample_aspect_ratio;
                out_stream->avg_frame_rate = in_stream->avg_frame_rate;
                out_stream->r_frame_rate = in_stream->r_frame_rate;
            }
            av_dict_copy(&out_stream->metadata, in_stream->metadata, 0);
        }
        av_dict_copy(&ofmt_ctx[FMT_HLS]->metadata, ifmt_ctx->metadata, 0);
        ret = avformat_write_header(ofmt_ctx[FMT_HLS], NULL);
        if (ret < 0) {
            av_log(ofmt_ctx[FMT_HLS], AV_LOG_WARNING, "Error occured while writing header: %s\n", av_err2str(ret));
        }

        av_log(ofmt_ctx[FMT_HLS], AV_LOG_DEBUG, "Initialized hls.\n");
    }

    if (stream_formats[FMT_DASH]) {
        snprintf(playlist_filename, 1024, "%s/%s/%s_dash.mpd", info->server_name, info->config->stream_name,
                                                                                  info->config->stream_name);
        avformat_alloc_output_context2(&ofmt_ctx[FMT_DASH], NULL, "dash", playlist_filename);

        if (!ofmt_ctx[FMT_DASH]) {
            av_log(NULL, AV_LOG_ERROR, "Could not allocate hls output context.\n");
            goto end;
        }

        for (i = 0; i < ifmt_ctx->nb_streams; i++) {
            in_stream = ifmt_ctx->streams[i];
            out_stream = avformat_new_stream(ofmt_ctx[FMT_DASH], NULL);
            if (!out_stream) {
                av_log(ofmt_ctx[FMT_DASH], AV_LOG_WARNING, "Failed allocating output stream\n");
                continue;
            }
            ret = avcodec_parameters_copy(out_stream->codecpar, in_stream->codecpar);
            if (ret < 0) {
                av_log(ofmt_ctx[FMT_DASH], AV_LOG_WARNING, "Failed to copy context from input to output stream codec context\n");
                continue;
            }
            out_stream->codecpar->codec_tag = 0;
            if (out_stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
                if (in_stream->sample_aspect_ratio.num)
                    out_stream->sample_aspect_ratio = in_stream->sample_aspect_ratio;
                out_stream->avg_frame_rate = in_stream->avg_frame_rate;
                out_stream->r_frame_rate = in_stream->r_frame_rate;
            }
            av_dict_copy(&out_stream->metadata, in_stream->metadata, 0);
        }
        av_dict_copy(&ofmt_ctx[FMT_DASH]->metadata, ifmt_ctx->metadata, 0);
        ret = avformat_write_header(ofmt_ctx[FMT_DASH], NULL);
        if (ret < 0) {
            av_log(ofmt_ctx[FMT_DASH], AV_LOG_WARNING, "Error occured while writing header: %s\n", av_err2str(ret));
        }

        av_log(ofmt_ctx[FMT_DASH], AV_LOG_DEBUG, "Initialized dash.\n");
    }

    // 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 in AV_TIME_BASE
        pts = av_rescale_q_rnd(pkt.pts, in_stream->time_base, tb, AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);

        // current stream "uptime"
        now = av_gettime_relative() - start;

        // simulate real-time reading
        while (pts > now) {
            usleep(1000);
            now = av_gettime_relative() - start;
        }

        if (stream_formats[FMT_MATROSKA]) {
            // keyframe or first Segment or audio_only and more than AUDIO_ONLY_SEGMENT_SECONDS passed since last cut
            if ((pkt.flags & AV_PKT_FLAG_KEY && pkt.stream_index == video_idx) || !seg ||
                (audio_only && pts - last_cut >= AUDIO_ONLY_SEGMENT_SECONDS * AV_TIME_BASE)) {
                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");
                }
                last_cut = pts;
                segment_init(&seg, ifmt_ctx);
                if (!seg) {
                    av_log(NULL, AV_LOG_ERROR, "Segment initialization failed, shutting down.\n");
                    goto end;
                }
                seg->id = id++;
                av_log(NULL, AV_LOG_DEBUG, "Starting new segment, id: %d\n", seg->id);
            }
            dts_tmp = av_rescale_q_rnd(pkt.dts, in_stream->time_base, tb, AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            ts = av_dynarray2_add((void **)&seg->ts, &seg->ts_len, sizeof(int64_t),
                                (const void *)&dts_tmp);
            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 *)&pts);
            if (!ts) {
                av_log(seg->fmt_ctx, AV_LOG_ERROR, "could not write pts\n.");
                goto end;
            }
            pts_tmp = pkt.pts;
            dts_tmp = pkt.dts;
            pkt.pts = av_rescale_q_rnd(pkt.pts, in_stream->time_base, seg->fmt_ctx->streams[pkt.stream_index]->time_base, AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.dts = av_rescale_q_rnd(pkt.dts, in_stream->time_base, seg->fmt_ctx->streams[pkt.stream_index]->time_base, AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            ret = av_write_frame(seg->fmt_ctx, &pkt);
            pkt.pts = pts_tmp;
            pkt.dts = dts_tmp;
            if (ret < 0) {
                av_log(seg->fmt_ctx, AV_LOG_ERROR, "av_write_frame() failed.\n");
                goto end;
            }
        }

        if (stream_formats[FMT_DASH]) {
            pts_tmp = pkt.pts;
            dts_tmp = pkt.dts;
            pkt.pts = av_rescale_q_rnd(pkt.pts, in_stream->time_base, ofmt_ctx[FMT_DASH]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.dts = av_rescale_q_rnd(pkt.dts, in_stream->time_base, ofmt_ctx[FMT_DASH]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.duration = av_rescale_q_rnd(pkt.duration, in_stream->time_base, ofmt_ctx[FMT_DASH]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);

            ret = av_write_frame(ofmt_ctx[FMT_DASH], &pkt);
            pkt.pts = pts_tmp;
            pkt.dts = dts_tmp;
            if (ret < 0) {
                fprintf(stderr, "Error muxing packet\n");
                break;
            }
        }

        if (stream_formats[FMT_HLS]) {
            pkt.pts = av_rescale_q_rnd(pkt.pts, in_stream->time_base, ofmt_ctx[FMT_HLS]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.dts = av_rescale_q_rnd(pkt.dts, in_stream->time_base, ofmt_ctx[FMT_HLS]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.duration = av_rescale_q_rnd(pkt.duration, in_stream->time_base, ofmt_ctx[FMT_HLS]->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);

            ret = av_write_frame(ofmt_ctx[FMT_HLS], &pkt);
            if (ret < 0) {
                fprintf(stderr, "Error muxing packet\n");
                break;
            }
        }
        av_packet_unref(&pkt);
    }

    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;
    }
    if (stream_formats[FMT_MATROSKA]) {
        segment_close(seg);
        publisher_push_segment(info->pub, seg);
        publish(info->pub);
    }


end:
    avformat_close_input(&ifmt_ctx);
    if (info->pub)
        info->pub->shutdown = 1;
    for (i = 0; i < FMT_NB; i++) {
        if (ofmt_ctx[i]) {
            av_write_trailer(ofmt_ctx[i]);
            avformat_free_context(ofmt_ctx[i]);
        }
    }
    if (info->fs) {
        sleep(BUFFER_SECS);
        info->fs->shutdown = 1;
    }
    return NULL;
}


/** Write a segment to a client (mkv only). This muxes the input for the client */
void write_segment(struct Client *c)
{
    struct Segment *seg;
    int ret;
    int pkt_count = 0;
    AVRational tb = {1, AV_TIME_BASE};
    pthread_mutex_lock(&c->buffer_lock);
    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(seg), NULL);
        pthread_mutex_unlock(&c->buffer_lock);
        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 = av_malloc(AV_BUFSIZE);
        if (!avio_buffer) {
            av_log(fmt_ctx, AV_LOG_ERROR, "Could not allocate avio_buffer\n");
            avformat_free_context(fmt_ctx);
            client_disconnect(c, 0);
            return;
        }
        avio_ctx = avio_alloc_context(avio_buffer, AV_BUFSIZE, 0, &info, &segment_read, NULL, NULL);
        if (!avio_ctx) {
            av_log(fmt_ctx, AV_LOG_ERROR, "Could not allocate avio_ctx\n");
            avformat_free_context(fmt_ctx);
            av_free(avio_buffer);
            client_disconnect(c, 0);
            return;
        }
        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");
            avformat_close_input(&fmt_ctx);
            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");
            avformat_close_input(&fmt_ctx);
            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 = av_rescale_q_rnd(seg->ts[pkt_count], tb,
                                                        c->ofmt_ctx->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);
            pkt.pts = av_rescale_q_rnd(seg->ts[pkt_count+1], tb,
                                                        c->ofmt_ctx->streams[pkt.stream_index]->time_base,
                                                               AV_ROUND_NEAR_INF|AV_ROUND_PASS_MINMAX);

            pkt.pos = -1;
            pkt_count += 2;
            ret = av_interleaved_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;
            }
        }
        av_free(avio_ctx->buffer);
        avformat_close_input(&fmt_ctx);
        avio_context_free(&avio_ctx);
        pthread_mutex_lock(&c->buffer_lock);
        av_fifo_drain(c->buffer, sizeof(seg));
        pthread_mutex_unlock(&c->buffer_lock);
        segment_unref(seg);
        client_set_state(c, WRITABLE);
    } else {
        pthread_mutex_unlock(&c->buffer_lock);
        client_set_state(c, WAIT);
    }
}


/** Accept new clients and add them to the publisher or serve files */
void *accept_thread(void *arg)
{
    struct AcceptInfo *info = (struct AcceptInfo*) arg;
    struct FFServerInfo *ffinfo = NULL;
    struct PublisherContext *pub;
    char status[4096], requested_file[1024], sanitized_file[1024];
    char *stream_name, *resource;
    struct HTTPClient *client = NULL;
    void *server = NULL;
    AVIOContext *client_ctx = NULL;
    AVFormatContext *ofmt_ctx = NULL;
    AVFormatContext *ifmt_ctx;
    unsigned char *avio_buffer;
    AVOutputFormat *ofmt;
    AVDictionary *mkvopts = NULL;
    AVStream *in_stream, *out_stream;
    int ret, i, reply_code;
    int shutdown;
    struct HTTPDConfig *config = info->config;

    info->httpd->init(&server, *config);

    for (;;) {
        shutdown = 1;
        for (i = 0; i < config->nb_streams; i++) {
            if (info->pubs[i] && !info->pubs[i]->shutdown)
                shutdown = 0;
        }
        if (info->fs && !info->fs->shutdown)
            shutdown = 0;
        if (shutdown)
            break;
        for (i = 0; i < config->nb_streams; i++) {
            if (info->pubs[i]) {
                publisher_gen_status_json(info->pubs[i], status);
                av_log(NULL, AV_LOG_INFO, status);
            }
        }
        client = NULL;
        av_log(NULL, AV_LOG_DEBUG, "Accepting new clients.\n");
        reply_code = 200;

        if ((ret = info->httpd->accept(server, &client, NULL)) < 0) {
            if (ret == HTTPD_LISTEN_TIMEOUT) {
                continue;
            } else if (ret == HTTPD_CLIENT_ERROR) {
                info->httpd->close(server, client);
            }
            av_log(NULL, AV_LOG_WARNING, "Error during accept, retrying.\n");
            continue;
        }

        pub = NULL;
        ifmt_ctx = NULL;
        resource = client->resource;
        snprintf(requested_file, 1024, "%s", resource);
        for (i = 0; i < config->nb_streams; i++) {
            stream_name = info->config->streams[i].stream_name;
            //  skip leading '/'  ---v
            if (resource && strlen(resource) > strlen(stream_name)
                && !strncmp(resource + 1, stream_name, strlen(stream_name))) {
                resource++;
                while (resource && *resource++ != '/');
                if (strlen(resource) > 2 && !strncmp(resource, "mkv", 3)) {
                    pub = info->pubs[i];
                    ifmt_ctx = info->ifmt_ctxs[i];
                    memset(requested_file, 0, 1024);
                    break;
                } else if (strlen(resource) > 2 && !strncmp(resource, "hls", 3)) {
                    snprintf(requested_file, 1024, "/%s/%s_hls.m3u8", stream_name, stream_name);
                    break;
                } else if (strlen(resource) > 3 && !strncmp(resource, "dash", 4)) {
                    snprintf(requested_file, 1024, "/%s/%s_dash.mpd", stream_name, stream_name);
                    break;
                }
            }
        }


        if ((!pub || !ifmt_ctx) && !requested_file[0]) {
            av_log(client_ctx, AV_LOG_WARNING, "No suitable publisher found for resource: %s.\n",
                                                        client->resource ? client->resource : "(null)");
            reply_code = 404;
        }


        if (pub && ifmt_ctx && publisher_reserve_client(pub)) {
            av_log(client_ctx, AV_LOG_WARNING, "No more client slots free, Returning 503.\n");
            reply_code = 503;
        }

        if (reply_code != 200) {
            if (pub && ifmt_ctx)
                publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            continue;
        }

        ffinfo = av_malloc(sizeof(*ffinfo));
        if (!ffinfo) {
            av_log(client_ctx, AV_LOG_ERROR, "Could not allocate FFServerInfo struct.\n");
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            continue;
        }
        ffinfo->httpd = info->httpd;
        ffinfo->client = client;
        ffinfo->server = server;


        // try to serve file
        if (info->fs && requested_file[0]) {
            snprintf(sanitized_file, 1024, "%s", requested_file);
            resource = requested_file;
            while(resource && *resource == '/') {
                resource++;
            }

            snprintf(sanitized_file, 1024, "%s", resource);
            fileserver_schedule(info->fs, ffinfo, sanitized_file);
            continue;
        }

        if (!info->fs && requested_file[0]) {
            info->httpd->close(server, client);
            continue;
        }

        avio_buffer = av_malloc(AV_BUFSIZE);
        if (!avio_buffer) {
            av_log(client_ctx, AV_LOG_ERROR, "Could not allocate output format context.\n");
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            continue;
        }


        client_ctx = avio_alloc_context(avio_buffer, AV_BUFSIZE, 1, ffinfo, NULL, &ffserver_write, NULL);
        if (!client_ctx) {
            av_log(NULL, AV_LOG_ERROR, "Could not allocate output format context.\n");
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            av_free(client_ctx->buffer);
            avio_context_free(&client_ctx);
            av_free(ffinfo);
            continue;
        }
        avformat_alloc_output_context2(&ofmt_ctx, NULL, "matroska", NULL);
        if (!ofmt_ctx) {
            av_log(client_ctx, AV_LOG_ERROR, "Could not allocate output format context.\n");
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            avformat_free_context(ofmt_ctx);
            av_free(client_ctx->buffer);
            avio_context_free(&client_ctx);
            av_free(ffinfo);
            continue;
        }
        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));
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            avformat_free_context(ofmt_ctx);
            av_free(client_ctx->buffer);
            avio_context_free(&client_ctx);
            av_free(ffinfo);
            continue;
        }
        ofmt_ctx->flags |= AVFMT_FLAG_GENPTS;
        ofmt = ofmt_ctx->oformat;
        ofmt->flags |= AVFMT_NOFILE | AVFMT_FLAG_AUTO_BSF;

        for (i = 0; i < ifmt_ctx->nb_streams; i++) {
            in_stream = ifmt_ctx->streams[i];
            out_stream = avformat_new_stream(ofmt_ctx, NULL);

            if (!out_stream) {
                av_log(client, AV_LOG_ERROR, "Could not allocate output stream.\n");
                publisher_cancel_reserve(pub);
                info->httpd->close(server, client);
                avformat_free_context(ofmt_ctx);
                av_free(client_ctx->buffer);
                avio_context_free(&client_ctx);
                av_free(ffinfo);
                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));
                publisher_cancel_reserve(pub);
                info->httpd->close(server, client);
                avformat_free_context(ofmt_ctx);
                av_free(client_ctx->buffer);
                avio_context_free(&client_ctx);
                av_free(ffinfo);
                continue;
            }
            out_stream->codecpar->codec_tag = 0;
            if (out_stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
                if (in_stream->sample_aspect_ratio.num)
                    out_stream->sample_aspect_ratio = in_stream->sample_aspect_ratio;
                out_stream->avg_frame_rate = in_stream->avg_frame_rate;
                out_stream->r_frame_rate = in_stream->r_frame_rate;
            }
            av_dict_copy(&out_stream->metadata, in_stream->metadata, 0);
        }
        av_dict_copy(&ifmt_ctx->metadata, ofmt_ctx->metadata, 0);
        ofmt_ctx->pb = client_ctx;
        ret = avformat_write_header(ofmt_ctx, &mkvopts);
        if (ret < 0) {
            av_log(client_ctx, AV_LOG_ERROR, "Could not write header to client: %s.\n", av_err2str(ret));
            publisher_cancel_reserve(pub);
            info->httpd->close(server, client);
            avformat_free_context(ofmt_ctx);
            av_free(client_ctx->buffer);
            avio_context_free(&client_ctx);
            av_free(ffinfo);
            continue;
        }
        publisher_add_client(pub, ofmt_ctx, ffinfo);
        ofmt_ctx = NULL;

    }
    av_log(NULL, AV_LOG_INFO, "Shutting down http server.\n");
    info->httpd->shutdown(server);
    av_log(NULL, AV_LOG_INFO, "Shut down http server.\n");
    return NULL;
}


/** Check clients and write data to them if possible */
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:
                    client_set_state(c, BUSY);
                    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;
}

void *fileserver_thread(void *arg)
{
    struct FileserverContext *fs = (struct FileserverContext*) arg;
    int i, clients_served;
    struct FileserverClient *c;
    for (;;) {
        usleep(500000);
        clients_served = 0;
        for (i = 0; i < MAX_CLIENTS; i++) {
            c = &fs->clients[i];
            pthread_mutex_lock(&c->client_lock);
            if (c->buf) {
                c->ffinfo->httpd->write(c->ffinfo->server, c->ffinfo->client, c->buf, c->size);
                av_file_unmap(c->buf, c->size);
                c->ffinfo->httpd->close(c->ffinfo->server, c->ffinfo->client);
                c->buf = NULL;
                c->size = 0;
                av_freep(&c->ffinfo);
                clients_served++;
            }
            pthread_mutex_unlock(&c->client_lock);
        }
        av_log(NULL, AV_LOG_INFO, "Checked clients, fileserver-thread %s %d/%d served\n", fs->server_name, clients_served, MAX_CLIENTS);
        if (fs->shutdown) {
            printf("shutting down\n");
            break;
        }
    }

    return NULL;
}


/** Run a server given a HTTPDConfiguration */
void *run_server(void *arg) {
    struct AcceptInfo ainfo;
    struct ReadInfo *rinfos;
    struct WriteInfo **winfos_p;
    struct HTTPDConfig *config = (struct HTTPDConfig*) arg;
    struct PublisherContext **pubs;
    struct FileserverContext *fs = NULL;
    AVFormatContext **ifmt_ctxs;
    int ret, i, stream_index;
    int stream_formats[FMT_NB] = { 0 };
    pthread_t *r_threads;
    pthread_t **w_threads_p;
    pthread_t fs_thread = 0;

    pubs = av_mallocz_array(config->nb_streams, sizeof(struct PublisherContext*));
    if (!pubs) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate publishers\n");
        goto error_cleanup;
    }
    ifmt_ctxs = av_mallocz_array(config->nb_streams, sizeof(AVFormatContext*));
    if (!ifmt_ctxs) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate input format contexts.\n");
        goto error_cleanup;
    }

    for (stream_index = 0; stream_index < config->nb_streams; stream_index++) {
        for (i = 0; i < config->streams[stream_index].nb_formats; i++)
            stream_formats[config->streams[stream_index].formats[i]] = 1;
    }

    if (stream_formats[FMT_HLS] || stream_formats[FMT_DASH]) {
        fileserver_init(&fs, config->server_name);
        ret = mkdir(config->server_name, 0755);
        if (ret < 0 && errno != EEXIST) {
            av_log(NULL, AV_LOG_WARNING, "Could not create server directory (%d) dropping hls/dash\n", errno);
            stream_formats[FMT_HLS] = 0;
            stream_formats[FMT_DASH] = 0;
            fileserver_free(fs);
            fs = NULL;
        }
    }

    av_log_set_level(AV_LOG_INFO);

    ainfo.pubs = pubs;
    ainfo.fs = fs;
    ainfo.ifmt_ctxs = ifmt_ctxs;
    ainfo.nb_pub = config->nb_streams;
    ainfo.httpd = &lmhttpd;
    ainfo.config = config;

    rinfos = av_mallocz_array(config->nb_streams, sizeof(struct ReadInfo));
    if (!rinfos) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate read infos.\n");
        goto error_cleanup;
    }
    winfos_p = av_mallocz_array(config->nb_streams, sizeof(struct WriteInfo*));
    if (!winfos_p) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate write info pointers.\n");
        goto error_cleanup;
    }
    r_threads = av_mallocz_array(config->nb_streams, sizeof(pthread_t));
    if (!r_threads) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate read thread handles.\n");
        goto error_cleanup;
    }
    w_threads_p = av_mallocz_array(config->nb_streams, sizeof(pthread_t*));
    if (!w_threads_p) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate write thread handle pointers.\n");
        goto error_cleanup;
    }

    for (stream_index = 0; stream_index < config->nb_streams; stream_index++) {
        struct PublisherContext *pub = NULL;
        struct AVFormatContext *ifmt_ctx = NULL;
        struct ReadInfo rinfo;
        struct WriteInfo *winfos = NULL;
        pthread_t *w_threads = NULL;
        pthread_t r_thread;
        int stream_formats[FMT_NB] = { 0 };
        rinfo.input_uri = config->streams[stream_index].input_uri;
        rinfo.server_name = config->server_name;
        rinfo.config = &config->streams[stream_index];
        rinfo.fs = fs;

        for (i = 0; i < config->streams[stream_index].nb_formats; i++)
            stream_formats[config->streams[stream_index].formats[i]] = 1;

        if ((ret = avformat_open_input(&ifmt_ctx, rinfo.input_uri, NULL, NULL))) {
            av_log(NULL, AV_LOG_ERROR, "run_server: Could not open input\n");
            continue;
        }


        ifmt_ctxs[stream_index] = ifmt_ctx;
        if (stream_formats[FMT_MATROSKA])
            publisher_init(&pub, config->streams[stream_index].stream_name);

        pubs[stream_index] = pub;

        rinfo.ifmt_ctx = ifmt_ctx;
        rinfo.pub = pub;

        rinfos[stream_index] = rinfo;

        ret = pthread_create(&r_thread, NULL, read_thread, &rinfos[stream_index]);
        if (ret != 0) {
            pub->shutdown = 1;
            r_thread = 0;
            goto end;
        }
        r_threads[stream_index] = r_thread;

        if (stream_formats[FMT_MATROSKA]) {
            w_threads = av_mallocz_array(pub->nb_threads, sizeof(pthread_t));
            if (!w_threads) {
                av_log(NULL, AV_LOG_ERROR, "Could not allocate write thread handles.\n");
                continue;
            }
            winfos = av_mallocz_array(pub->nb_threads, sizeof(struct WriteInfo));
            if (!winfos) {
                av_log(NULL, AV_LOG_ERROR, "Could not allocate write infos.\n");
                continue;
            }
            w_threads_p[stream_index] = w_threads;
            winfos_p[stream_index] = winfos;

            for (i = 0; i < pub->nb_threads; i++) {
                winfos[i].pub = pub;
                winfos[i].thread_id = i;
                ret = pthread_create(&w_threads[i], NULL, write_thread, &winfos_p[stream_index][i]);
                if (ret != 0) {
                    pub->shutdown = 1;
                    w_threads[i] = 0;
                    goto end;
                }
            }
            w_threads_p[stream_index] = w_threads;
        }

    }
    if (stream_formats[FMT_HLS] || stream_formats[FMT_DASH]) {
        ret = pthread_create(&fs_thread, NULL, fileserver_thread, fs);
        if (ret != 0) {
            fs->shutdown = 1;
            fs_thread = 0;
            goto end;
        }
    }



    //pthread_create(&a_thread, NULL, accept_thread, &ainfo);
    accept_thread(&ainfo);

end:
    for (stream_index = 0; stream_index < config->nb_streams; stream_index++) {
        // in case of thread creation failure this might NULL
        if (r_threads[stream_index])
            pthread_join(r_threads[stream_index], NULL);
        if (pubs[stream_index]) {
            for (i = 0; i < pubs[stream_index]->nb_threads; i++) {
                // might also be NULL because of thread creation failure
                if (w_threads_p[stream_index][i])
                    pthread_join(w_threads_p[stream_index][i], NULL);
            }
        }
        av_free(winfos_p[stream_index]);
        av_free(w_threads_p[stream_index]);
        // pubs[stream_index] could be null if the file could not be opened or mkv was not requested
        if (pubs[stream_index])
            publisher_free(pubs[stream_index]);
    }

    if (fs_thread) {
        fs->shutdown = 1;
        pthread_join(fs_thread, NULL);
        fileserver_free(fs);
    }
error_cleanup:
    av_free(rinfos);
    av_free(winfos_p);
    av_free(r_threads);
    av_free(w_threads_p);
    av_free(pubs);
    av_free(ifmt_ctxs);
    return NULL;
}

int main(int argc, char *argv[])
{
    struct HTTPDConfig *configs;
    int nb_configs;
    pthread_t *server_threads;
    int i, ret;

    if (argc < 2) {
        printf("Usage: %s config.lua\n", argv[0]);
        return 1;
    }

    nb_configs = configs_read(&configs, argv[1]);
    if (nb_configs <= 0) {
        printf("No valid configurations parsed.\n");
        return 1;
    }
    server_threads = av_mallocz_array(nb_configs, sizeof(pthread_t));
    if (!server_threads) {
        av_log(NULL, AV_LOG_ERROR, "Could not allocate server thread handles.\n");
        return AVERROR(ENOMEM);
    }
    for (i = 0; i < nb_configs; i++) {
        config_dump(configs + i, stderr);
        ret = pthread_create(&server_threads[i], NULL, run_server, configs + i);
        if (ret != 0) {
            server_threads[i] = 0;
        }
    }

    for (i = 0; i < nb_configs; i++) {
        if (server_threads[i])
            pthread_join(server_threads[i], NULL);
        config_free(configs + i);
    }
    av_free(configs);
    av_free(server_threads);
    return 0;
}