rtmp.bend relies on unsafe/foreign

raw source on the hub · import 0x2355f41b6f5561f50104a75ccca27cf7/rtmp.bend as Rtmp

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=...).

7 imports

import Base
import ./bytes.bend as B
import ./amf.bend as A
import ./rtmp_core.bend as K
import ./rtsp_core.bend as C
import ./conn.bend as N
import ./opts.bend as E

Types

type Sess source · line 33 · raw

Type

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.

Sess

@sock:Socket -> @buf:List<&2, U32> -> @dec:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @wait:U32 -> @rx:U32 -> @acked:U32 -> Sess

type End source · line 37 · raw

Type

How the steps inside end: a session opened, or a tag read.

Opened
```
@s:Sess -> End
```

Next

@s:Sess -> @t:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Tag -> End

Definitions

def Fin source · line 41 · raw

Type

def Rtmp.more source · line 50 · raw

@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)

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.loop source · line 61 · raw

@fuel:Nat -> @e:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Ev -> @+st:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @buf:List<&2, U32> -> @s:Socket -> @+wait:U32 -> @+rx:U32 -> @+acked:U32 -> IO(Fin)

Until a tag: a media message given, a ping answered, the rest passed over.

def Rtmp.await source · line 91 · raw

@fuel:Nat -> @e:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Ev -> @+st:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @+buf:List<&2, U32> -> @s:Socket -> @+wait:U32 -> @k:(@_:Socket -> @_:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @_:List<&2, U32> -> @_:List<&2, 0x2355f41b6f5561f50104a75ccca27cf7/amf.Tok> -> IO(Fin)) -> IO(Fin)

The answer to a command: the next _result, with pings answered and everything else passed over meanwhile.

def Rtmp.created source · line 118 · raw

@s:Socket -> @st:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @buf:List<&2, U32> -> @ts:List<&2, 0x2355f41b6f5561f50104a75ccca27cf7/amf.Tok> -> @stream:String -> @wait:U32 -> IO(Fin)

The stream is created (its id is the result's fourth value): play, and the session is open.

def Rtmp.connected source · line 124 · raw

@s:Socket -> @+st:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Dec -> @+buf:List<&2, U32> -> @stream:String -> @+wait:U32 -> IO(Fin)

Connected to the application: create a stream.

def Rtmp.shaking source · line 132 · raw

@fuel:Nat -> @h:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Shake -> @buf:List<&2, U32> -> @s:Socket -> @+u:0x2355f41b6f5561f50104a75ccca27cf7/rtsp_core.Url -> @+w:0x2355f41b6f5561f50104a75ccca27cf7/rtmp_core.Where -> @+wait:U32 -> IO(Fin)

The handshake's answer is here (or more of it is read): C2, our chunk size, and connect.

def Rtmp.go source · line 146 · raw

@bad:Bool -> @+u:0x2355f41b6f5561f50104a75ccca27cf7/rtsp_core.Url -> @tls:Bool -> @o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> IO(Fin)

def Rtmp.start.url source · line 159 · raw

@+u:0x2355f41b6f5561f50104a75ccca27cf7/rtsp_core.Url -> @tls:Bool -> @o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> IO(Fin)

def Rtmp.start.tls source · line 162 · raw

@+tls:Bool -> @url:String -> @o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> IO(Fin)

def Rtmp.start source · line 166 · raw

@+o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> IO(Fin)

def Opened source · line 173 · raw

Type

def Read source · line 176 · raw

Type

def Rtmp.open.end source · line 179 · raw

@r:Fin -> Opened

def Rtmp.next.end source · line 188 · raw

@r:Fin -> Read

def Rtmp.open source · line 199 · raw

@o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> IO(Opened)

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.reopen.end source · line 204 · raw

@r:Opened -> @rest:IO(Opened) -> IO(Opened)

def Rtmp.reopen.go source · line 211 · raw

@late:Bool -> @o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> @rest:IO(Opened) -> IO(Opened)

def Rtmp.reopen.at source · line 220 · raw

@tries:Nat -> @+o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> @+until:Nat -> @+delay:U32 -> IO(Opened)

def Rtmp.reopen source · line 235 · raw

@o:0x2355f41b6f5561f50104a75ccca27cf7/opts.Opts -> @until:Nat -> IO(Opened)

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.next source · line 241 · raw

@x:Sess -> IO(Read)

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.close source · line 249 · raw

@x:Sess -> IO(Unit)

End the session: the connection closed.