#include <u.h>#include <libc.h>#include <thread.h>#include <bio.h>#include <libsec.h>#include "amf0.h"#include "ivf.h"#include "rtmp.h"#include "util.h" #define min(a,b) ((a)<(b)?(a):(b)) enum { Port = 1935, CSsz = 1536, ChunkDefault = 128, ChunkDesired = 65536, DataHdr = 0, DataFrame, Type0 = 0, Type1, Type2, Type3, CSUserCtl = 2, CSCtl = 3, CSData = 4, CbCommand = 0, CbTransID, CbObject, CbResponse, NumCb, /* UserControl */ CtlStreamBegin = 0, CtlStreamEOF, CtlStreamDry, CtlSetBufferLen, CtlStreamIsRecorded, CtlPingRequest = 6, CtlPingResponse, /* Message.type */ SetChunkSize = 1, Abort, Ack, UserControl, WindowAckSize, SetBandwidth, Audio = 8, Video, AMF3Metadata = 15, AMF3SharedObject, AMF3Command, AMF0Metadata, AMF0SharedObject, AMF0Command, Aggregate = 22, /* RTMP.bwlimit */ LimitHard = 0, LimitSoft, LimitDynamic, Biobufsz = 64*1024, /* FIXME don't know if it helps with anything */ Bufsz = 64*1024,}; typedef struct Buffer Buffer;typedef struct Command Command;typedef struct Message Message; #pragma varargck type "T" int#pragma varargck type "M" Message* struct Command { void (*cb)(RTMP *r, int ok, A₀ *a[NumCb], void *aux); char *code; void *aux; int tid; int cs; Command *prev, *next;}; struct Message { int type; int fmt; int cs; int sid; u32int ts; u8int *data; int sz; Command cmd;}; struct Buffer { Message msg; u8int *b, *p, *e; int bsz;}; struct RTMP { Biobufhdr; QLock; Buffer i; Buffer o; Channel *c; char *tcurl; char *app; char *inst; char *path; /* FIXME no idea what this is for */ int chunkin; int chunkout; int mode; int fd; int winacksz; int bw; u8int bwlimit; struct { int tid; Command *w; }cmds; int sps; int aacpd; u8int biobuf[Biobufsz];}; #define putnull() do{ r->o.p = a₀null(r->o.p, r->o.e); }while(0)#define putbyte(b) do{ r->o.p = a₀byte(r->o.p, r->o.e, b); }while(0)#define putdata(d, sz) do { r->o.p = a₀data(r->o.p, r->o.e, d, sz); }while(0)#define puti16(i) do{ r->o.p = a₀i16(r->o.p, r->o.e, i); }while(0)#define puti24(i) do{ r->o.p = a₀i24(r->o.p, r->o.e, i); }while(0)#define puti32(i) do{ r->o.p = a₀i32(r->o.p, r->o.e, i); }while(0)#define putnum(v) do{ r->o.p = a₀num(r->o.p, r->o.e, v); }while(0)#define putstr(s) do{ r->o.p = a₀str(r->o.p, r->o.e, s); }while(0)#define putarr() do{ r->o.p = a₀arr(r->o.p, r->o.e); }while(0)#define putobj() do{ r->o.p = a₀obj(r->o.p, r->o.e); }while(0)#define putend() do{ r->o.p = a₀end(r->o.p, r->o.e); }while(0)#define putkvnum(name, v) do{ r->o.p = a₀kvnum(r->o.p, r->o.e, name, v); }while(0)#define putkvstr(name, s) do{ r->o.p = a₀kvstr(r->o.p, r->o.e, name, s); }while(0)#define putkvbool(name, s) do{ r->o.p = a₀kvbool(r->o.p, r->o.e, name, s); }while(0) #define putcommand(name, cb_) do { \ putstr(name); \ putnum(r->o.msg.cmd.tid); \ putobj(); \ r->o.msg.cmd.cb = cb_; \}while(0) #define putcall(name) do { \ putstr(name); \ putnum(r->o.msg.cmd.tid); \}while(0) static int szs[] = { [Type3] = 0, [Type2] = 3, [Type1] = 7, [Type0] = 11,}; static char *msgtype2s[] = { [SetChunkSize] = "SetChunkSize", [Abort] = "Abort", [Ack] = "Ack", [UserControl] = "UserControl", [WindowAckSize] = "WindowAckSize", [SetBandwidth] = "SetBandwidth", [Audio] = "Audio", [Video] = "Video", [AMF3Metadata] = "AMF3Metadata", [AMF3SharedObject] = "AMF3SharedObject", [AMF3Command] = "AMF3Command", [AMF0Metadata] = "AMF0Metadata", [AMF0SharedObject] = "AMF0SharedObject", [AMF0Command] = "AMF0Command", [Aggregate] = "Aggregate",}; static char *ctl2s[] = { [CtlStreamBegin] = "StreamBegin", [CtlStreamEOF] = "StreamEOF", [CtlStreamDry] = "StreamDry", [CtlSetBufferLen] = "SetBufferLen", [CtlStreamIsRecorded] = "StreamIsRecorded", [CtlPingRequest] = "PingRequest", [CtlPingResponse] = "PingResponse",}; static char *bwlimit2s[] = { [LimitHard] = "hard", [LimitSoft] = "soft", [LimitDynamic] = "dynamic",}; static char *pubtype2s[] = { [PubLive] = "live", [PubAppend] = "append", [PubRecord] = "record",}; static voidnewmsg(RTMP *r, int type, int fmt, int cs){ memset(&r->o.msg, 0, sizeof(r->o.msg)); r->o.msg.type = type; r->o.msg.fmt = fmt; r->o.msg.cs = cs; r->o.p = r->o.b; if(type == AMF0Command) r->o.msg.cmd.tid = ++r->cmds.tid; else r->o.msg.cmd.tid = 0; r->o.msg.cmd.cs = cs;} static voidnotransaction(RTMP *r){ r->o.msg.cmd.tid = 0;} static voidbextend(Buffer *b, int bsz){ u8int *ob; if(b->bsz >= bsz) return; ob = b->b; b->b = erealloc(ob, bsz*2); if(ob != nil) b->p = b->b + (intptr)(ob - b->p); b->bsz = bsz*2; b->e = b->b + b->bsz;} static intrtmprecv(RTMP *r){ int hsz, len, n, msid; u8int *h, *e, byte; u32int ts; memset(&r->i.msg, 0, sizeof(r->i.msg)); r->i.p = r->i.b; if((n = readn(r->fd, &byte, 1)) != 1){ if(n == 0) werrstr("eof"); goto err; } r->i.msg.fmt = (byte & 0xc0)>>6; r->i.msg.cs = byte & 0x3f; n = r->i.msg.cs + 1; if(n <= 2){ if(readn(r->fd, r->i.p, n) != n) goto err; r->i.msg.cs = 64 + r->i.p[0]; if(n == 2) r->i.msg.cs += 256 * r->i.p[1]; } hsz = szs[r->i.msg.fmt]; if(readn(r->fd, r->i.p, hsz) != hsz) goto err; h = r->i.p; e = r->i.p + hsz; r->i.msg.type = -1; msid = 0; ts = 0; len = 0; if(hsz >= szs[Type2]){ h = a₀i24get(h, e, (s32int*)&ts); /* FIXME proper timestamps? */ if(hsz >= szs[Type1]){ h = a₀i24get(h, e, &len); h = a₀byteget(h, e, &byte); r->i.msg.type = byte; if(hsz >= szs[Type0]) h = a₀i32leget(h, e, &msid); } } if(ts == 0xffffff){ /* exntended timestamp */ if(readn(r->fd, h, 4) != 4) goto err; h = a₀i32get(h, h+4, (s32int*)&ts); } /* FIXME do all consecutive chunks use Type3? */ bextend(&r->i, len); r->i.msg.data = h; r->i.msg.sz = len; for(;;){ n = min(len, r->chunkin); if(readn(r->fd, h, n) != n) goto err; len -= n; h += n; if(len < 1) break; if(readn(r->fd, h, 1) != 1) goto err; if((r->i.msg.cs | Type3<<6) != *h){ werrstr("cs/fmt does not match: %02x", *h); goto err; } } return 0;err: werrstr("rtmprecv: %r"); return -1;} static intrtmpsend(RTMP *r, int flush){ u8int *p, *h, *e, hdata[24]; int len, n, hsz; Command *c; Message *m; if(r->o.p == nil) goto err; m = &r->o.msg; m->data = r->o.b; m->sz = r->o.p - r->o.b; h = hdata; *h++ = m->fmt<<6 | m->cs; hsz = szs[m->fmt]; e = h + hsz; if(hsz >= szs[Type2]){ h = a₀i24(h, e, m->ts); if(hsz >= szs[Type1]){ h = a₀i24(h, e, m->sz); h = a₀byte(h, e, m->type); if(hsz >= szs[Type0]) h = a₀i32le(h, e, m->sid); } } assert(h != nil); memset(h, 0, e-h); if(Bwrite(r, hdata, h-hdata) < 0) goto err; for(p = m->data, len = m->sz; len > 0;){ n = min(len, r->chunkout); if(Bwrite(r, p, n) < 0) goto err; p += n; len -= n; if(len > 0){ *h = m->cs | Type3<<6; Bputc(r, *h); } } if(flush && Bflush(r) < 0) goto err; if(debug){ fprint(2, "← %M", m); if(m->type == AMF0Command){ A₀ *a; u8int *s, *e; fprint(2, ":"); s = m->data; e = s + m->sz; for(; s != nil && s != e;){ if((s = a₀parse(&a, s, e)) != nil) fprint(2, " %A", a); else fprint(2, " %r"); a₀free(a); } } fprint(2, "\n"); } if(m->type == AMF0Command){ c = emalloc(sizeof(*c)); *c = m->cmd; assert(c->cb != nil); if((c->next = r->cmds.w) != nil) c->next->prev = c; r->cmds.w = c; } return 0;err: werrstr("rtmpsend: %r"); return -1;} static intpong(RTMP *r, s32int n){ newmsg(r, UserControl, Type0, CSUserCtl); puti16(CtlPingResponse); puti32(n); return rtmpsend(r, 1);} static intsetchunksz(RTMP *r, int sz){ int n; newmsg(r, SetChunkSize, Type0, CSUserCtl); puti32(sz); n = rtmpsend(r, 1); r->chunkout = sz; return n;} static voidloop(void *aux){ int res, n, ok, i; A₀ *a[NumCb], *v; u8int *s, *e; s16int s16; Message *m; Command *c; char *k; RTMP *r; r = aux; m = &r->i.msg; res = 0; memset(a, 0, sizeof(a)); for(;;){ for(n = 0; n < nelem(a); n++) a₀free(a[n]); memset(a, 0, sizeof(a)); if(res != 0 || (res = rtmprecv(r)) != 0){ if(debug) fprint(2, "rtmp loop: %r\n"); for(n = 0; n < nelem(a); n++) a₀free(a[n]); break; } s = m->data; e = s + m->sz; qlock(r); if(debug) fprint(2, "→ %M", m); switch(m->type){ case AMF0Command: c = nil; ok = 1; for(n = 0; n < NumCb; n++){ if((s = a₀parse(&a[n], s, e)) == nil) goto err; switch(n){ case CbCommand: if(a[n]->type != Tstr){ werrstr("command name is not a string: %A", a[n]); goto err; } if(strcmp(a[n]->str, "_error") == 0) ok = 0; /* other values: "_result", etc */ break; case CbTransID: if(a[n]->type != Tnum){ werrstr("transaction ID is not a number"); goto err; } if(a[n]->num == 0) /* no transaction, will try matching with response */ break; for(c = r->cmds.w; c != nil; c = c->next){ /* transaction id match */ if(c->tid == a[n]->num) break; } break; case CbResponse: if(a[CbTransID]->num != 0) /* should have matches with the transaction */ break; if(a[n]->type != Tobj) break;Lines 1–500 of 1156. Load more
#include <u.h> #include <libc.h> #include <thread.h> #include <bio.h> #include <libsec.h> #include "amf0.h" #include "ivf.h" #include "rtmp.h" #include "util.h" #define min(a,b) ((a)<(b)?(a):(b)) enum { Port = 1935, CSsz = 1536, ChunkDefault = 128, ChunkDesired = 65536, DataHdr = 0, DataFrame, Type0 = 0, Type1, Type2, Type3, CSUserCtl = 2, CSCtl = 3, CSData = 4, CbCommand = 0, CbTransID, CbObject, CbResponse, NumCb, /* UserControl */ CtlStreamBegin = 0, CtlStreamEOF, CtlStreamDry, CtlSetBufferLen, CtlStreamIsRecorded, CtlPingRequest = 6, CtlPingResponse, /* Message.type */ SetChunkSize = 1, Abort, Ack, UserControl, WindowAckSize, SetBandwidth, Audio = 8, Video, AMF3Metadata = 15, AMF3SharedObject, AMF3Command, AMF0Metadata, AMF0SharedObject, AMF0Command, Aggregate = 22, /* RTMP.bwlimit */ LimitHard = 0, LimitSoft, LimitDynamic, Biobufsz = 64*1024, /* FIXME don't know if it helps with anything */ Bufsz = 64*1024, }; typedef struct Buffer Buffer; typedef struct Command Command; typedef struct Message Message; #pragma varargck type "T" int #pragma varargck type "M" Message* struct Command { void (*cb)(RTMP *r, int ok, A₀ *a[NumCb], void *aux); char *code; void *aux; int tid; int cs; Command *prev, *next; }; struct Message { int type; int fmt; int cs; int sid; u32int ts; u8int *data; int sz; Command cmd; }; struct Buffer { Message msg; u8int *b, *p, *e; int bsz; }; struct RTMP { Biobufhdr; QLock; Buffer i; Buffer o; Channel *c; char *tcurl; char *app; char *inst; char *path; /* FIXME no idea what this is for */ int chunkin; int chunkout; int mode; int fd; int winacksz; int bw; u8int bwlimit; struct { int tid; Command *w; }cmds; int sps; int aacpd; u8int biobuf[Biobufsz]; }; #define putnull() do{ r->o.p = a₀null(r->o.p, r->o.e); }while(0) #define putbyte(b) do{ r->o.p = a₀byte(r->o.p, r->o.e, b); }while(0) #define putdata(d, sz) do { r->o.p = a₀data(r->o.p, r->o.e, d, sz); }while(0) #define puti16(i) do{ r->o.p = a₀i16(r->o.p, r->o.e, i); }while(0) #define puti24(i) do{ r->o.p = a₀i24(r->o.p, r->o.e, i); }while(0) #define puti32(i) do{ r->o.p = a₀i32(r->o.p, r->o.e, i); }while(0) #define putnum(v) do{ r->o.p = a₀num(r->o.p, r->o.e, v); }while(0) #define putstr(s) do{ r->o.p = a₀str(r->o.p, r->o.e, s); }while(0) #define putarr() do{ r->o.p = a₀arr(r->o.p, r->o.e); }while(0) #define putobj() do{ r->o.p = a₀obj(r->o.p, r->o.e); }while(0) #define putend() do{ r->o.p = a₀end(r->o.p, r->o.e); }while(0) #define putkvnum(name, v) do{ r->o.p = a₀kvnum(r->o.p, r->o.e, name, v); }while(0) #define putkvstr(name, s) do{ r->o.p = a₀kvstr(r->o.p, r->o.e, name, s); }while(0) #define putkvbool(name, s) do{ r->o.p = a₀kvbool(r->o.p, r->o.e, name, s); }while(0) #define putcommand(name, cb_) do { \ putstr(name); \ putnum(r->o.msg.cmd.tid); \ putobj(); \ r->o.msg.cmd.cb = cb_; \ }while(0) #define putcall(name) do { \ putstr(name); \ putnum(r->o.msg.cmd.tid); \ }while(0) static int szs[] = { [Type3] = 0, [Type2] = 3, [Type1] = 7, [Type0] = 11, }; static char *msgtype2s[] = { [SetChunkSize] = "SetChunkSize", [Abort] = "Abort", [Ack] = "Ack", [UserControl] = "UserControl", [WindowAckSize] = "WindowAckSize", [SetBandwidth] = "SetBandwidth", [Audio] = "Audio", [Video] = "Video", [AMF3Metadata] = "AMF3Metadata", [AMF3SharedObject] = "AMF3SharedObject", [AMF3Command] = "AMF3Command", [AMF0Metadata] = "AMF0Metadata", [AMF0SharedObject] = "AMF0SharedObject", [AMF0Command] = "AMF0Command", [Aggregate] = "Aggregate", }; static char *ctl2s[] = { [CtlStreamBegin] = "StreamBegin", [CtlStreamEOF] = "StreamEOF", [CtlStreamDry] = "StreamDry", [CtlSetBufferLen] = "SetBufferLen", [CtlStreamIsRecorded] = "StreamIsRecorded", [CtlPingRequest] = "PingRequest", [CtlPingResponse] = "PingResponse", }; static char *bwlimit2s[] = { [LimitHard] = "hard", [LimitSoft] = "soft", [LimitDynamic] = "dynamic", }; static char *pubtype2s[] = { [PubLive] = "live", [PubAppend] = "append", [PubRecord] = "record", }; static void newmsg(RTMP *r, int type, int fmt, int cs) { memset(&r->o.msg, 0, sizeof(r->o.msg)); r->o.msg.type = type; r->o.msg.fmt = fmt; r->o.msg.cs = cs; r->o.p = r->o.b; if(type == AMF0Command) r->o.msg.cmd.tid = ++r->cmds.tid; else r->o.msg.cmd.tid = 0; r->o.msg.cmd.cs = cs; } static void notransaction(RTMP *r) { r->o.msg.cmd.tid = 0; } static void bextend(Buffer *b, int bsz) { u8int *ob; if(b->bsz >= bsz) return; ob = b->b; b->b = erealloc(ob, bsz*2); if(ob != nil) b->p = b->b + (intptr)(ob - b->p); b->bsz = bsz*2; b->e = b->b + b->bsz; } static int rtmprecv(RTMP *r) { int hsz, len, n, msid; u8int *h, *e, byte; u32int ts; memset(&r->i.msg, 0, sizeof(r->i.msg)); r->i.p = r->i.b; if((n = readn(r->fd, &byte, 1)) != 1){ if(n == 0) werrstr("eof"); goto err; } r->i.msg.fmt = (byte & 0xc0)>>6; r->i.msg.cs = byte & 0x3f; n = r->i.msg.cs + 1; if(n <= 2){ if(readn(r->fd, r->i.p, n) != n) goto err; r->i.msg.cs = 64 + r->i.p[0]; if(n == 2) r->i.msg.cs += 256 * r->i.p[1]; } hsz = szs[r->i.msg.fmt]; if(readn(r->fd, r->i.p, hsz) != hsz) goto err; h = r->i.p; e = r->i.p + hsz; r->i.msg.type = -1; msid = 0; ts = 0; len = 0; if(hsz >= szs[Type2]){ h = a₀i24get(h, e, (s32int*)&ts); /* FIXME proper timestamps? */ if(hsz >= szs[Type1]){ h = a₀i24get(h, e, &len); h = a₀byteget(h, e, &byte); r->i.msg.type = byte; if(hsz >= szs[Type0]) h = a₀i32leget(h, e, &msid); } } if(ts == 0xffffff){ /* exntended timestamp */ if(readn(r->fd, h, 4) != 4) goto err; h = a₀i32get(h, h+4, (s32int*)&ts); } /* FIXME do all consecutive chunks use Type3? */ bextend(&r->i, len); r->i.msg.data = h; r->i.msg.sz = len; for(;;){ n = min(len, r->chunkin); if(readn(r->fd, h, n) != n) goto err; len -= n; h += n; if(len < 1) break; if(readn(r->fd, h, 1) != 1) goto err; if((r->i.msg.cs | Type3<<6) != *h){ werrstr("cs/fmt does not match: %02x", *h); goto err; } } return 0; err: werrstr("rtmprecv: %r"); return -1; } static int rtmpsend(RTMP *r, int flush) { u8int *p, *h, *e, hdata[24]; int len, n, hsz; Command *c; Message *m; if(r->o.p == nil) goto err; m = &r->o.msg; m->data = r->o.b; m->sz = r->o.p - r->o.b; h = hdata; *h++ = m->fmt<<6 | m->cs; hsz = szs[m->fmt]; e = h + hsz; if(hsz >= szs[Type2]){ h = a₀i24(h, e, m->ts); if(hsz >= szs[Type1]){ h = a₀i24(h, e, m->sz); h = a₀byte(h, e, m->type); if(hsz >= szs[Type0]) h = a₀i32le(h, e, m->sid); } } assert(h != nil); memset(h, 0, e-h); if(Bwrite(r, hdata, h-hdata) < 0) goto err; for(p = m->data, len = m->sz; len > 0;){ n = min(len, r->chunkout); if(Bwrite(r, p, n) < 0) goto err; p += n; len -= n; if(len > 0){ *h = m->cs | Type3<<6; Bputc(r, *h); } } if(flush && Bflush(r) < 0) goto err; if(debug){ fprint(2, "← %M", m); if(m->type == AMF0Command){ A₀ *a; u8int *s, *e; fprint(2, ":"); s = m->data; e = s + m->sz; for(; s != nil && s != e;){ if((s = a₀parse(&a, s, e)) != nil) fprint(2, " %A", a); else fprint(2, " %r"); a₀free(a); } } fprint(2, "\n"); } if(m->type == AMF0Command){ c = emalloc(sizeof(*c)); *c = m->cmd; assert(c->cb != nil); if((c->next = r->cmds.w) != nil) c->next->prev = c; r->cmds.w = c; } return 0; err: werrstr("rtmpsend: %r"); return -1; } static int pong(RTMP *r, s32int n) { newmsg(r, UserControl, Type0, CSUserCtl); puti16(CtlPingResponse); puti32(n); return rtmpsend(r, 1); } static int setchunksz(RTMP *r, int sz) { int n; newmsg(r, SetChunkSize, Type0, CSUserCtl); puti32(sz); n = rtmpsend(r, 1); r->chunkout = sz; return n; } static void loop(void *aux) { int res, n, ok, i; A₀ *a[NumCb], *v; u8int *s, *e; s16int s16; Message *m; Command *c; char *k; RTMP *r; r = aux; m = &r->i.msg; res = 0; memset(a, 0, sizeof(a)); for(;;){ for(n = 0; n < nelem(a); n++) a₀free(a[n]); memset(a, 0, sizeof(a)); if(res != 0 || (res = rtmprecv(r)) != 0){ if(debug) fprint(2, "rtmp loop: %r\n"); for(n = 0; n < nelem(a); n++) a₀free(a[n]); break; } s = m->data; e = s + m->sz; qlock(r); if(debug) fprint(2, "→ %M", m); switch(m->type){ case AMF0Command: c = nil; ok = 1; for(n = 0; n < NumCb; n++){ if((s = a₀parse(&a[n], s, e)) == nil) goto err; switch(n){ case CbCommand: if(a[n]->type != Tstr){ werrstr("command name is not a string: %A", a[n]); goto err; } if(strcmp(a[n]->str, "_error") == 0) ok = 0; /* other values: "_result", etc */ break; case CbTransID: if(a[n]->type != Tnum){ werrstr("transaction ID is not a number"); goto err; } if(a[n]->num == 0) /* no transaction, will try matching with response */ break; for(c = r->cmds.w; c != nil; c = c->next){ /* transaction id match */ if(c->tid == a[n]->num) break; } break; case CbResponse: if(a[CbTransID]->num != 0) /* should have matches with the transaction */ break; if(a[n]->type != Tobj) break;