rtmp.bend source
rtmp.bend on the hub · documented module
# RTMP client: a session that gives a stream's audio and video.## s <- Rtmp.open(E.Opts.new("rtmp://server/app/stream"))# (s, tag) <- Rtmp.next(s) as many times as wanted# Rtmp.close(s)## What goes on the wire:## C: C0 C1 S: S0 S1 S2 C: C2# C: Set Chunk Size, connect(app) S: _result# C: createStream S: _result, the stream id# C: play(stream), Set Buffer Length S: onStatus Play.Start# S: data (the metadata), audio and video messages## A tag is one of those messages, in the form FLV keeps them: written# one after the other after a header they make an FLV file, and# flv.bend takes the frames out of them. This file is the IO; the bytes# and their meaning are in rtmp_core.bend and amf.bend, which are pure.## rtmps:// is RTMP under TLS. A login goes in the URL's query, as the# servers that use one expect (rtmp://host/app?user=...&pass=...).import Baseimport ./bytes.bend as Bimport ./amf.bend as Aimport ./rtmp_core.bend as Kimport ./rtsp_core.bend as Cimport ./conn.bend as Nimport ./opts.bend as E# An open session: the connection and what was read past the last tag,# the chunk streams' state, how long to wait for the server (ms), the# bytes received and how many of them were acknowledged.type Sess is Type: Sess{sock: Socket, buf: List<&2, U32>, dec: K.Dec, wait: U32, rx: U32, acked: U32}# How the steps inside end: a session opened, or a tag read.type End is Type: Opened{s: Sess} Next{s: Sess, t: K.Tag}def Fin() -> Type: N.Res(End)# Reading tags# ------------# With nothing whole in the buffer: an acknowledgement when a megabyte# came since the last (a server may stop sending without them), and# more bytes.def Rtmp.more(due: Bool, buf: List<&2, U32>, s: Socket, wait: U32, +rx: U32, acked: U32, again: Socket -> List<&2, U32> -> U32 -> U32 -> IO(Fin())) -> IO(Fin()): match due: case True{}: N.Conn.send(End, s, K.Rtmp.ack(rx), s2 => N.Conn.recv.n(End, s2, wait, buf, s3 => b3 => n => again(s3, b3, U32.add(rx, n), rx))) case False{}: N.Conn.recv.n(End, s, wait, buf, s3 => b3 => n => again(s3, b3, U32.add(rx, n), acked))# Until a tag: a media message given, a ping answered, the rest passed# over.def Rtmp.loop(fuel: Nat, e: K.Ev, +st: K.Dec, buf: List<&2, U32>, s: Socket, +wait: U32, +rx: U32, +acked: U32) -> IO(Fin()): match fuel e: case 0n _: N.Conn.fail(End, s, E.Garbled{}, 0, "no media in a very long stream of messages") case 1n+q K.Media{typ, ts, payload, st2, rest}: IO.pure(Fin(), Done{Next{Sess{s, rest, st2, wait, rx, acked}, K.Tag{typ, ts, payload}}}) case 1n+q K.Ping{data, +st2, +rest}: N.Conn.send(End, s, K.Rtmp.pong(data), s2 => Rtmp.loop(q, K.Rtmp.event(st2, rest), st2, rest, s2, wait, rx, acked)) case 1n+q K.Over{_, _}: N.Conn.fail(End, s, E.Ended{}, 0, "the publisher stopped") case 1n+q K.Refused{msg, _, _}: N.Conn.fail(End, s, E.NoStream{}, 0, msg) case 1n+q K.Skip{+st2, +rest}: Rtmp.loop(q, K.Rtmp.event(st2, rest), st2, rest, s, wait, rx, acked) case 1n+q K.Result{_, +st2, +rest}: Rtmp.loop(q, K.Rtmp.event(st2, rest), st2, rest, s, wait, rx, acked) case 1n+q K.Error{_, +st2, +rest}: Rtmp.loop(q, K.Rtmp.event(st2, rest), st2, rest, s, wait, rx, acked) case 1n+q K.Idle{}: Rtmp.more((U32.sub(rx, acked) >= 1000000 : U32), buf, s, wait, rx, acked, s3 => +b3 => +rx3 => +ack3 => Rtmp.loop(q, K.Rtmp.event(st, b3), st, b3, s3, wait, rx3, ack3))# Setting up# ----------# The answer to a command: the next _result, with pings answered and# everything else passed over meanwhile.def Rtmp.await(fuel: Nat, e: K.Ev, +st: K.Dec, +buf: List<&2, U32>, s: Socket, +wait: U32, k: Socket -> K.Dec -> List<&2, U32> -> List<&2, A.Tok> -> IO(Fin())) -> IO(Fin()): match fuel e: case 0n _: N.Conn.fail(End, s, E.Garbled{}, 0, "no answer to a command") case 1n+p K.Result{ts, st2, rest}: k(s, st2, rest, ts) case 1n+p K.Error{msg, _, _}: N.Conn.fail(End, s, E.Refused{}, 0, msg) case 1n+p K.Refused{msg, _, _}: N.Conn.fail(End, s, E.Refused{}, 0, msg) case 1n+p K.Ping{data, +st2, +rest}: N.Conn.send(End, s, K.Rtmp.pong(data), s2 => Rtmp.await(p, K.Rtmp.event(st2, rest), st2, rest, s2, wait, k)) case 1n+p K.Skip{+st2, +rest}: Rtmp.await(p, K.Rtmp.event(st2, rest), st2, rest, s, wait, k) case 1n+p K.Over{+st2, +rest}: Rtmp.await(p, K.Rtmp.event(st2, rest), st2, rest, s, wait, k) case 1n+p K.Media{_, _, _, +st2, +rest}: Rtmp.await(p, K.Rtmp.event(st2, rest), st2, rest, s, wait, k) case 1n+p K.Idle{}: N.Conn.recv(End, s, wait, buf, s2 => +b2 => Rtmp.await(p, K.Rtmp.event(st, b2), st, b2, s2, wait, k))# The stream is created (its id is the result's fourth value): play,# and the session is open.def Rtmp.created(s: Socket, st: K.Dec, buf: List<&2, U32>, ts: List<&2, A.Tok>, stream: String, wait: U32) -> IO(Fin()): N.Conn.send(End, s, K.Rtmp.play(A.Amf.number(A.Amf.at(ts, 3n)), stream), s2 => IO.pure(Fin(), Done{Opened{Sess{s2, buf, st, wait, 0, 0}}}))# Connected to the application: create a stream.def Rtmp.connected(s: Socket, +st: K.Dec, +buf: List<&2, U32>, stream: String, +wait: U32) -> IO(Fin()): N.Conn.send(End, s, K.Rtmp.create(), s2 => Rtmp.await(1000000n, K.Rtmp.event(st, buf), st, buf, s2, wait, s3 => st3 => b3 => ts => Rtmp.created(s3, st3, b3, ts, stream, wait)))# The handshake's answer is here (or more of it is read): C2, our chunk# size, and connect.def Rtmp.shaking(fuel: Nat, h: K.Shake, buf: List<&2, U32>, s: Socket, +u: C.Url, +w: K.Where, +wait: U32) -> IO(Fin()): match fuel h: case 0n _: N.Conn.fail(End, s, E.Garbled{}, 0, "the handshake did not end") case 1n+p K.Early{}: N.Conn.recv(End, s, wait, buf, s2 => +b2 => Rtmp.shaking(p, K.Rtmp.shake(b2), b2, s2, u, w, wait)) case 1n+p K.Shake{c2, +rest}: N.Conn.send(End, s, B.Bytes.cat(c2, B.Bytes.cat(K.Rtmp.chunk_size(K.Rtmp.out()), K.Rtmp.connect(K.Where.app(w), C.Url.host.of(u), C.Url.port.of(u)))), s2 => Rtmp.await(1000000n, K.Rtmp.event(K.Dec.new(), rest), K.Dec.new(), rest, s2, wait, s3 => st3 => b3 => _ => Rtmp.connected(s3, st3, b3, K.Where.stream(w), wait)))def Rtmp.go(bad: Bool, +u: C.Url, tls: Bool, o: E.Opts) -> IO(Fin()): match bad: case True{}: IO.pure(Fin(), Fail{E.Err{E.BadUrl{}, 0, "not an RTMP URL (rtmp://host[:port]/app/stream)"}}) case False{}: E.Opts{_, _, _, +wait, _, ca} = o do IO<Fin()>: seed : U32 <- IO.try(U32, IO.random_u32()) N.Conn.open(End, C.Url.host.of(u), C.Url.port.of(u), tls, ca, s => N.Conn.send(End, s, K.Rtmp.hello(seed), s2 => Rtmp.shaking(4096n, K.Early{}, Nil{}, s2, u, K.Where.of(C.Url.path(u)), wait)))def Rtmp.start.url(+u: C.Url, tls: Bool, o: E.Opts) -> IO(Fin()): Rtmp.go(String.is_empty(C.Url.host.of(u)), u, tls, o)def Rtmp.start.tls(+tls: Bool, url: String, o: E.Opts) -> IO(Fin()): Rtmp.start.url(C.Url.of(url, Bool.pick(String, tls, "rtmps://", "rtmp://"), Bool.pick(U32, tls, 443, 1935)), tls, o)def Rtmp.start(+o: E.Opts) -> IO(Fin()): E.Opts{+url, _, _, _, _, _} = o Rtmp.start.tls(String.starts_with(String.to_lower(url), "rtmps://"), url, o)# The public face# ---------------def Opened() -> Type: Result<&1, &1, E.Err, Sess>def Read() -> Type: Result<&1, &1, E.Err, Sess & K.Tag>def Rtmp.open.end(r: Fin()) -> Opened(): match r: case Done{Opened{s}}: Done{s} case Done{Next{_, _}}: Fail{E.Err{E.Garbled{}, 0, "a tag before the session was open"}} case Fail{e}: Fail{e}def Rtmp.next.end(r: Fin()) -> Read(): match r: case Done{Next{s, t}}: Done{(s, t)} case Done{Opened{_}}: Fail{E.Err{E.Garbled{}, 0, "the session opened twice"}} case Fail{e}: Fail{e}# Open a session: connect, shake hands, connect to the application and# ask to play the stream. On a failure the connection is already closed.def Rtmp.open(o: E.Opts) -> IO(Opened()): do IO<Opened()>: r : Fin() <- Rtmp.start(o) IO.pure(Opened(), Rtmp.open.end(r))def Rtmp.reopen.end(r: Opened(), rest: IO(Opened())) -> IO(Opened()): match r: case Done{s}: IO.pure(Opened(), Done{s}) case Fail{+e}: Bool.pick(IO(Opened()), E.Err.lasting(e), IO.pure(Opened(), Fail{e}), rest)def Rtmp.reopen.go(late: Bool, o: E.Opts, rest: IO(Opened())) -> IO(Opened()): match late: case True{}: IO.pure(Opened(), Fail{E.Err{E.Silent{}, 0, "the stream did not come back in time"}}) case False{}: do IO<Opened()>: r : Opened() <- Rtmp.open(o) Rtmp.reopen.end(r, rest)def Rtmp.reopen.at(tries: Nat, +o: E.Opts, +until: Nat, +delay: U32) -> IO(Opened()): match tries: case 0n: IO.pure(Opened(), Fail{E.Err{E.Silent{}, 0, "the stream did not come back"}}) case 1n+p: rest = Rtmp.reopen.at(p, o, until, U32.min((delay * 2 : U32), 15000)) do IO<Opened()>: u : Unit <- IO.sleep(delay) now : Nat <- IO.now() Rtmp.reopen.go(Nat.is_le(until, now), o, rest)# Open again a session that was lost: it waits half a second, tries,# and doubles the wait (up to 15 s) after each failure, until the# session opens, an error says it never will (E.Err.lasting), or the# clock (IO.now's, ms) reaches until.def Rtmp.reopen(o: E.Opts, until: Nat) -> IO(Opened()): Rtmp.reopen.at(100000n, o, until, 500)# The next tag, and the session to go on with. It waits for the server# at most the options' wait; on a failure (or at the stream's end, which# comes as the reason Ended) the connection is already closed.def Rtmp.next(x: Sess) -> IO(Read()): Sess{s, +buf, +dec, wait, rx, acked} = x do IO<Read()>: r : Fin() <- Rtmp.loop(4000000000n, K.Rtmp.event(dec, buf), dec, buf, s, wait, rx, acked) IO.pure(Read(), Rtmp.next.end(r))# End the session: the connection closed.def Rtmp.close(x: Sess) -> IO(Unit): Sess{s, _, _, _, _, _} = x N.Conn.shut(s, Nil{}, 0)