From patchwork Thu May 10 15:41:23 2018 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Stephan Holljes X-Patchwork-Id: 8909 Delivered-To: ffmpegpatchwork@gmail.com Received: by 2002:a02:155:0:0:0:0:0 with SMTP id c82-v6csp1215966jad; Thu, 10 May 2018 08:42:08 -0700 (PDT) X-Google-Smtp-Source: AB8JxZqYbZpP6fq/7p91JJ3tj6xOXnjvkIRlyxZb5BqRptUeBP/tm8GN+OUm4Lw78WxvWw4+qHPb X-Received: by 2002:adf:cd08:: with SMTP id w8-v6mr1768963wrm.187.1525966928724; Thu, 10 May 2018 08:42:08 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1525966928; cv=none; d=google.com; s=arc-20160816; b=VwUsw/SufP1dCEyzUPW1D/yQwfk+tdDqVtyuRaohCr+pl2SoQwGXKeKsWiW5ZLBOKN jkyRQWkOs7JSdWN3DXEHB0fwjsSpOkYhLQZF9ciptlxgDjJaWLts0YDKkgu6DZ3U7i2n Q7aw+xJbltaQc2BRkuVJwljAuYP5OORoDxEoOp/2NrUV2NYzAmcCU2mIP1y4rDEK0qd2 S5mybUb85JmaHencHR+T+WYoDZED5PtacEZj1bJKiE8Hw3IUOmVPNJFm5YkQc96LFRgx mWERysbr29F+Mcep/mSLEGN9ouIj+SdKnQS1OyrqVwjFrZX37ROsW64MpMY1T4DDOVdr Zy/A== 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=xwT2VFw+Se3tDeeCoKBHdrZlCNJ7fne/8pJ9huiZw0Q=; b=EO2IcIrXIofJSJQhrZvpivSKHa7KdzkxbtOjyuwYd+R2lBhvRTf/HX4vCEWnPaaBs+ oGr32GcSFxdJsDjVs7EdkjafJTNSSCrNoZJQSKotTLt5ViM1grUIA3zK5kk1o18W76dL fb7Wfv/VmEE5cEv05uTMY1W92dWnrfMm/EoGZJPHQMRwyepduty4MfsC+gr+Ku/KAZTr 3eYkeL5MGNw12cowb313CcwCMEjhGycaqqonsUEbaDKU/ZeZHLtlQTE++E8536LLLFoJ wMVH/rO1UzsXAQm8voVT6YKPno8hBM61li6Ckf4Xmbo2FpGtWlN3GwQs2FdeN8o+9y3+ P2iw== ARC-Authentication-Results: i=1; mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=aknaNnSR; 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 e79-v6si1001646wmc.193.2018.05.10.08.42.08; Thu, 10 May 2018 08:42:08 -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=aknaNnSR; 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 9B35368A83E; Thu, 10 May 2018 18:41:19 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail-wm0-f42.google.com (mail-wm0-f42.google.com [74.125.82.42]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id 7110468A83E for ; Thu, 10 May 2018 18:41:12 +0300 (EEST) Received: by mail-wm0-f42.google.com with SMTP id x12-v6so27930754wmc.0 for ; Thu, 10 May 2018 08:41:48 -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=1hYp29ElhWUmePkBhmW4Q2Xzg7nsGRJwK3TwDFLqJek=; b=aknaNnSRWkY0iygUTsRZqlYE41o9kdecRY+4PktLPjDuHg/ZRTd8VN/ok9ZiluCDnD XCePEi9NCYX16IceIw8XcEaPxOLZet6biO2oRpxdfZgFLO15lj6ExNoAnwWR76hbrk6m U0HiN1wOI143wvxvQ+XS+f3SAB1zrRukPbwVYZNAP88N8ORq1wyWXL+UsXv6FdZQGnnn S8OP12DHK29sEfp+Ve6g+sx+Rr14xgDTPL4TyLTGd81gMQdAaJLZ908K3eAjSQBAfskC 8kPx+2lKJHZYPbSPFrQPxP0DF2d7DXenT5fvTa3k4IP1VvMNm/hrKzz5jdvuXOxjNvd3 DcuA== 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=1hYp29ElhWUmePkBhmW4Q2Xzg7nsGRJwK3TwDFLqJek=; b=bvYPp2nBlTmytDbUgofT0nFw5zVI9Uq0lBfPtgI5WfvB+a2FPN1cBDab7infL/8CUU hHiHqDqRIp9c+RmMHxNvyAFBvWqZqI3X6RE1pLHt72DYJEXuRKk9tmoA113R2cBC4k+K I1nYFa/k3IrOQEHuWmTzeSAWBcHEtaSHDNNDMmjvb/IBCxxwivyHM4BbVF+1IRy6uR1a 7hFRvmPkwx/dOqCwoJE0+4qWj0pBLV4J8vkObKZpS3DZoQAw2ZAJ0FGvRucuZagZsnsZ /74OwyC7CFQeWLeM96vRTXMI7Tv3sqrfTRyuK/IN/MhxKXSlHCFzN+Q3/SC7RAuJ+r3M FiDQ== X-Gm-Message-State: ALKqPwcw9XTs9F156MY/SrmjT4Ulf+eJUgftLXn/WUktmVyAJh86fOzQ 6Qovz3cxNA9vwGWyWHgdlLJzbw== X-Received: by 2002:a1c:f509:: with SMTP id t9-v6mr1730080wmh.13.1525966907716; Thu, 10 May 2018 08:41:47 -0700 (PDT) Received: from localhost.localdomain (cebit2017.teco.kit.edu. [129.13.169.177]) by smtp.gmail.com with ESMTPSA id x63-v6sm1853472wma.25.2018.05.10.08.41.45 (version=TLS1_2 cipher=ECDHE-RSA-AES128-GCM-SHA256 bits=128/128); Thu, 10 May 2018 08:41:46 -0700 (PDT) From: Stephan Holljes To: ffmpeg-devel@ffmpeg.org Date: Thu, 10 May 2018 17:41:23 +0200 Message-Id: <20180510154126.30789-3-klaxa1337@googlemail.com> X-Mailer: git-send-email 2.16.2 In-Reply-To: <20180510154126.30789-1-klaxa1337@googlemail.com> References: <20180510154126.30789-1-klaxa1337@googlemail.com> Subject: [FFmpeg-devel] [PATCH 2/5] ffserver: Implement publisher 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" Signed-off-by: Stephan Holljes --- publisher.c | 306 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ publisher.h | 171 +++++++++++++++++++++++++++++++++ 2 files changed, 477 insertions(+) create mode 100644 publisher.c create mode 100644 publisher.h diff --git a/publisher.c b/publisher.c new file mode 100644 index 0000000..f637d23 --- /dev/null +++ b/publisher.c @@ -0,0 +1,306 @@ +/* + * 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 + */ + +#include "publisher.h" +#include "segment.h" +#include + +void client_log(struct Client *c) +{ + av_log(NULL, AV_LOG_INFO, "State: "); + switch(c->state) { + case FREE: + av_log(NULL, AV_LOG_INFO, "FREE\n"); + break; + case RESERVED: + av_log(NULL, AV_LOG_INFO, "RESERVED\n"); + break; + case WAIT: + av_log(NULL, AV_LOG_INFO, "WAIT\n"); + break; + case WRITABLE: + av_log(NULL, AV_LOG_INFO, "WRITABLE\n"); + break; + case BUSY: + av_log(NULL, AV_LOG_INFO, "BUSY\n"); + break; + case BUFFER_FULL: + av_log(NULL, AV_LOG_INFO, "BUFFER_FULL\n"); + break; + default: + av_log(NULL, AV_LOG_INFO, "UNKOWN\n"); + break; + } +} + +void client_disconnect(struct Client *c, int write_trailer) +{ + struct Segment *seg; + if (write_trailer) + av_write_trailer(c->ofmt_ctx); + c->ffinfo->httpd->close(c->ffinfo->server, c->ffinfo->client); + av_free(c->ofmt_ctx->pb->buffer); + avformat_free_context(c->ofmt_ctx); + av_free(c->ffinfo); + c->ofmt_ctx = NULL; + c->ffinfo = NULL; + pthread_mutex_lock(&c->buffer_lock); + while(av_fifo_size(c->buffer)) { + av_fifo_generic_read(c->buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + pthread_mutex_unlock(&c->buffer_lock); + client_set_state(c, FREE); + c->current_segment_id = -1; +} + +void client_set_state(struct Client *c, enum State state) +{ + pthread_mutex_lock(&c->state_lock); + c->state = state; + pthread_mutex_unlock(&c->state_lock); +} + +void client_push_segment(struct Client *c, struct Segment *seg) +{ + pthread_mutex_lock(&c->buffer_lock); + if (av_fifo_space(c->buffer) == 0) { + av_log(NULL, AV_LOG_WARNING, "Client buffer full, dropping Segment.\n"); + client_set_state(c, BUFFER_FULL); + pthread_mutex_unlock(&c->buffer_lock); + return; + } + segment_ref(seg); + av_fifo_generic_write(c->buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&c->buffer_lock); + client_set_state(c, WRITABLE); +} + +void publisher_init(struct PublisherContext **pub) +{ + int i; + struct PublisherContext *pc = (struct PublisherContext*) av_malloc(sizeof(struct PublisherContext)); + pc->nb_threads = 4; + pc->current_segment_id = -1; + pc->shutdown = 0; + pc->buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + pc->fs_buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + pthread_mutex_init(&pc->buffer_lock, NULL); + pthread_mutex_init(&pc->fs_buffer_lock, NULL); + for (i = 0; i < MAX_CLIENTS; i++) { + struct Client *c = &pc->clients[i]; + c->buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + c->ofmt_ctx = NULL; + c->ffinfo = NULL; + c->id = i; + c->current_segment_id = -1; + pthread_mutex_init(&c->state_lock, NULL); + pthread_mutex_init(&c->buffer_lock, NULL); + client_set_state(c, FREE); + } + *pub = pc; +} + +void publisher_push_segment(struct PublisherContext *pub, struct Segment *seg) +{ + struct Segment *drop; + pthread_mutex_lock(&pub->buffer_lock); + pthread_mutex_lock(&pub->fs_buffer_lock); + av_fifo_generic_write(pub->buffer, &seg, sizeof(struct Segment*), NULL); + segment_ref(seg); + if (av_fifo_size(pub->fs_buffer) >= BUFFER_SEGMENTS * sizeof(struct Segment*)) { + av_fifo_generic_read(pub->fs_buffer, &drop, sizeof(struct Segment*), NULL); + segment_unref(drop); + } + av_fifo_generic_write(pub->fs_buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&pub->buffer_lock); + pthread_mutex_unlock(&pub->fs_buffer_lock); + segment_ref(seg); +} + +int publisher_reserve_client(struct PublisherContext *pub) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case FREE: + client_set_state(&pub->clients[i], RESERVED); + return 0; + default: + continue; + } + } + return 1; +} + +void publisher_cancel_reserve(struct PublisherContext *pub) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case RESERVED: + client_set_state(&pub->clients[i], FREE); + return; + default: + continue; + } + } + return; +} + +void client_push_prebuffer(struct PublisherContext *pub, struct Client *c) +{ + int off; + int size; + struct Segment *seg; + pthread_mutex_lock(&pub->fs_buffer_lock); + size = av_fifo_size(pub->fs_buffer); + for (off = 0; off < size; off += sizeof(struct Segment*)) { + av_fifo_generic_peek_at(pub->fs_buffer, &seg, off, sizeof(struct Segment*), NULL); + client_push_segment(c, seg); + } + pthread_mutex_unlock(&pub->fs_buffer_lock); +} + +void publisher_add_client(struct PublisherContext *pub, AVFormatContext *ofmt_ctx, struct FFServerInfo *ffinfo) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case RESERVED: + pub->clients[i].ofmt_ctx = ofmt_ctx; + pub->clients[i].ffinfo = ffinfo; + client_set_state(&pub->clients[i], WRITABLE); + client_push_prebuffer(pub, &pub->clients[i]); + return; + default: + continue; + } + } +} + +void publisher_free(struct PublisherContext *pub) +{ + int i; + struct Segment *seg; + pthread_mutex_lock(&pub->buffer_lock); + while(av_fifo_size(pub->buffer)) { + av_fifo_generic_read(pub->buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + av_fifo_freep(&pub->buffer); + pthread_mutex_unlock(&pub->buffer_lock); + + pthread_mutex_lock(&pub->fs_buffer_lock); + while(av_fifo_size(pub->fs_buffer)) { + av_fifo_generic_read(pub->fs_buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + av_fifo_freep(&pub->fs_buffer); + for (i = 0; i < MAX_CLIENTS; i++) { + av_fifo_freep(&pub->clients[i].buffer); + } + pthread_mutex_unlock(&pub->fs_buffer_lock); + av_free(pub); + return; +} + +void publisher_freep(struct PublisherContext **pub) +{ + publisher_free(*pub); + *pub = NULL; + return; +} + +void publish(struct PublisherContext *pub) +{ + int i; + struct Segment *seg; + char filename[128] = {0}; + pthread_mutex_lock(&pub->buffer_lock); + av_log(NULL, AV_LOG_DEBUG, "pub->buffer size: %d\n", av_fifo_size(pub->buffer)); + if (av_fifo_size(pub->buffer) == 0) { + pthread_mutex_unlock(&pub->buffer_lock); + return; + } + av_fifo_generic_read(pub->buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&pub->buffer_lock); + if (seg) { + pub->current_segment_id = seg->id; + snprintf(filename, 127, "segment-%04d.mkv", seg->id); +// segment_save(seg, filename); + + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case BUFFER_FULL: + av_log(pub, AV_LOG_WARNING, "Dropping segment for client %d, buffer full.\n", i); + continue; + case WAIT: + case WRITABLE: + client_push_segment(&pub->clients[i], seg); + default: + continue; + } + } + segment_unref(seg); + } +} + +void publisher_gen_status_json(struct PublisherContext *pub, char *status) +{ + int states[STATE_NB] = {0}; + int current_read = 0, newest_write = 0, oldest_write = 0; + int i; + struct Client *c; + + current_read = pub->current_segment_id; + oldest_write = current_read; + + for (i = 0; i < MAX_CLIENTS; i++) { + c = &pub->clients[i]; + if (c->current_segment_id > 0 && c->current_segment_id < oldest_write) { + oldest_write = c->current_segment_id; + } + if (c->current_segment_id > newest_write) { + newest_write = c->current_segment_id; + } + states[c->state]++; + } + + + snprintf(status, 4095, + "{\n\t\"free\": %d,\n" + "\t\"reserved\": %d,\n" + "\t\"wait\": %d,\n" + "\t\"writable\": %d,\n" + "\t\"busy\": %d,\n" + "\t\"buffer_full\": %d,\n" + "\t\"current_read\": %d,\n" + "\t\"newest_write\": %d,\n" + "\t\"oldest_write\": %d\n" + "}\n", + states[FREE], + states[RESERVED], + states[WAIT], + states[WRITABLE], + states[BUSY], + states[BUFFER_FULL], + current_read, + newest_write, + oldest_write); +} diff --git a/publisher.h b/publisher.h new file mode 100644 index 0000000..5d158a8 --- /dev/null +++ b/publisher.h @@ -0,0 +1,171 @@ +/* + * 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 + */ + +#ifndef PUBLISHER_H +#define PUBLISHER_H + +#include +#include +#include +#include "segment.h" +#include "httpd.h" + +#define MAX_CLIENTS 16 +#define MAX_SEGMENTS 16 +#define BUFFER_SEGMENTS 10 + +/* Client State enum */ + +enum State { + FREE, // no client connected + RESERVED, // reserved for a client that just connected + WAIT, // up to date, no new Segments to write + WRITABLE, // buffer is not full, new Segments can be pushed + BUSY, // currently writing to this client + BUFFER_FULL, // client buffer is full, new Segments will be dropped + STATE_NB +}; + + +/* struct containing server and client info per client AVIOContext */ + +struct FFServerInfo { + struct HTTPDInterface *httpd; + void *server; + void *client; +}; + + +struct Client { + AVFormatContext *ofmt_ctx; // writable AVFormatContext, basically our tcp connection to the client + AVFifoBuffer *buffer; // Client buffer of Segment references + char *method; + char *resource; + void *httpd_data; // The data needed by the httpd implementation for this client. + struct FFServerInfo *ffinfo; + enum State state; + pthread_mutex_t buffer_lock; + pthread_mutex_t state_lock; + int id; + int current_segment_id; // The stream-based id of the segment that has last been worked on. +}; + +struct PublisherContext { + struct Client clients[MAX_CLIENTS]; // currently compile-time configuration, easly made dynamic with malloc? + AVFifoBuffer *buffer; // publisher buffer for new Segments + AVFifoBuffer *fs_buffer; // fast start buffer + pthread_mutex_t buffer_lock; + pthread_mutex_t fs_buffer_lock; + int nb_threads; + int current_segment_id; + int shutdown; // indicate shutdown, gracefully close client connections and files and exit +}; + +/** + * Log a client's stats to the console. + * + * @param c pointer to the client to print + */ +void client_log(struct Client *c); + +/** + * Disconnect a client. + * + * @param c pointer to the client to disconnect. + */ +void client_disconnect(struct Client *c, int write_trailer); + +/** + * Set a client's state. Note: This is protected by mutex locks. + * + * @param c pointer to the client to set the state of + * @param state the state to set the client to + */ +void client_set_state(struct Client *c, enum State state); + +/** + * Allocate and initialize a PublisherContext + * + * @param pub pointer to a pointer to a PublisherContext. It will be allocated and initialized. + */ +void publisher_init(struct PublisherContext **pub); + +/** + * Push a Segment to a PublisherContext. + * + * @param pub pointer to a PublisherContext + * @param seg pointer to the Segment to add + */ +void publisher_push_segment(struct PublisherContext *pub, struct Segment *seg); + +/** + * Reserve a slot in the client struct of a PublisherContext. May fail if the number + * of maximum clients has been reached. + * + * @param pub pointer to a PublisherContext + * @return 0 in case of success, 1 in case of failure + */ +int publisher_reserve_client(struct PublisherContext *pub); + +/** + * Cancel a single reservation. This can be used if a client spot was reserved, but the client + * unexpectedly disconnects or sends an invalid request. + * + * @param pub pointer to a PublisherContext + */ +void publisher_cancel_reserve(struct PublisherContext *pub); + +/** + * Add a client by its ofmt_ctx. This initializes an element in the client struct of the PublisherContext + * that has been reserved prior to calling this function. + * + * @param pub pointer to a PublisherContext + * @param ofmt_ctx AVFormatContext of a client + * @param ffinfo pointer to struct containing custom IO information for server independent write implementation + */ +void publisher_add_client(struct PublisherContext *pub, AVFormatContext *ofmt_ctx, struct FFServerInfo *ffinfo); + +/** + * Free buffers and associated client buffers. + * + * @param pub pointer to the PublisherContext to free + */ +void publisher_free(struct PublisherContext *pub); + +/** + * Free buffers and associated client buffers and set *pub to NULL. + * + * @param pub pointer to the PublisherContext pointer to free + */ +void publisher_freep(struct PublisherContext **pub); + +/** + * Signal to the PublisherContext to check its buffer and publish pending Segments. + * + * @param pub pointer to a PublisherContext + */ +void publish(struct PublisherContext *pub); + +/** + * Print the current client and file reading status to a json string. + * @param pub pointer to a PublisherContext + * @param status string of at least 4096 bytes size. + */ +void publisher_gen_status_json(struct PublisherContext *pub, char *status); + +#endif // PUBLISHER_H