Muxer wasm work basically.

src/wasms/muxer.c

 #include "wasm.h"
 #include <stdio.h>
 #include <stdint.h>
+#include <stdbool.h>
 
 #include <libavformat/avformat.h>
 
+#include "pktBuffer.h"
 #include "basic/container/list.h"
 #include "streamContext.h"
 #include "MovMem/mm.h"
 
+#define OK 0
+#define ERROR 1
+
+
+typedef enum {
+    CONTEXT_STOP = 0,
+    CONTEXT_RUN = 1,
+    CONTEXT_FIN = 2,
+    CONTEXT_FAIL = 3,
+} ContextState;
+
+typedef struct Context {
+    ContextState state;
+    StreamContext *sc;
+} Context;
+
+StreamContext *octx;
+
 static int numOfStreams;
-static bool *finished;
-static bool *inited;
-static bool *failed;
+static int finNum;
+static int initNum;
+static int inProcContext = 0;
 
 static List *protos;
 static List *ctxs;
-static List *pktBuffer;
+static PktBuffer *pktBuffer;
 
-EM_PROT_API(int) muxInit(int numOfStreams_) {
+EM_PORT_API(int) muxInit(int numOfStreams_) {
 
     numOfStreams = numOfStreams_;
 
     protos    = createList();
     ctxs      = createList();
-    pktBuffer = createList();
-
-    finished = (bool*)malloc(numOfStreams);
-    inited   = (bool*)malloc(numOfStreams);
-    failed   = (bool*)malloc(numOfStreams);
+    pktBuffer = createPktBuffer(numOfStreams);
 
     for (int i = 0; i < numOfStreams; ++i) {
         listPush(protos, createMMProto());
+
+        Context *ctx = (Context*)malloc(sizeof(Context));
+        ctx->state = CONTEXT_STOP;
+        ctx->sc = NULL;
+
+        listPush(ctxs, ctx);
     }
 
-    return 0;
+    return OK;
 }
 
 EM_PORT_API(int) muxPush(int sIdx, uint8_t *data, size_t size) {
+
     MMProto *p = listGet(protos, sIdx);
-    if (!p) mmpPush(data, size);
-    else return 1;
+    if (p) {
+        printf("MUXER WASM: Proto %d Push Size %ld\n", sIdx, size);
+        mmpPush(p, data, size);
+        printf("MUXER WASM: Proto %d Buffer Size is %d\n", sIdx, mmpBufferSize(p));
+    } else {
+        printf("MUXER WASM: muxPush fail to get proto %d\n", sIdx);
+        return 1;
+    }
 
     return 0;
 }
 
+AVPacket *createNULLPacket() {
+    AVPacket *emptyPkt = av_packet_alloc();
+    emptyPkt->data = NULL;
+    emptyPkt->size = 0;
+
+    return emptyPkt;
+}
+
+int buffering() {
+
+    static int PKT_UPPER_BOUND = 100;
+
+    int ret = 0, total = 0, pktsALoop;
+    AVPacket *pkt = NULL;
+
+    while (true) {
+
+        pktsALoop = 0;
+
+        for (int i = 0; i < numOfStreams; ++i) {
+            Context *ctx = listGet(ctxs, i);
+            if (ctx->state != CONTEXT_RUN)
+                continue;
+
+            printf("MUXER WASM: Buffer for context %d\n", i);
+
+            pkt = pkt == NULL ? av_packet_alloc() : pkt;
+            ret = streamReadFrame(ctx->sc, pkt);
+            if (ret < 0) {
+                if (ret == AVERROR_EOF) {
+                    printf("MUXER WASM: buffering() EOF\n");
+                    ctx->state = CONTEXT_FIN;
+                    pktBufPush(pktBuffer, i, createNULLPacket());
+                    break;
+                }
+
+                continue;
+            }
+
+            printf("MUXER WASM: BUFFER INTO %d\n", i);
+            /* Buffer the packet */
+            pktBufPush(pktBuffer, i, pkt);
+            pkt = NULL;
+
+            ++pktsALoop;
+        }
+
+        total += pktsALoop;
+
+        if (pktsALoop < numOfStreams ||
+            total >= PKT_UPPER_BOUND) {
+
+            break;
+        }
+    }
+
+    if (pkt) av_packet_free(&pkt);
+
+    return OK;
+}
+
+
+int contextInitializes() {
+
+    for (int i = 0; i < numOfStreams; ++i) {
+        Context *ctx = listGet(ctxs, i);
+        MMProto *proto = listGet(protos, i);
+
+        if (ctx->state != CONTEXT_STOP ||
+            mmpBufferSize(proto) < 25000) {
+
+            continue;
+        }
+
+        ProtoInfo info = {
+            .proto = (Proto*)listGet(protos, i),
+            .destructor = (ProtoDestructor)destroyMMProto,
+            .ofmt = NULL
+        };
+
+        ctx->sc = createInStreamContext("", info, AVMEDIA_TYPE_VIDEO);
+        ctx->state = CONTEXT_RUN;
+
+        printf("MUXER WASM: CONTEXT %d init\n", i);
+
+        ++initNum;
+
+        if (!octx) {
+            octx = createOutStreamContext(
+                "/tmp/output.mp4",
+                (ProtoInfo){NULL,NULL,NULL},
+                AVMEDIA_TYPE_VIDEO);
+            streamNewStream(octx, streamStream(ctx->sc)->codecpar);
+            streamWriteHeader(octx);
+        }
+    }
+
+    return OK;
+}
+
+int writeToOFormat() {
+
+    int i = inProcContext;
+
+    while (true) {
+        AVPacket *pkt = pktBufPop(pktBuffer, i);
+
+        if (pkt == NULL) {
+            printf("MUXER WASM: writeToOFormat() NULL PKT\n");
+            goto EXIT;
+        }
+
+        if (pkt->data == NULL && pkt->size == 0) {
+            /* EOF */
+            if (++finNum == numOfStreams)
+                goto FINISH;
+        }
+
+        streamWriteFrame(octx, pkt);
+        printf("MUXER WASM: WriteToOutput(): Index %d...Done\n", i);
+
+        av_packet_free(&pkt);
+
+      NEXT:
+        i = (i+1) % numOfStreams;
+    }
+
+
+  EXIT:
+    inProcContext = i;
+    return OK;
+
+  FINISH:
+    streamWriteTrailer(octx);
+    streamClose(&octx);
+
+    return OK;
+}
+
+
 EM_PORT_API(int) muxStep(void) {
+    printf("MuxStep Start\n");
+
+
+    printf("MuxStep Buffering()\n");
+    /* Buffer all packets from muxPush() into
+     * Packet Buffer */
+    buffering();
+
 
+    printf("MuxStep Init()\n");
+    /* Do Streaming Context Initialization
+     * if required */
+    if (initNum < numOfStreams)
+        contextInitializes();
+
+    printf("MuxStep Write()\n");
+    /* Write into output format in order of
+     * stream index */
+    writeToOFormat();
+
+    printf("MuxStep Done\n");
+
+    return OK;
 }
 
 EM_PORT_API(int) eof(int i) {
-
+    return OK;
 }
 
 
-int main(int argc, char *argv[]) {}
+int main(int argc, char *argv[]) { return 0; }

src/wasms/pktBuffer.c

+#include "pktBuffer.h"
+
+static void destructor(void *v) {
+    if (v) av_packet_free((AVPacket**)&v);
+}
+
+
+PktBuffer* createPktBuffer(int numOfStreams_) {
+    PktBuffer *buf = (PktBuffer*)malloc(sizeof(PktBuffer));
+    memset(buf, 0, sizeof(PktBuffer));
+
+    buf->lists = (List**)malloc(numOfStreams_*sizeof(List*));
+    for (int i = 0; i < numOfStreams_; ++i) {
+        buf->lists[i] = createList();
+        listSetDestructor(buf->lists[i], destructor);
+    }
+
+    buf->numOfStreams = numOfStreams_;
+
+    return buf;
+}
+
+void destroyPktBuffer(PktBuffer **buf_) {
+
+    PktBuffer *buf = *buf_;
+
+    int numOfStreams = pktBufNumOfStream(buf);
+    for (int i = 0; i < numOfStreams; ++i)
+        destroyList(&buf->lists[i]);
+
+    free(buf->lists);
+    free(buf);
+    *buf_ = NULL;
+}
+
+int pktBufPush(PktBuffer *buf, int sid, AVPacket *pkt) {
+    if (sid >= pktBufNumOfStream(buf)) {
+        return PKT_BUFFER_ERROR;
+    }
+
+    List *s = pktBufGetStream(buf, sid);
+    listPush(s, pkt);
+
+    ++buf->numOfPackets;
+
+    return PKT_BUFFER_OK;
+}
+
+AVPacket* pktBufPop(PktBuffer *buf, int sid) {
+    if (sid >= pktBufNumOfStream(buf)) {
+        return NULL;
+    }
+
+    List *s = pktBufGetStream(buf, sid);
+    AVPacket *pkt = listPop(s);
+
+    return pkt;
+}

src/wasms/pktBuffer.h

+#include "basic/container/list.h"
+#include <libavformat/avformat.h>
+
+#ifndef PKTBUFFER_H
+#define PKTBUFFER_H
+
+typedef enum {
+    PKT_BUFFER_OK = 0,
+    PKT_BUFFER_ERROR = 1,
+} PKT_BUFFER_STATE;
+
+typedef struct PktBuffer {
+    List **lists;
+    int numOfStreams;
+    int numOfPackets;
+} PktBuffer;
+
+/* Member functions implement as macro */
+#define pktBufNumOfPkt(P) ((P)->numOfPackets)
+#define pktBufNumOfStream(P) ((P)->numOfStreams)
+#define pktBufGetStream(P, I) ((P)->lists[(I)])
+
+/* Prototypes */
+PktBuffer* createPktBuffer(int numOfStreams_);
+void destroyPktBuffer(PktBuffer **buf);
+int pktBufPush(PktBuffer *buf, int sid, AVPacket *pkt);
+AVPacket* pktBufPop(PktBuffer *buf, int sid);
+
+
+#endif /* PKTBUFFER_H */

wasm-build.sh

@@ -123,7 +123,7 @@ FLAGS_MUXER=(
   -Wno-deprecated-declarations -Wno-pointer-sign -Wno-implicit-int-float-conversion -Wno-switch -Wno-parentheses -Qunused-arguments
   -lavdevice -lavfilter -lavformat -lavcodec -lswresample -lswscale -lavutil -lpostproc -lm -lx264 -lz -lffmpegprotos
 
-  $WASM_DIR/muxer.c
+  $WASM_DIR/muxer.c $WASM_DIR/pktBuffer.c
 
   -s FORCE_FILESYSTEM=1
   -s WASM=1