av/rtmp/rtmp.go

858 lines
22 KiB
Go
Raw Normal View History

/*
NAME
rtmp.go
DESCRIPTION
RTMP command functionality.
AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org>
Dan Kortschak <dan@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
rtmp.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
Derived from librtmp under the GNU Lesser General Public License 2.1
Copyright (C) 2005-2008 Team XBMC http://www.xbmc.org
Copyright (C) 2008-2009 Andrej Stepanchuk
Copyright (C) 2009-2010 Howard Chu
*/
package rtmp
import (
"bytes"
"encoding/binary"
"errors"
2018-07-24 15:00:50 +03:00
"math/rand"
"net"
"strconv"
"time"
)
2018-08-08 10:36:16 +03:00
const (
pkg = "rtmp:"
signatureSize = 1536
fullHeaderSize = 12
)
// Link flags.
const (
linkAuth = 0x0001 // using auth param
linkLive = 0x0002 // stream is live
linkSWF = 0x0004 // do SWF verification - not implemented
linkPlaylist = 0x0008 // send playlist before play - not implemented
linkBufx = 0x0010 // toggle stream on BufferEmpty msg - not implemented
)
// Protocol features.
const (
featureHTTP = 0x01 // not implemented
featureEncode = 0x02 // not implemented
featureSSL = 0x04 // not implemented
featureMFP = 0x08 // not implemented
featureWrite = 0x10 // publish, not play
featureHTTP2 = 0x20 // server-side RTMPT - not implemented
)
// RTMP protocols.
const (
protoRTMP = 0
protoRTMPE = featureEncode
protoRTMPT = featureHTTP
protoRTMPS = featureSSL
protoRTMPTE = (featureHTTP | featureEncode)
protoRTMPTS = (featureHTTP | featureSSL)
protoRTMFP = featureMFP
)
// RTMP tokens (lexemes).
// NB: Underscores are deliberately preserved in const names where they exist in the corresponding tokens.
const (
av_checkbw = "_checkbw"
av_onbwcheck = "_onbwcheck"
av_onbwdone = "_onbwdone"
av_result = "_result"
avApp = "app"
avAudioCodecs = "audioCodecs"
avCapabilities = "capabilities"
avClose = "close"
avCode = "code"
avConnect = "connect"
avCreatestream = "createStream"
avDeletestream = "deleteStream"
avFCPublish = "FCPublish"
avFCUnpublish = "FCUnpublish"
avFlashver = "flashVer"
avFpad = "fpad"
avLevel = "level"
avLive = "live"
avNetConnectionConnectInvalidApp = "NetConnection.Connect.InvalidApp"
avNetStreamFailed = "NetStream.Failed"
avNetStreamPauseNotify = "NetStream.Pause.Notify"
avNetStreamPlayComplete = "NetStream.Play.Complete"
avNetStreamPlayFailed = "NetStream.Play.Failed"
avNetStreamPlayPublishNotify = "NetStream.Play.PublishNotify"
avNetStreamPlayStart = "NetStream.Play.Start"
avNetStreamPlayStop = "NetStream.Play.Stop"
avNetStreamPlayStreamNotFound = "NetStream.Play.StreamNotFound"
avNetStreamPlayUnpublishNotify = "NetStream.Play.UnpublishNotify"
avNetStreamPublish_Start = "NetStream.Publish.Start"
avNetStreamSeekNotify = "NetStream.Seek.Notify"
avNonprivate = "nonprivate"
avObjectEncoding = "objectEncoding"
avOnBWDone = "onBWDone"
avOnFCSubscribe = "onFCSubscribe"
avOnFCUnsubscribe = "onFCUnsubscribe"
avOnStatus = "onStatus"
avPageUrl = "pageUrl"
avPing = "ping"
avPlay = "play"
avPlaylist_ready = "playlist_ready"
avPublish = "publish"
avReleasestream = "releaseStream"
avSecureToken = "secureToken"
avSet_playlist = "set_playlist"
avSwfUrl = "swfUrl"
avTcUrl = "tcUrl"
avType = "type"
avVideoCodecs = "videoCodecs"
avVideoFunction = "videoFunction"
)
// RTMP protocol strings.
var rtmpProtocolStrings = [...]string{
"rtmp",
"rtmpt",
"rtmpe",
"rtmpte",
"rtmps",
"rtmpts",
"",
"",
"rtmfp",
}
// RTMP errors.
var (
2019-01-06 04:36:16 +03:00
errUnknownScheme = errors.New("rtmp: unknown scheme")
2019-01-07 16:29:41 +03:00
errConnected = errors.New("rtmp: already connected")
errNotConnected = errors.New("rtmp: not connected")
2019-01-10 16:18:11 +03:00
errNotWritable = errors.New("rtmp: connection not writable")
2019-01-06 04:36:16 +03:00
errHandshake = errors.New("rtmp: handshake failed")
errConnSend = errors.New("rtmp: connection send error")
errConnStream = errors.New("rtmp: connection stream error")
errInvalidHeader = errors.New("rtmp: invalid header")
errInvalidBody = errors.New("rtmp: invalid body")
errInvalidFlvTag = errors.New("rtmp: invalid FLV tag")
errEncoding = errors.New("rtmp: encoding error")
errDecoding = errors.New("rtmp: decoding error")
2019-01-10 15:22:43 +03:00
errUnimplemented = errors.New("rtmp: unimplemented feature")
)
// setupURL parses the RTMP URL.
func setupURL(s *Session) (err error) {
s.link.protocol, s.link.host, s.link.port, s.link.app, s.link.playpath, err = parseURL(s.url)
if err != nil {
2019-01-06 04:36:16 +03:00
return err
2018-07-15 21:09:36 +03:00
}
2018-07-19 18:35:21 +03:00
if s.link.tcUrl == "" {
if s.link.app != "" {
2019-01-07 16:29:41 +03:00
s.link.tcUrl = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
2018-07-19 18:35:21 +03:00
} else {
s.link.tcUrl = s.url
2018-07-19 18:35:21 +03:00
}
2018-07-15 21:09:36 +03:00
}
2018-07-22 17:01:19 +03:00
if s.link.port == 0 {
2018-07-19 18:35:21 +03:00
switch {
case (s.link.protocol & featureSSL) != 0:
s.link.port = 433
2019-01-07 16:29:41 +03:00
s.log(FatalLevel, pkg+"SSL not supported")
case (s.link.protocol & featureHTTP) != 0:
s.link.port = 80
2018-07-19 18:35:21 +03:00
default:
s.link.port = 1935
2018-07-19 18:35:21 +03:00
}
}
return nil
2018-07-15 21:09:36 +03:00
}
// connect establishes an RTMP connection.
func connect(s *Session) error {
addr, err := net.ResolveTCPAddr("tcp4", s.link.host+":"+strconv.Itoa(int(s.link.port)))
if err != nil {
return err
}
s.link.conn, err = net.DialTCP("tcp4", nil, addr)
if err != nil {
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
return err
}
2019-01-07 16:29:41 +03:00
s.log(DebugLevel, pkg+"connected")
err = handshake(s)
if err != nil {
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
return errHandshake
2018-07-22 16:31:50 +03:00
}
2019-01-07 16:29:41 +03:00
s.log(DebugLevel, pkg+"handshaked")
err = sendConnectPacket(s)
if err != nil {
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
return errConnSend
2018-07-22 16:31:50 +03:00
}
return nil
2018-07-22 16:31:50 +03:00
}
// connectStream reads a packet and handles it
func connectStream(s *Session) error {
2019-01-10 05:19:52 +03:00
var err error
2019-01-10 16:31:26 +03:00
for !s.isPlaying {
2019-01-10 05:19:52 +03:00
pkt := packet{}
err = pkt.read(s)
if err != nil {
break
}
2019-01-10 16:31:26 +03:00
switch pkt.packetType {
case packetTypeAudio, packetTypeVideo, packetTypeInfo:
2019-01-10 16:31:26 +03:00
s.log(WarnLevel, pkg+"got packet before play; ignoring", "type", pkt.packetType)
default:
err = handlePacket(s, &pkt)
if err != nil {
break
}
2019-01-10 05:19:52 +03:00
}
}
if !s.isPlaying {
2019-01-10 05:19:52 +03:00
return err
}
return nil
}
// handlePacket handles a packet that the client has received.
// NB: cases have been commented out that are not currently used by AusOcean
2019-01-10 05:19:52 +03:00
func handlePacket(s *Session, pkt *packet) error {
switch pkt.packetType {
case packetTypeChunkSize:
if pkt.bodySize >= 4 {
s.inChunkSize = int32(C_AMF_DecodeInt32(pkt.body[:4]))
}
case packetTypeBytesReadReport:
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
case packetTypeControl:
s.log(FatalLevel, pkg+"unsupported packet type packetTypeControl")
case packetTypeServerBW:
s.serverBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
case packetTypeClientBW:
s.clientBW = int32(C_AMF_DecodeInt32(pkt.body[:4]))
if pkt.bodySize > 4 {
s.clientBW2 = pkt.body[4]
} else {
s.clientBW2 = 0xff
}
case packetTypeAudio:
s.log(FatalLevel, pkg+"unsupported packet type packetTypeAudio")
case packetTypeVideo:
s.log(FatalLevel, pkg+"unsupported packet type packetTypeVideo")
case packetTypeFlexMessage:
s.log(FatalLevel, pkg+"unsupported packet type packetTypeFlexMessage")
case packetTypeInfo:
s.log(FatalLevel, pkg+"unsupported packet type packetTypeInfo")
case packetTypeInvoke:
err := handleInvoke(s, pkt.body[:pkt.bodySize])
if err != nil {
2018-09-06 05:20:58 +03:00
// This will never happen with the methods we implement.
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
2019-01-10 05:19:52 +03:00
return err
}
case packetTypeFlashVideo:
s.log(FatalLevel, pkg+"unsupported packet type packetType_FLASHVideo")
default:
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
}
2019-01-10 05:19:52 +03:00
return nil
}
func sendConnectPacket(s *Session) error {
var pbuf [4096]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avConnect)
if enc == nil {
return errEncoding
}
s.numInvokes += 1
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_OBJECT
enc = enc[1:]
enc = C_AMF_EncodeNamedString(enc, avApp, s.link.app)
if enc == nil {
return errEncoding
}
if s.link.protocol&featureWrite != 0 {
enc = C_AMF_EncodeNamedString(enc, avType, avNonprivate)
if enc == nil {
return errEncoding
}
}
if s.link.flashVer != "" {
enc = C_AMF_EncodeNamedString(enc, avFlashver, s.link.flashVer)
if enc == nil {
return errEncoding
}
}
if s.link.swfUrl != "" {
enc = C_AMF_EncodeNamedString(enc, avSwfUrl, s.link.swfUrl)
if enc == nil {
return errEncoding
}
}
if s.link.tcUrl != "" {
enc = C_AMF_EncodeNamedString(enc, avTcUrl, s.link.tcUrl)
if enc == nil {
return errEncoding
}
}
if s.link.protocol&featureWrite == 0 {
enc = C_AMF_EncodeNamedBoolean(enc, avFpad, false)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, avCapabilities, 15)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, avAudioCodecs, s.audioCodecs)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, avVideoCodecs, s.videoCodecs)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, avVideoFunction, 1)
if enc == nil {
return errEncoding
}
if s.link.pageUrl != "" {
enc = C_AMF_EncodeNamedString(enc, avPageUrl, s.link.pageUrl)
if enc == nil {
return errEncoding
}
}
}
if s.encoding != 0.0 || s.sendEncoding {
enc = C_AMF_EncodeNamedNumber(enc, avObjectEncoding, s.encoding)
if enc == nil {
return errEncoding
}
}
2019-01-10 05:19:52 +03:00
copy(enc, []byte{0, 0, AMF_OBJECT_END})
enc = enc[3:]
2019-01-10 05:19:52 +03:00
// add auth string
if s.link.auth != "" {
enc = C_AMF_EncodeBoolean(enc, s.link.flags&linkAuth != 0)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeString(enc, s.link.auth)
if enc == nil {
return errEncoding
}
}
for i := range s.link.extras.o_props {
enc = C_AMF_PropEncode(&s.link.extras.o_props[i], enc)
if enc == nil {
return errEncoding
}
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.write(s, true) // response expected
2018-07-25 04:26:59 +03:00
}
func sendCreateStream(s *Session) error {
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avCreatestream)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.write(s, true) // response expected
}
func sendReleaseStream(s *Session) error {
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avReleasestream)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, s.link.playpath)
if enc == nil {
return errEncoding
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.write(s, false)
}
func sendFCPublish(s *Session) error {
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avFCPublish)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, s.link.playpath)
if enc == nil {
return errEncoding
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.write(s, false)
}
func sendFCUnpublish(s *Session) error {
2018-09-06 04:56:52 +03:00
var pbuf [1024]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avFCUnpublish)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, s.link.playpath)
2018-09-06 04:56:52 +03:00
if enc == nil {
return errEncoding
2018-09-06 04:56:52 +03:00
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
2018-09-06 04:56:52 +03:00
return pkt.write(s, false)
}
func sendPublish(s *Session) error {
var pbuf [1024]byte
pkt := packet{
channel: chanSource,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avPublish)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, s.link.playpath)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeString(enc, avLive)
if enc == nil {
return errEncoding
2018-07-22 16:31:50 +03:00
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
2018-07-22 16:31:50 +03:00
return pkt.write(s, true) // response expected
2018-07-22 16:31:50 +03:00
}
func sendDeleteStream(s *Session, dStreamId float64) error {
2018-09-06 05:15:48 +03:00
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeMedium,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, avDeletestream)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeNumber(enc, dStreamId)
if enc == nil {
return errEncoding
}
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
2018-09-06 05:15:48 +03:00
return pkt.write(s, false)
2018-09-06 05:15:48 +03:00
}
// sendBytesReceived tells the server how many bytes the client has received.
func sendBytesReceived(s *Session) error {
var pbuf [256]byte
pkt := packet{
channel: chanBytesRead,
headerType: headerSizeMedium,
packetType: packetTypeBytesReadReport,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
2019-01-07 06:03:07 +03:00
}
enc := pkt.body
2019-01-07 06:03:07 +03:00
s.nBytesInSent = s.nBytesIn
enc = C_AMF_EncodeInt32(enc, s.nBytesIn)
if enc == nil {
return errEncoding
}
pkt.bodySize = 4
return pkt.write(s, false)
}
func sendCheckBW(s *Session) error {
var pbuf [256]byte
pkt := packet{
channel: chanControl,
headerType: headerSizeLarge,
packetType: packetTypeInvoke,
header: pbuf[:],
body: pbuf[fullHeaderSize:],
}
enc := pkt.body
enc = C_AMF_EncodeString(enc, av_checkbw)
if enc == nil {
return errEncoding
}
s.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(s.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))
return pkt.write(s, false)
}
func eraseMethod(m []method, i int) []method {
2018-09-12 04:59:38 +03:00
copy(m[i:], m[i+1:])
m[len(m)-1] = method{}
2018-09-12 04:59:38 +03:00
return m[:len(m)-1]
}
// int handleInvoke handles a packet invoke request
// Side effects: s.isPlaying set to true upon avNetStreamPublish_Start
func handleInvoke(s *Session, body []byte) error {
if body[0] != 0x02 {
return errInvalidBody
2018-08-14 11:52:37 +03:00
}
var obj C_AMFObject
nRes := C_AMF_Decode(&obj, body, 0)
2018-08-14 11:52:37 +03:00
if nRes < 0 {
return errDecoding
2018-08-14 11:52:37 +03:00
}
meth := C_AMFProp_GetString(C_AMF_GetProp(&obj, "", 0))
s.log(DebugLevel, pkg+"invoking method "+meth)
txn := C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 1))
2018-09-06 11:33:55 +03:00
switch meth {
case av_result:
2018-09-06 12:30:28 +03:00
var methodInvoked string
for i, m := range s.methodCalls {
2018-09-12 04:59:38 +03:00
if float64(m.num) == txn {
methodInvoked = m.name
s.methodCalls = eraseMethod(s.methodCalls, i)
2018-09-06 11:33:55 +03:00
break
2018-08-14 11:52:37 +03:00
}
2018-09-06 11:33:55 +03:00
}
2018-09-06 12:30:28 +03:00
if methodInvoked == "" {
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
2018-09-06 11:33:55 +03:00
goto leave
}
2019-01-09 10:03:19 +03:00
s.log(DebugLevel, pkg+"received result for "+methodInvoked)
2019-01-07 16:29:41 +03:00
2018-09-06 12:30:28 +03:00
switch methodInvoked {
case avConnect:
if s.link.token != "" {
s.log(FatalLevel, pkg+"no support for link token")
2018-09-06 11:33:55 +03:00
}
if (s.link.protocol & featureWrite) == 0 {
2019-01-10 16:18:11 +03:00
return errNotWritable
}
err := sendReleaseStream(s)
if err != nil {
return err
}
err = sendFCPublish(s)
if err != nil {
return err
}
err = sendCreateStream(s)
if err != nil {
return err
2018-09-06 11:33:55 +03:00
}
case avCreatestream:
s.streamID = int32(C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 3)))
if s.link.protocol&featureWrite == 0 {
2019-01-10 16:18:11 +03:00
return errNotWritable
}
err := sendPublish(s)
if err != nil {
return err
2018-08-14 11:52:37 +03:00
}
2018-09-06 11:33:55 +03:00
case avPlay, avPublish:
s.log(FatalLevel, pkg+"unsupported method avPlay/avPublish")
2018-08-13 05:45:43 +03:00
}
2018-09-06 11:33:55 +03:00
case avOnBWDone:
2019-01-09 15:21:07 +03:00
if s.checkCounter == 0 { // ToDo: why is this always zero?
2019-01-10 16:18:11 +03:00
err := sendCheckBW(s)
if err != nil {
return err
}
}
case avOnFCUnsubscribe, avOnFCSubscribe:
s.log(FatalLevel, pkg+"unsupported method avOnFCUnsubscribe/avOonfcsubscribe")
case avPing:
s.log(FatalLevel, pkg+"unsupported method avPing")
case av_onbwcheck:
s.log(FatalLevel, pkg+"unsupported method av_onbwcheck")
case av_onbwdone:
s.log(FatalLevel, pkg+"unsupported method av_onbwdone")
case avClose:
s.log(FatalLevel, pkg+"unsupported method avClose")
case avOnStatus:
var obj2 C_AMFObject
C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2)
code := C_AMFProp_GetString(C_AMF_GetProp(&obj2, avCode, -1))
level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, avLevel, -1))
2019-01-07 16:29:41 +03:00
s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)
2018-09-06 11:33:55 +03:00
switch code {
case avNetStreamFailed, avNetStreamPlayFailed,
avNetStreamPlayStreamNotFound, avNetConnectionConnectInvalidApp:
s.log(FatalLevel, pkg+"unsupported method avNetStream/avNetStreamPlayFailed/avNetstream_play_streamnotfound/av_netConnection_Connect_invalidApp")
case avNetStreamPlayStart, avNetStreamPlayPublishNotify:
s.log(FatalLevel, pkg+"unsupported method avNetStreamPlayStart/avNetStreamPlayPublishNotify")
case avNetStreamPublish_Start:
2019-01-10 05:19:52 +03:00
s.log(DebugLevel, pkg+"playing")
s.isPlaying = true
for i, m := range s.methodCalls {
if m.name == avPublish {
s.methodCalls = eraseMethod(s.methodCalls, i)
2018-08-14 11:52:37 +03:00
break
2018-08-13 05:45:43 +03:00
}
2018-08-14 11:52:37 +03:00
}
// ToDo: handle case when avPublish method not found
case avNetStreamPlayComplete, avNetStreamPlayStop, avNetStreamPlayUnpublishNotify:
s.log(FatalLevel, pkg+"unsupported method avNetStreamPlayComplete/avNetStreamPlayStop/avNetStreamPlayUnpublishNotify")
case avNetStreamSeekNotify:
s.log(FatalLevel, pkg+"unsupported method avNetstream_seek_notify")
case avNetStreamPauseNotify:
s.log(FatalLevel, pkg+"unsupported method avNetStreamPauseNotify")
}
case avPlaylist_ready:
s.log(FatalLevel, pkg+"unsupported method avPlaylist_ready")
default:
s.log(FatalLevel, pkg+"unknown method "+meth)
}
leave:
C_AMF_Reset(&obj)
return nil
}
func handshake(s *Session) error {
var clientbuf [signatureSize + 1]byte
2018-09-19 08:27:13 +03:00
clientsig := clientbuf[1:]
var serversig [signatureSize]byte
clientbuf[0] = chanControl
binary.BigEndian.PutUint32(clientsig, uint32(time.Now().UnixNano()/1000000))
2018-09-19 08:27:13 +03:00
copy(clientsig[4:8], []byte{0, 0, 0, 0})
2018-06-27 21:02:16 +03:00
for i := 8; i < signatureSize; i++ {
2018-09-19 08:27:13 +03:00
clientsig[i] = byte(rand.Intn(256))
}
_, err := s.write(clientbuf[:])
if err != nil {
return err
}
2019-01-10 05:19:52 +03:00
s.log(DebugLevel, pkg+"handshake sent")
2018-07-12 22:51:17 +03:00
2018-09-19 08:27:13 +03:00
var typ [1]byte
_, err = s.read(typ[:])
if err != nil {
return err
}
2019-01-10 05:19:52 +03:00
s.log(DebugLevel, pkg+"handshake received")
2018-07-14 15:40:01 +03:00
2018-09-19 05:20:46 +03:00
if typ[0] != clientbuf[0] {
2019-01-07 16:29:41 +03:00
s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
}
_, err = s.read(serversig[:])
if err != nil {
return err
2018-07-05 09:55:46 +03:00
}
// decode server response
2018-09-19 08:27:13 +03:00
suptime := binary.BigEndian.Uint32(serversig[:4])
2019-01-07 16:29:41 +03:00
s.log(DebugLevel, pkg+"server uptime", "uptime", suptime)
2018-08-11 07:18:21 +03:00
// 2nd part of handshake
_, err = s.write(serversig[:])
if err != nil {
return err
2018-08-11 07:18:21 +03:00
}
_, err = s.read(serversig[:])
if err != nil {
return err
}
if !bytes.Equal(serversig[:signatureSize], clientbuf[1:signatureSize+1]) {
s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:signatureSize], "clientsig", clientbuf[1:signatureSize+1])
}
return nil
2018-08-11 07:18:21 +03:00
}