History

rewrite according to the rtmp spec

Sigrid Solveig Haflínudóttir committed 5 years ago

Commit 5a2a9721 parent c6ee1036 files · +363 −764Browse files

Unified · Split

6 files
  1. LICENSE +1 −1
  2. amf.c +0 −366
  3. amf.h +0 −52
  4. flv.c +22 −22
  5. mkfile +3 −2
  6. rtmp.c +337 −321
LICENSE+1−1
@@ -1 +1 @@
1−This is partially based on librtmp which is LGPLv2.1, same goes for this project.
1+Public domain.
amf.c+0−366
@@ -1,366 +0,0 @@
1−#include <u.h>
2−#include <libc.h>
3−#include "amf.h"
4−#include "util.h"
5−
6−enum {
7− Anum,
8− Abool,
9− Astr,
10− Aobj,
11− Aarr = 8,
12− Aend,
13− Alstr = 12,
14−};
15−
16−#define atleast(what, x) do{ \
17− if(p == nil) \
18− return nil; \
19− if(e-p < x){ \
20− werrstr(what ": buffer short: expected %d, have %d", x, (int)(e-p)); \
21− return nil; \
22− } \
23−}while(0)
24−
25−u8int *
26−amfbool(u8int *p, u8int *e, int v)
27−{
28− atleast("bool", 2);
29− *p++ = Abool;
30− *p++ = !!v;
31− return p;
32−}
33−
34−u8int *
35−amfbyte(u8int *p, u8int *e, u8int byte)
36−{
37− atleast("byte", 1);
38− *p++ = byte;
39− return p;
40−}
41−
42−u8int *
43−amfi16(u8int *p, u8int *e, s16int i)
44−{
45− atleast("i16", 2);
46− *p++ = i >> 8;
47− *p++ = i;
48− return p;
49−}
50−
51−u8int *
52−amfi24(u8int *p, u8int *e, s32int i)
53−{
54− atleast("i24", 3);
55− *p++ = i >> 16;
56− *p++ = i >> 8;
57− *p++ = i;
58− return p;
59−}
60−
61−u8int *
62−amfi32(u8int *p, u8int *e, s32int i)
63−{
64− atleast("i32", 4);
65− *p++ = i >> 24;
66− *p++ = i >> 16;
67− *p++ = i >> 8;
68− *p++ = i;
69− return p;
70−}
71−
72−u8int *
73−amfnum(u8int *p, u8int *e, double v)
74−{
75− union {
76− double v;
77− u64int u;
78− }x;
79−
80− atleast("num", 1+8);
81− *p++ = Anum;
82− x.v = v;
83− *p++ = x.u >> 56;
84− *p++ = x.u >> 48;
85− *p++ = x.u >> 40;
86− *p++ = x.u >> 32;
87− *p++ = x.u >> 24;
88− *p++ = x.u >> 16;
89− *p++ = x.u >> 8;
90− *p++ = x.u;
91−
92− return p;
93−}
94−
95−u8int *
96−amfstr(u8int *p, u8int *e, char *s)
97−{
98− int n;
99−
100− n = strlen(s);
101− if(n <= 0xffff){
102− atleast("str", 1+2+n);
103− *p++ = Astr;
104− return (u8int*)memmove(amfi16(p, e, n), s, n) + n;
105− }
106−
107− atleast("lstr", 1+4+n);
108− *p++ = Alstr;
109− return (u8int*)memmove(amfi32(p, e, n), s, n) + n;
110−}
111−
112−u8int *
113−amfarr(u8int *p, u8int *e)
114−{
115− return amfbyte(p, e, Aarr);
116−}
117−
118−u8int *
119−amfobj(u8int *p, u8int *e)
120−{
121− return amfbyte(p, e, Aobj);
122−}
123−
124−u8int *
125−amfend(u8int *p, u8int *e)
126−{
127− return amfi24(p, e, Aend);
128−}
129−
130−static u8int *
131−amfkv(u8int *p, u8int *e, char *name)
132−{
133− int n;
134−
135− if((n = strlen(name)) > 0xffff){
136− werrstr("string too long");
137− return nil;
138− }
139− atleast("kv", 2+n);
140− p = amfi16(p, e, n);
141−
142− return (u8int*)memmove(p, name, n) + n;
143−}
144−
145−u8int *
146−amfkvbool(u8int *p, u8int *e, char *name, int v)
147−{
148− return amfbool(amfkv(p, e, name), e, v);
149−}
150−
151−u8int *
152−amfkvnum(u8int *p, u8int *e, char *name, double v)
153−{
154− return amfnum(amfkv(p, e, name), e, v);
155−}
156−
157−u8int *
158−amfkvstr(u8int *p, u8int *e, char *name, char *v)
159−{
160− return amfstr(amfkv(p, e, name), e, v);
161−}
162−
163−u8int *
164−amfbyteget(u8int *p, u8int *e, u8int *byte)
165−{
166− atleast("byte", 1);
167− *byte = *p;
168− return p+1;
169−}
170−
171−u8int *
172−amfi16get(u8int *p, u8int *e, s16int *i)
173−{
174− atleast("i16", 2);
175− *i = p[0]<<8 | p[1];
176− return p+2;
177−}
178−
179−u8int *
180−amfi24get(u8int *p, u8int *e, s32int *i)
181−{
182− atleast("i24", 3);
183− *i = p[0]<<16 | p[1]<<8 | p[2];
184− return p+3;
185−}
186−
187−u8int *
188−amfi32get(u8int *p, u8int *e, s32int *i)
189−{
190− atleast("i32", 4);
191− *i = p[0]<<16 | p[1]<<16 | p[2]<<8 | p[3];
192− return p+4;
193−}
194−
195−void
196−amffree(Amf *a)
197−{
198− int i;
199−
200− if(a == nil)
201− return;
202−
203− switch(a->type){
204− case Tstr:
205− free(a->str);
206− break;
207− case Tobj:
208− for(i = 0; i < a->obj.n; i++){
209− free(a->obj.k[i]);
210− amffree(a->obj.v[i]);
211− }
212− break;
213− case Tarr:
214− for(i = 0; i < a->arr.n; i++)
215− amffree(a->arr.v[i]);
216− case Tnum:
217− case Tbool:
218− break;
219− default:
220− sysfatal("unknown amf type %d", a->type);
221− }
222−
223− free(a);
224−}
225−
226−u8int *
227−amfparse(Amf **amf, u8int *p, u8int *e)
228−{
229− s16int s16;
230− union {
231− double v;
232− u64int u;
233− }x;
234− int n;
235− Amf *a;
236−
237− atleast("type", 1);
238−
239− a = ecalloc(1, sizeof(Amf));
240− *amf = nil;
241−
242− switch(*p++){
243− case Anum:
244− atleast("num", 8);
245− for(n = 0, x.u = 0; n < 8; n++)
246− x.u = x.u<<8 | *p++;
247− a->type = Tnum;
248− a->num = x.v;
249− break;
250− case Abool:
251− atleast("bool", 1);
252− a->type = Tbool;
253− a->bool = *p++;
254− break;
255− case Alstr:
256− if((p = amfi32get(p, e, &n)) == nil)
257− return nil;
258− if(0){
259− case Astr:
260− if((p = amfi16get(p, e, &s16)) == nil)
261− return nil;
262− n = s16;
263− }
264− atleast("str", n);
265− a->str = estrndup(p, n);
266− p += n;
267− break;
268− case Aobj:
269− atleast("obj", 3); /* Aend should be there anyway */
270− a->type = Tobj;
271− for(a->obj.n = 0;;){
272− if((p = amfi16get(p, e, &s16)) == nil)
273− break;
274− if(s16 == 0){
275− atleast("obj end?", 1);
276− if(*p != Aend){
277− werrstr("object doesn't end well");
278− p = nil;
279− break;
280− }
281− p++;
282− break;
283− }
284− n = s16;
285− atleast("obj key", n);
286− a->obj.n++;
287− a->obj.k = erealloc(a->obj.k, sizeof(*a->obj.k)*a->obj.n);
288− a->obj.v = erealloc(a->obj.v, sizeof(*a->obj.v)*a->obj.n);
289− a->obj.k[a->obj.n-1] = estrndup(p, n);
290− p += n;
291− if((p = amfparse(&a->obj.v[a->obj.n-1], p, e)) == nil){
292− werrstr("obj val: %r");
293− break;
294− }
295− }
296− break;
297− case Aarr:
298− a->type = Tarr;
299− if((p = amfi32get(p, e, &a->arr.n)) == nil)
300− break;
301− a->arr.v = emalloc(sizeof(*a->arr.v)*a->arr.n);
302− for(n = 0; n < a->arr.n; n++){
303− if((p = amfparse(&a->arr.v[n], p, e)) == nil){
304− werrstr("arr el: %r");
305− break;
306− }
307− }
308− if((p = amfi24get(p, e, &n)) == nil || n != Aend){
309− p = nil;
310− werrstr("array doesn't end with Aend");
311− }
312− break;
313− case Aend:
314− p = nil;
315− werrstr("unexpected Aend");
316− break;
317− default:
318− werrstr("unexpected type %d", p[-1]);
319− p = nil;
320− break;
321− }
322−
323− if(p == nil)
324− amffree(a);
325− else
326− *amf = a;
327−
328− return p;
329−}
330−
331−int
332−amffmt(Fmt *f)
333−{
334− Amf *a;
335− int i;
336−
337− a = va_arg(f->args, Amf*);
338−
339− switch(a->type){
340− case Tstr:
341− fmtprint(f, "%#q", a->str);
342− break;
343− case Tobj:
344− fmtprint(f, "{");
345− for(i = 0; i < a->obj.n; i++)
346− fmtprint(f, "%s%q:%A", i > 0 ? ", " : "", a->obj.k[i], a->obj.v[i]);
347− fmtprint(f, "}");
348− break;
349− case Tarr:
350− fmtprint(f, "[");
351− for(i = 0; i < a->arr.n; i++)
352− fmtprint(f, "%s%A", i > 0 ? ", " : "", a->arr.v[i]);
353− fmtprint(f, "]");
354− break;
355− case Tnum:
356− fmtprint(f, "%g", a->num);
357− break;
358− case Tbool:
359− fmtprint(f, a->bool ? "true" : "false");
360− break;
361− default:
362− sysfatal("unknown amf type %d", a->type);
363− }
364−
365− return 0;
366−}
amf.h+0−52
@@ -1,52 +0,0 @@
1−enum {
2− Tstr,
3− Tnum,
4− Tbool,
5− Tarr,
6− Tobj,
7−};
8−
9−typedef struct Amf Amf;
10−
11−struct Amf {
12− int type;
13− union {
14− char *str;
15− double num;
16− u8int bool;
17− struct {
18− Amf **v;
19− int n;
20− }arr;
21− struct {
22− char **k;
23− Amf **v;
24− int n;
25− }obj;
26− };
27−};
28−
29−u8int *amfi16(u8int *p, u8int *e, s16int i);
30−u8int *amfbool(u8int *p, u8int *e, int v);
31−u8int *amfbyte(u8int *p, u8int *e, u8int byte);
32−u8int *amfi24(u8int *p, u8int *e, s32int i);
33−u8int *amfi32(u8int *p, u8int *e, s32int i);
34−u8int *amfnum(u8int *p, u8int *e, double v);
35−u8int *amfstr(u8int *p, u8int *e, char *s);
36−u8int *amfarr(u8int *p, u8int *e);
37−u8int *amfobj(u8int *p, u8int *e);
38−u8int *amfend(u8int *p, u8int *e);
39−u8int *amfkvnum(u8int *p, u8int *e, char *name, double v);
40−u8int *amfkvstr(u8int *p, u8int *e, char *name, char *v);
41−u8int *amfkvbool(u8int *p, u8int *e, char *name, int v);
42−
43−u8int *amfbyteget(u8int *p, u8int *e, u8int *byte);
44−u8int *amfi16get(u8int *p, u8int *e, s16int *i);
45−u8int *amfi24get(u8int *p, u8int *e, s32int *i);
46−u8int *amfi32get(u8int *p, u8int *e, s32int *i);
47−
48−u8int *amfparse(Amf **a, u8int *p, u8int *e);
49−void amffree(Amf *a);
50−
51−#pragma varargck type "A" Amf*
52−int amffmt(Fmt *f);
flv.c+22−22
@@ -1,6 +1,6 @@
11 #include <u.h>
22 #include <libc.h>
3−#include "amf.h"
3+#include "amf0.h"
44 #include "flv.h"
55
66 enum {
@@ -27,25 +27,25 @@ flvscript(u8int *p, u8int *e, int w, int h, int audio) · 20 unmodified lines
2727 p0 = p;
2828 *p++ = Fvideo;
2929 psz = p;
30− p = amfi24(p, e, 0); /* sz set later */
31− p = amfi24(p, e, ts);
30+ p = amf0i24(p, e, 0); /* sz set later */
31+ p = amf0i24(p, e, ts);
3232 *p++ = ts>>24;
33− p = amfi24(p, e, stream);
33+ p = amf0i24(p, e, stream);
3434
3535 d = p;
36− p = amfstr(p, e, "onMetaData");
37− p = amfarr(p, e);
38− p = amfi32(p, e, audio ? 5 : 4);
39− p = amfkvnum(p, e, "duration", 0.0);
40− p = amfkvnum(p, e, "width", w);
41− p = amfkvnum(p, e, "height", h);
42− p = amfkvnum(p, e, "videocodecid", EncH264);
36+ p = amf0str(p, e, "onMetaData");
37+ p = amf0arr(p, e);
38+ p = amf0i32(p, e, audio ? 5 : 4);
39+ p = amf0kvnum(p, e, "duration", 0.0);
40+ p = amf0kvnum(p, e, "width", w);
41+ p = amf0kvnum(p, e, "height", h);
42+ p = amf0kvnum(p, e, "videocodecid", EncH264);
4343 if(audio)
44− p = amfkvnum(p, e, "audiocodecid", EncAAC);
45− p = amfend(p, e);
46− amfi24(psz, e, p-d);
44+ p = amf0kvnum(p, e, "audiocodecid", EncAAC);
45+ p = amf0end(p, e);
46+ amf0i24(psz, e, p-d);
4747
48− return amfi32(p, e, p-p0);
48+ return amf0i32(p, e, p-p0);
4949 }
5050
5151 u8int *
@@ -61,10 +61,10 @@ flvdata(u8int *p, u8int *e, u32int pts, u32int dts, void *data, int sz, int type · 9 unmodified lines
6161 p0 = p;
6262 *p++ = type;
6363 psz = p;
64− p = amfi24(p, e, 0); /* size to be set later */
65− p = amfi24(p, e, dts);
64+ p = amf0i24(p, e, 0); /* size to be set later */
65+ p = amf0i24(p, e, dts);
6666 *p++ = dts >> 24;
67− p = amfi24(p, e, stream);
67+ p = amf0i24(p, e, stream);
6868
6969 d = p;
7070 if(type == Faudio){
@@ -75,12 +75,12 @@ flvdata(u8int *p, u8int *e, u32int pts, u32int dts, void *data, int sz, int type · 4 unmodified lines
7575 *p++ = ((fl & FlKey) ? 0x10 : 0x20) | EncH264;
7676 *p++ = (fl & FlHdr) ? 0 : 1;
7777 pts = ((fl & FlHdr) || pts < dts) ? 0 : (pts - dts);
78− p = amfi24(p, e, pts);
78+ p = amf0i24(p, e, pts);
7979 if((fl & FlHdr) == 0)
80− p = amfi32(p, e, sz);
80+ p = amf0i32(p, e, sz);
8181 }
8282 p = (u8int*)memmove(p, data, sz) + sz;
83− amfi24(psz, e, p-d);
83+ amf0i24(psz, e, p-d);
8484
85− return amfi32(p, e, p-p0);
85+ return amf0i32(p, e, p-p0);
8686 }
mkfile+3−2
@@ -4,13 +4,14 @@ BIN=/$objtype/bin/video · 3 unmodified lines
44 TARG=rtmp
55
66 HFILES=\
7− amf.h\
7+ amf0.h\
88 flv.h\
99 ivf.h\
1010 rtmp.h\
11+ util.h\
1112
1213 OFILES=\
13− amf.$O\
14+ amf0.$O\
1415 flv.$O\
1516 ivf.$O\
1617 main.$O\
rtmp.c+337−321
@@ -3,7 +3,7 @@ · 2 unmodified lines
33 #include <thread.h>
44 #include <bio.h>
55 #include <libsec.h>
6−#include "amf.h"
6+#include "amf0.h"
77 #include "ivf.h"
88 #include "rtmp.h"
99 #include "util.h"
@@ -11,69 +11,80 @@ · 1 unmodified line
1111 #define min(a,b) ((a)<(b)?(a):(b))
1212
1313 enum {
14− SzLarge,
15− SzMedium,
16− SzSmall,
17− SzTiny,
18−
1914 Port = 1935,
20− Sigsz = 1536,
15+ CSsz = 1536,
2116 Chunk = 128,
2217
23− ChanCtl = 3,
18+ Type0 = 0,
19+ Type1,
20+ Type2,
21+ Type3,
22+
23+ CSCtl = 3,
2424
25− CbWhat = 0,
26− CbInvoke,
27− CbData,
28− CbStatus,
29− NumCbA,
25+ CbCommand = 0,
26+ CbTransID,
27+ CbObject,
28+ CbResponse,
29+ NumCb,
3030
31+ /* UserControl */
3132 CtlStreamBegin = 0,
32− CtlStreamEnd,
33+ CtlStreamEOF,
3334 CtlStreamDry,
34− CtlStreamRecorded,
35− CtlPing = 6,
36− CtlBufferEmpty = 31,
37− CtlBufferReady,
38−
39− PktChunkSz = 1,
40− PktBytesReadReport = 3,
41− PktControl = 4,
42− PktServerBW = 5,
43− PktClientBW,
44− PktAudio = 8,
45− PktVideo,
46− PktFlexStreamSend = 15,
47− PktFlexSharedObj,
48− PkgFlexMessage,
49− PktFlexInfo,
50− PktSharedObj = 19,
51− PktInvoke = 20,
52− PktFlashVideo = 22,
35+ CtlSetBufferLen,
36+ CtlStreamIsRecorded,
37+ CtlPingRequest = 6,
38+ CtlPingResponse,
39+
40+ /* Message.type */
41+ SetChunkSize = 1,
42+ Abort,
43+ Ack,
44+ UserControl,
45+ WindowAckSize,
46+ SetBandwidth,
47+ Audio = 8,
48+ Video,
49+ AMF3Metadata = 15,
50+ AMF3SharedObject,
51+ AMF3Command,
52+ AMF0Metadata,
53+ AMF0SharedObject,
54+ AMF0Command,
55+ Aggregate = 22,
56+
57+ /* RTMP.bwlimit */
58+ LimitHard = 0,
59+ LimitSoft,
60+ LimitDynamic,
5361
5462 Biobufsz = 64*1024, /* FIXME don't know if it helps with anything */
5563 Bufsz = 64*1024,
5664 };
5765
58−typedef struct Invoke Invoke;
59−typedef struct Packet Packet;
66+typedef struct Command Command;
67+typedef struct Message Message;
6068
61−struct Invoke {
62− void (*cb)(RTMP *r, int ok, Amf *a[NumCbA], void *aux);
69+#pragma varargck type "T" int
70+#pragma varargck type "M" Message*
71+
72+struct Command {
73+ void (*cb)(RTMP *r, int ok, Amf0 *a[NumCb], void *aux);
6374 void *aux;
64− int n;
75+ int tid;
6576
66− Invoke *prev, *next;
77+ Command *prev, *next;
6778 };
6879
69−struct Packet {
80+struct Message {
7081 int type;
71− int ht;
72− int chan;
82+ int fmt;
83+ int cs;
7384 u32int ts;
7485 u8int *data;
7586 int sz;
76− Invoke invoke;
87+ Command cmd;
7788 };
7889
7990 struct RTMP {
@@ -82,130 +93,89 @@ struct RTMP { · 2 unmodified lines
8293 char *app;
8394 char *path;
8495 char *tcurl;
85− Packet pk;
96+ Message msg;
8697 u8int *b, *p, *e;
8798 int chunk;
8899 int mode;
89100 int bsz;
90101 int i;
91− int svbw;
92− int clbw;
93− u8int clbw2;
102+ int winacksz;
103+ int bw;
104+ u8int bwlimit;
94105 struct {
95− int n;
96− Invoke *w;
97− }invokes;
106+ int tid;
107+ Command *w;
108+ }cmds;
98109 u8int biobuf[Biobufsz];
99110 };
100111
101−#define puti16(i) do{ r->p = amfi16(r->p, r->e, i); }while(0)
102−#define puti24(i) do{ r->p = amfi24(r->p, r->e, i); }while(0)
103−#define puti32(i) do{ r->p = amfi32(r->p, r->e, i); }while(0)
104−#define putnum(v) do{ r->p = amfnum(r->p, r->e, v); }while(0)
105−#define putstr(s) do{ r->p = amfstr(r->p, r->e, s); }while(0)
106−#define putarr() do{ r->p = amfarr(r->p, r->e); }while(0)
107−#define putobj() do{ r->p = amfobj(r->p, r->e); }while(0)
108−#define putend() do{ r->p = amfend(r->p, r->e); }while(0)
109−#define putkvnum(name, v) do{ r->p = amfkvnum(r->p, r->e, name, v); }while(0)
110−#define putkvstr(name, s) do{ r->p = amfkvstr(r->p, r->e, name, s); }while(0)
111−#define putkvbool(name, s) do{ r->p = amfkvbool(r->p, r->e, name, s); }while(0)
112−
113−#define putinvoke(name) do { \
112+#define puti16(i) do{ r->p = amf0i16(r->p, r->e, i); }while(0)
113+#define puti24(i) do{ r->p = amf0i24(r->p, r->e, i); }while(0)
114+#define puti32(i) do{ r->p = amf0i32(r->p, r->e, i); }while(0)
115+#define putnum(v) do{ r->p = amf0num(r->p, r->e, v); }while(0)
116+#define putstr(s) do{ r->p = amf0str(r->p, r->e, s); }while(0)
117+#define putarr() do{ r->p = amf0arr(r->p, r->e); }while(0)
118+#define putobj() do{ r->p = amf0obj(r->p, r->e); }while(0)
119+#define putend() do{ r->p = amf0end(r->p, r->e); }while(0)
120+#define putkvnum(name, v) do{ r->p = amf0kvnum(r->p, r->e, name, v); }while(0)
121+#define putkvstr(name, s) do{ r->p = amf0kvstr(r->p, r->e, name, s); }while(0)
122+#define putkvbool(name, s) do{ r->p = amf0kvbool(r->p, r->e, name, s); }while(0)
123+
124+#define putcommand(name, cb_) do { \
114125 putstr(name); \
115− putnum(r->pk.invoke.n); \
126+ putnum(r->msg.cmd.tid); \
116127 putobj(); \
128+ r->msg.cmd.cb = cb_; \
117129 }while(0)
118130
119131 static int szs[] = {
120− [SzTiny] = 1,
121− [SzSmall] = 4,
122− [SzMedium] = 8,
123− [SzLarge] = 12,
132+ [Type3] = 0,
133+ [Type2] = 3,
134+ [Type1] = 7,
135+ [Type0] = 11,
124136 };
125137
126−static char *pktype2s[] = {
127− [PktChunkSz] = "ChunkSz",
128− [PktBytesReadReport] = "BytesReadReport",
129− [PktControl] = "Control",
130− [PktServerBW] = "ServerBW",
131− [PktClientBW] = "ClientBW",
132− [PktAudio] = "Audio",
133− [PktVideo] = "Video",
134− [PktFlexStreamSend] = "FlexStreamSend",
135− [PktFlexSharedObj] = "FlexSharedObj",
136− [PktFlexInfo] = "FlexInfo",
137− [PktSharedObj] = "SharedObj",
138− [PktInvoke] = "Invoke",
139− [PktFlashVideo] = "FlashVideo",
138+static char *msgtype2s[] = {
139+ [SetChunkSize] = "SetChunkSize",
140+ [Abort] = "Abort",
141+ [Ack] = "Ack",
142+ [UserControl] = "UserControl",
143+ [WindowAckSize] = "WindowAckSize",
144+ [SetBandwidth] = "SetBandwidth",
145+ [Audio] = "Audio",
146+ [Video] = "Video",
147+ [AMF3Metadata] = "AMF3Metadata",
148+ [AMF3SharedObject] = "AMF3SharedObject",
149+ [AMF3Command] = "AMF3Command",
150+ [AMF0Metadata] = "AMF0Metadata",
151+ [AMF0SharedObject] = "AMF0SharedObject",
152+ [AMF0Command] = "AMF0Command",
153+ [Aggregate] = "Aggregate",
140154 };
141155
142156 static char *ctl2s[] = {
143157 [CtlStreamBegin] = "StreamBegin",
144− [CtlStreamEnd] = "StreamEnd",
158+ [CtlStreamEOF] = "StreamEOF",
145159 [CtlStreamDry] = "StreamDry",
146− [CtlStreamRecorded] = "StreamRecorded",
147− [CtlPing] = "Ping",
148− [CtlBufferEmpty] = "BufferEmpty",
149− [CtlBufferReady] = "BufferReady",
160+ [CtlSetBufferLen] = "SetBufferLen",
161+ [CtlStreamIsRecorded] = "StreamIsRecorded",
162+ [CtlPingRequest] = "PingRequest",
163+ [CtlPingResponse] = "PingResponse",
150164 };
151165
152166 extern int debug;
153167
154−#pragma varargck type "T" int
155−static int
156−pktypefmt(Fmt *f)
157−{
158− char *s;
159− int t;
160−
161− if((t = va_arg(f->args, int)) >= 0 &&
162− t < nelem(pktype2s) &&
163− (s = pktype2s[t]) != nil)
164− return fmtprint(f, "%s", s);
165−
166− return fmtprint(f, "%d", t);
167−}
168−
169−#pragma varargck type "P" Packet*
170−static int
171−pkfmt(Fmt *f)
172−{
173− u8int *s, *e;
174− Packet *p;
175− Amf *a;
176−
177− p = va_arg(f->args, Packet*);
178−
179− fmtprint(f, "type=%T chan=%d ts=%ud sz=%d", p->type, p->chan, p->ts, p->sz);
180−
181− s = p->data;
182− e = s + p->sz;
183−
184− if(p->type == PktInvoke){
185− fmtprint(f, ":");
186− for(; s != nil && s != e;){
187− if((s = amfparse(&a, s, e)) != nil)
188− fmtprint(f, " %A", a);
189− else
190− fmtprint(f, " %r");
191− amffree(a);
192− }
193− }
194−
195− return 0;
196−}
197−
198168 static void
199−newpacket(RTMP *r, int type, int ht, int chan)
169+newmsg(RTMP *r, int type, int fmt, int cs)
200170 {
201− memset(&r->pk, 0, sizeof(r->pk));
171+ memset(&r->msg, 0, sizeof(r->msg));
202172
203− r->pk.type = type;
204− r->pk.ht = ht;
205− r->pk.chan = chan;
173+ r->msg.type = type;
174+ r->msg.fmt = fmt;
175+ r->msg.cs = cs;
206176 r->p = r->b;
207− if(type == PktInvoke)
208− r->pk.invoke.n = ++r->invokes.n;
177+ if(type == AMF0Command)
178+ r->msg.cmd.tid = ++r->cmds.tid;
209179 }
210180
211181 static void
@@ -223,101 +193,79 @@ bextend(RTMP *r, int bsz) · 11 unmodified lines
223193 r->e = r->b + r->bsz;
224194 }
225195
226−static int
227−handshake(int f)
228−{
229− u8int cl[1+Sigsz], sv[1+Sigsz];
230−
231− cl[0] = 3; /* no encryption */
232− memset(cl+1, 0, 8);
233− prng(cl+1+8, Sigsz-8);
234− if(write(f, cl, sizeof(cl)) != sizeof(cl))
235− goto err;
236− if(readn(f, sv, sizeof(sv)) != sizeof(sv))
237− goto err;
238− if(cl[0] != sv[0]){
239− werrstr("expected %d (no encryption), got %d", cl[0], sv[0]);
240− goto err;
241− }
242− if(write(f, sv+1, Sigsz) != Sigsz)
243− goto err;
244− if(readn(f, sv+1, Sigsz) != Sigsz)
245− goto err;
246− if(memcmp(cl, sv, sizeof(cl)) != 0){
247− werrstr("signature mismatch");
248− goto err;
249− }
250−
251− return 0;
252−
253−err:
254− werrstr("handshake: %r");
255− return -1;
256−}
257−
258196 static int
259197 rtmprecv(RTMP *r)
260198 {
261− int hsz, bodysz, dummy, n;
262− u8int *h, *e, type;
199+ int hsz, len, n, msid;
200+ u8int *h, *e, byte;
263201 u32int ts;
264202
265− memset(&r->pk, 0, sizeof(r->pk));
203+ memset(&r->msg, 0, sizeof(r->msg));
266204
267205 r->p = r->b;
268206 if(readn(r->i, r->p, 1) != 1)
269207 goto err;
270− r->pk.ht = (r->p[0] & 0xc0)>>6;
271− r->pk.chan = r->p[0] & 0x3f;
272− hsz = szs[r->pk.ht];
273− if(readn(r->i, r->p+1, hsz-1) != hsz-1)
208+ r->msg.fmt = (r->p[0] & 0xc0)>>6;
209+ r->msg.cs = r->p[0] & 0x3f;
210+ n = r->msg.cs + 1;
211+ if(n <= 2){
212+ if(readn(r->i, r->p, n) != n)
213+ goto err;
214+ r->msg.cs = 64 + r->p[0];
215+ if(n == 2)
216+ r->msg.cs += 256 * r->p[1];
217+ }
218+
219+ hsz = szs[r->msg.fmt];
220+ if(readn(r->i, r->p, hsz) != hsz)
274221 goto err;
275222
276− h = r->p + 1;
223+ h = r->p;
277224 e = r->p + hsz;
278− r->pk.type = -1;
279− bodysz = 0;
280− if(hsz >= szs[SzSmall]){
281− h = amfi24get(h, e, (s32int*)&ts); /* FIXME proper timestamps? */
282− if(hsz >= szs[SzMedium]){
283− h = amfi24get(h, e, &bodysz);
284− h = amfbyteget(h, e, &type);
285− r->pk.type = type;
286− if(hsz >= szs[SzLarge]){
287− dummy = 0;
288− h = amfi32get(h, e, &dummy); /* FIXME seems to be always 0? */
289− }
225+
226+ r->msg.type = -1;
227+ msid = 0;
228+ ts = 0;
229+ len = 0;
230+ if(hsz >= szs[Type2]){
231+ h = amf0i24get(h, e, (s32int*)&ts); /* FIXME proper timestamps? */
232+ if(hsz >= szs[Type1]){
233+ h = amf0i24get(h, e, &len);
234+ h = amf0byteget(h, e, &byte);
235+ r->msg.type = byte;
236+ if(hsz >= szs[Type0])
237+ h = amf0i32leget(h, e, &msid);
290238 }
291239 }
292240
293241 if(ts == 0xffffff){ /* exntended timestamp */
294242 if(readn(r->i, h, 4) != 4)
295243 goto err;
296− h = amfi32get(h, h+4, (s32int*)&ts);
244+ h = amf0i32get(h, h+4, (s32int*)&ts);
297245 }
298246
299− /* FIXME do all consecutive chunks use Tiny? */
300− bextend(r, bodysz);
301− r->pk.data = h;
302− r->pk.sz = bodysz;
247+ /* FIXME do all consecutive chunks use Type3? */
248+ bextend(r, len);
249+ r->msg.data = h;
250+ r->msg.sz = len;
303251 for(;;){
304− n = min(bodysz, r->chunk);
252+ n = min(len, r->chunk);
305253 if(readn(r->i, h, n) != n)
306254 goto err;
307− bodysz -= n;
255+ len -= n;
308256 h += n;
309− if(bodysz < 1)
257+ if(len < 1)
310258 break;
311259 if(readn(r->i, h, 1) != 1)
312260 goto err;
313− if((r->pk.chan | SzTiny<<6) != *h){
314− werrstr("chan/size does not match: %02x", *h);
261+ if((r->msg.cs | Type3<<6) != *h){
262+ werrstr("cs/fmt does not match: %02x", *h);
315263 goto err;
316264 }
317265 }
318266
319267 if(debug)
320− fprint(2, "→ %P\n", &r->pk);
268+ fprint(2, "→ %M\n", &r->msg);
321269
322270 return 0;
323271
@@ -329,28 +277,27 @@ err: · 5 unmodified lines
329277 static int
330278 rtmpsend(RTMP *r)
331279 {
332− u8int *p, *h, *e, hdata[32];
333− int bodysz, n, hsz;
334− Invoke *i;
280+ u8int *p, *h, *e, hdata[24];
281+ int len, n, hsz;
282+ Command *c;
335283
336284 if(r->p == nil)
337285 goto err;
338286
339− r->pk.data = r->b;
340− r->pk.sz = r->p - r->b;
341− /* FIXME special case when bodysz is 0 */
287+ r->msg.data = r->b;
288+ r->msg.sz = r->p - r->b;
342289
343− hsz = szs[r->pk.ht];
344290 h = hdata;
291+ *h++ = r->msg.fmt<<6 | r->msg.cs;
292+ hsz = szs[r->msg.fmt];
345293 e = h + hsz;
346− *h++ = r->pk.ht<<6 | r->pk.chan;
347− if(hsz >= szs[SzSmall]){
348− h = amfi24(h, e, 0); /* FIXME proper timestamps? */
349− if(hsz >= szs[SzMedium]){
350− h = amfi24(h, e, r->pk.sz);
351− h = amfbyte(h, e, r->pk.type);
352− if(hsz >= szs[SzLarge])
353− h = amfi32(h, e, 0); /* FIXME seems to be always 0? */
294+ if(hsz >= szs[Type2]){
295+ h = amf0i24(h, e, 0); /* FIXME put actual timestamps? */
296+ if(hsz >= szs[Type1]){
297+ h = amf0i24(h, e, r->msg.sz);
298+ h = amf0byte(h, e, r->msg.type);
299+ if(hsz >= szs[Type0])
300+ h = amf0i32(h, e, 0); /* FIXME message stream id */
354301 }
355302 }
356303 assert(h != nil);
@@ -358,14 +305,14 @@ rtmpsend(RTMP *r) · 1 unmodified line
358305 if(Bwrite(r, hdata, h-hdata) < 0)
359306 goto err;
360307
361− for(p = r->pk.data, bodysz = r->pk.sz; bodysz > 0;){
362− n = min(bodysz, r->chunk);
308+ for(p = r->msg.data, len = r->msg.sz; len > 0;){
309+ n = min(len, r->chunk);
363310 if(Bwrite(r, p, n) < 0)
364311 goto err;
365312 p += n;
366− bodysz -= n;
367− if(bodysz > 0){
368− *h = r->pk.chan | SzTiny<<6;
313+ len -= n;
314+ if(len > 0){
315+ *h = r->msg.cs | Type3<<6;
369316 Bputc(r, *h);
370317 }
371318 }
@@ -373,15 +320,15 @@ rtmpsend(RTMP *r) · 1 unmodified line
373320 Bflush(r);
374321
375322 if(debug)
376− fprint(2, "← %P\n", &r->pk);
377−
378− if(r->pk.type == PktInvoke){
379− i = emalloc(sizeof(*i));
380− *i = r->pk.invoke;
381− assert(i->cb != nil);
382− if((i->next = r->invokes.w) != nil)
383− i->next->prev = i;
384− r->invokes.w = i;
323+ fprint(2, "← %M\n", &r->msg);
324+
325+ if(r->msg.type == AMF0Command){
326+ c = emalloc(sizeof(*c));
327+ *c = r->msg.cmd;
328+ assert(c->cb != nil);
329+ if((c->next = r->cmds.w) != nil)
330+ c->next->prev = c;
331+ r->cmds.w = c;
385332 }
386333
387334 return 0;
@@ -390,34 +337,6 @@ err: · 2 unmodified lines
390337 return -1;
391338 }
392339
393−static void
394−connected(RTMP *r, int ok, Amf *a[NumCbA], void *)
395−{
396− sendp(r->c, ok ? nil : smprint("%A", a[CbStatus]));
397−}
398−
399−static int
400−connect(RTMP *r)
401−{
402− newpacket(r, PktInvoke, SzLarge, ChanCtl);
403− putinvoke("connect");
404− putkvstr("app", r->app);
405− putkvstr("tcUrl", r->tcurl);
406− if(r->mode & OWRITE)
407− putkvstr("type", "nonprivate");
408− else{
409− putkvbool("fpad", 0);
410− putkvnum("capabilities", 15);
411− putkvnum("audioCodecs", 3191);
412− putkvnum("videoCodecs", 252);
413− putkvnum("videoFunction", 1);
414− }
415− putend();
416− r->pk.invoke.cb = connected;
417−
418− return rtmpsend(r);
419−}
420−
421340 static void
422341 rtmpfree(RTMP *r)
423342 {
@@ -437,80 +356,75 @@ static void · 13 unmodified lines
437356 loop(void *aux)
438357 {
439358 int res, n, ok;
440− Amf *a[NumCbA];
359+ Amf0 *a[NumCb];
441360 u8int *s, *e;
442361 s16int s16;
443− Packet *p;
444− Invoke *i;
362+ Message *m;
363+ Command *c;
445364 RTMP *r;
446365
447366 r = aux;
448− p = &r->pk;
367+ m = &r->msg;
449368 res = 0;
450369 memset(a, 0, sizeof(a));
451370 for(;;){
452371 for(n = 0; n < nelem(a); n++)
453− amffree(a[n]);
372+ amf0free(a[n]);
454373 memset(a, 0, sizeof(a));
455374
456375 if(res != 0 || (res = rtmprecv(r)) != 0){
457376 if(debug)
458377 fprint(2, "rtmp loop: %r\n");
459378 for(n = 0; n < nelem(a); n++)
460− amffree(a[n]);
379+ amf0free(a[n]);
461380 break;
462381 }
463382
464− s = r->pk.data;
465− e = s + r->pk.sz;
383+ s = r->msg.data;
384+ e = s + r->msg.sz;
466385
467− switch(p->type){
468− case PktInvoke:
469− i = nil;
470− ok = 0;
471− for(n = 0; n < NumCbA; n++){
472− if((s = amfparse(&a[n], s, e)) == nil)
386+ switch(m->type){
387+ case AMF0Command:
388+ c = nil;
389+ ok = 1;
390+ for(n = 0; n < NumCb; n++){
391+ if((s = amf0parse(&a[n], s, e)) == nil)
473392 goto err;
474393 switch(n){
475− case CbWhat:
394+ case CbCommand:
476395 if(a[n]->type != Tstr){
477− werrstr("invoke a[%d] not a string", n);
396+ werrstr("command name is not a string");
478397 goto err;
479398 }
480− if(strcmp(a[n]->str, "_result") == 0)
481− ok = 1;
482− else if(strcmp(a[n]->str, "_error") == 0)
399+ if(strcmp(a[n]->str, "_error") == 0)
483400 ok = 0;
484− else{
485− werrstr("unexpected a[%d]: %#q", n, a[n]->str);
486− goto err;
487− }
401+ /* other values: "_result", etc */
488402 break;
489− case CbInvoke:
403+ case CbTransID:
490404 if(a[n]->type != Tnum){
491− werrstr("invoke a[%d] not a number", n);
405+ werrstr("transaction ID is not a number");
492406 goto err;
493407 }
494− for(i = r->invokes.w; i != nil && i->n != a[n]->num; i = i->next);
495− if(i == nil){
496− werrstr("did not expect invoke %d result", (int)a[n]->num);
408+ for(c = r->cmds.w; c != nil && c->tid != a[n]->num; c = c->next);
409+ if(c == nil){
410+ werrstr("response to non-existent transaction %d", (int)a[n]->num);
497411 goto err;
498412 }
499413 break;
500414 }
501415 }
502− if(i->prev != nil)
503− i->prev->next = i->next;
504− if(i->next != nil)
505− i->next->prev = i->prev;
506− if(r->invokes.w == i)
507− r->invokes.w = i->next;
508− i->cb(r, ok, a, i->aux);
509− free(i);
416+ if(c->prev != nil)
417+ c->prev->next = c->next;
418+ if(c->next != nil)
419+ c->next->prev = c->prev;
420+ if(r->cmds.w == c)
421+ r->cmds.w = c->next;
422+ c->cb(r, ok, a, c->aux);
423+ free(c);
510424 break;
511425
512− case PktChunkSz:
513− if(amfi32get(s, e, &r->chunk) == nil)
426+ case SetChunkSize:
427+ if(amf0i32get(s, e, &r->chunk) == nil)
514428 goto err;
515429 if(r->chunk < 2){
516430 werrstr("invalid chunk size: %d", r->chunk);
@@ -520,51 +434,57 @@ loop(void *aux) · 3 unmodified lines
520434 fprint(2, "new chunk size: %d bytes\n", r->chunk);
521435 break;
522436
523− case PktBytesReadReport:
524− case PktControl:
525− if((s = amfi16get(s, e, &s16)) == nil)
437+ case UserControl:
438+ if((s = amf0i16get(s, e, &s16)) == nil)
526439 goto err;
527− if((s = amfi32get(s, e, &n)) == nil)
440+ if(amf0i32get(s, e, &n) == nil)
528441 n = -1;
529442 switch(s16){
530443 case CtlStreamBegin:
531− case CtlStreamEnd:
444+ case CtlStreamEOF:
532445 case CtlStreamDry:
533− case CtlStreamRecorded:
534− case CtlBufferEmpty:
535− case CtlBufferReady:
446+ case CtlSetBufferLen:
447+ case CtlStreamIsRecorded:
536448 if(0){
537− case CtlPing:
449+ case CtlPingRequest:
538450 /* FIXME pong */
539451 USED(n);
540452 }
541453 if(debug)
542− fprint(2, "control packet: %s %d\n", ctl2s[s16], n);
454+ fprint(2, "control message: %s %d\n", ctl2s[s16], n);
543455 break;
544456 default:
545457 if(debug)
546− fprint(2, "unknown control packet %d (value %d)\n", s16, n);
458+ fprint(2, "unknown control message %d (value %d)\n", s16, n);
547459 break;
548460 }
549461 break;
550462
551− case PktServerBW:
552− if((s = amfi32get(s, e, &r->svbw)) == nil)
463+ case WindowAckSize:
464+ if(amf0i32get(s, e, &r->winacksz) == nil)
553465 goto err;
554466 break;
555467
556− case PktClientBW:
557− if((s = amfi32get(s, e, &r->clbw)) == nil || (s = amfbyteget(s, e, &r->clbw2)) == nil)
468+ case SetBandwidth:
469+ if((s = amf0i32get(s, e, &r->bw)) == nil || amf0byteget(s, e, &r->bwlimit) == nil)
558470 goto err;
559471 break;
560472
561− case PktAudio:
562− case PktVideo:
563− case PktFlexStreamSend:
564− case PktFlexSharedObj:
565− case PktFlexInfo:
566− case PktSharedObj:
567− case PktFlashVideo:
473+ /* FIXME */
474+ case Aggregate:
475+ case Abort:
476+ case Ack:
477+ case Audio:
478+ case Video:
479+ case AMF0Metadata:
480+ case AMF0SharedObject:
481+ break;
482+
483+ case AMF3Metadata:
484+ case AMF3SharedObject:
485+ case AMF3Command:
486+ if(debug)
487+ fprint(2, "AMF3 message, ignoring\n");
568488 break;
569489 err:
570490 res = -1;
@@ -576,6 +496,102 @@ err: · 5 unmodified lines
576496 threadexitsall(res == 0 ? nil : "error");
577497 }
578498
499+static int
500+handshake(int f)
501+{
502+ u8int c[1+CSsz], s[1+CSsz];
503+
504+ c[0] = 3; /* rtmp v3 */
505+ memset(c+1, 0, 4+4); /* timestamp + zero */
506+ prng(c+1+8, CSsz-4-4);
507+ if(write(f, c, sizeof(c)) != sizeof(c))
508+ goto err;
509+ if(readn(f, s, sizeof(s)) != sizeof(s))
510+ goto err;
511+ if(c[0] != s[0]){
512+ werrstr("expected version %d, got %d", c[0], s[0]);
513+ goto err;
514+ }
515+ if(write(f, s+1, CSsz) != CSsz)
516+ goto err;
517+ if(readn(f, s+1, CSsz) != CSsz)
518+ goto err;
519+ if(memcmp(c, s, sizeof(c)) != 0){
520+ werrstr("C1 != S2");
521+ goto err;
522+ }
523+
524+ return 0;
525+
526+err:
527+ werrstr("handshake: %r");
528+ return -1;
529+}
530+
531+static void
532+connected(RTMP *r, int ok, Amf0 *a[NumCb], void *)
533+{
534+ sendp(r->c, ok ? nil : smprint("%A", a[CbResponse]));
535+}
536+
537+static int
538+connect(RTMP *r)
539+{
540+ newmsg(r, AMF0Command, Type0, CSCtl);
541+ putcommand("connect", connected);
542+ putkvstr("app", r->app);
543+ putkvstr("tcUrl", r->tcurl);
544+ putkvbool("fpad", 0); /* no proxy */
545+ putkvnum("audioCodecs", 0x4 | 0x400); /* mp3 + aac */
546+ putkvnum("videoCodecs", 0x80); /* h.264 */
547+ putkvnum("videoFunction", 0); /* no frame-accurate seek */
548+ putend();
549+
550+ return rtmpsend(r);
551+}
552+
553+static int
554+msgtypefmt(Fmt *f)
555+{
556+ char *s;
557+ int t;
558+
559+ if((t = va_arg(f->args, int)) >= 0 &&
560+ t < nelem(msgtype2s) &&
561+ (s = msgtype2s[t]) != nil)
562+ return fmtprint(f, "%s", s);
563+
564+ return fmtprint(f, "%d", t);
565+}
566+
567+static int
568+msgfmt(Fmt *f)
569+{
570+ u8int *s, *e;
571+ Message *m;
572+ Amf0 *a;
573+
574+ m = va_arg(f->args, Message*);
575+
576+ fmtprint(f, "type=%T cs=%d ts=%ud sz=%d", m->type, m->cs, m->ts, m->sz);
577+
578+ s = m->data;
579+ e = s + m->sz;
580+
581+ if(m->type == AMF0Command){
582+ fmtprint(f, ":");
583+ for(; s != nil && s != e;){
584+ if((s = amf0parse(&a, s, e)) != nil)
585+ fmtprint(f, " %A", a);
586+ else
587+ fmtprint(f, " %r");
588+ amf0free(a);
589+ }
590+ }
591+
592+ return 0;
593+}
594+
579595 RTMP *
580596 rtmpdial(char *url, int w, int h, int withaudio)
581597 {
@@ -583,9 +599,9 @@ rtmpdial(char *url, int w, int h, int withaudio) · 1 unmodified line
583599 int f, port, ctl;
584600 RTMP *r;
585601
586− fmtinstall('A', amffmt);
587− fmtinstall('T', pktypefmt);
588− fmtinstall('P', pkfmt);
602+ fmtinstall('A', amf0fmt);
603+ fmtinstall('T', msgtypefmt);
604+ fmtinstall('M', msgfmt);
589605 quotefmtinstall();
590606
591607 r = nil;

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t