mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
av/rtmp/rtmp.go

1685 lines
42 KiB
Go
Raw Blame History

/*
NAME
rtmp.go
DESCRIPTION
See Readme.md
AUTHORS
Saxon Nelson-Milton <saxon@ausocean.org>
Dan Kortschak <dan@ausocean.org>
LICENSE
rtmp.go is Copyright (C) 2017 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"
"fmt"
"io"
"log"
"math/rand"
"net"
"strconv"
"time"
)
const (
minDataSize = 11
debugMode = true
length = 512
)
const (
// av_setDataFrame is a static const global in rtmp.c
setDataFrame = "@setDataFrame"
av__checkbw = "_checkbw"
av__onbwcheck = "_onbwcheck"
av__onbwdone = "_onbwdone"
av__result = "_result"
av_app = "app"
av_audioCodecs = "audioCodecs"
av_capabilities = "capabilities"
av_close = "close"
av_code = "code"
av_connect = "connect"
av_createStream = "createStream"
av_deleteStream = "deleteStream"
av_FCPublish = "FCPublish"
av_FCUnpublish = "FCUnpublish"
av_flashVer = "flashVer"
av_fpad = "fpad"
av_level = "level"
av_live = "live"
av_NetConnection_Connect_InvalidApp = "NetConnection.Connect.InvalidApp"
av_NetStream_Failed = "NetStream.Failed"
av_NetStream_Pause_Notify = "NetStream.Pause.Notify"
av_NetStream_Play_Complete = "NetStream.Play.Complete"
av_NetStream_Play_Failed = "NetStream.Play.Failed"
av_NetStream_Play_PublishNotify = "NetStream.Play.PublishNotify"
av_NetStream_Play_Start = "NetStream.Play.Start"
av_NetStream_Play_Stop = "NetStream.Play.Stop"
av_NetStream_Play_StreamNotFound = "NetStream.Play.StreamNotFound"
av_NetStream_Play_UnpublishNotify = "NetStream.Play.UnpublishNotify"
av_NetStream_Publish_Start = "NetStream.Publish.Start"
av_NetStream_Seek_Notify = "NetStream.Seek.Notify"
av_nonprivate = "nonprivate"
av_objectEncoding = "objectEncoding"
av_onBWDone = "onBWDone"
av_onFCSubscribe = "onFCSubscribe"
av_onFCUnsubscribe = "onFCUnsubscribe"
av_onStatus = "onStatus"
av_pageUrl = "pageUrl"
av_ping = "ping"
av_play = "play"
av_playlist_ready = "playlist_ready"
av_publish = "publish"
av_releaseStream = "releaseStream"
av_secureToken = "secureToken"
av_set_playlist = "set_playlist"
av_swfUrl = "swfUrl"
av_tcUrl = "tcUrl"
av_type = "type"
av_videoCodecs = "videoCodecs"
av_videoFunction = "videoFunction"
)
var RTMPT_cmds = []string{
"open",
"send",
"idle",
"close",
}
var (
packetSize = [...]int{12, 8, 4, 1}
RTMPProtocolStringsLower = [...]string{
"rtmp",
"rtmpt",
"rtmpe",
"rtmpte",
"rtmps",
"rtmpts",
"",
"",
"rtmfp",
}
)
var (
errUnknownScheme = errors.New("rtmp: unknown scheme")
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")
errTinyPacket = errors.New("rtmp: packet too small")
errEncoding = errors.New("rtmp: encoding error")
errDecoding = errors.New("rtmp: decoding error")
errCopying = errors.New("rtmp: copying error")
)
func startSession(rtmp *C_RTMP, u string, timeout uint) (*C_RTMP, error) {
rtmp = C_RTMP_Alloc()
C_RTMP_Init(rtmp)
rtmp.Link.timeout = timeout
err := C_RTMP_SetupURL(rtmp, u)
if err != nil {
C_RTMP_Close(rtmp)
return nil, err
}
C_RTMP_EnableWrite(rtmp)
rtmp.nBufferMS = 3600 * 1000
err = C_RTMP_Connect(rtmp, nil)
if err != nil {
C_RTMP_Close(rtmp)
return nil, err
}
err = C_RTMP_ConnectStream(rtmp, 0)
if err != nil {
C_RTMP_Close(rtmp)
return nil, err
}
return rtmp, nil
}
func endSession(rtmp *C_RTMP) uint32 {
if rtmp == nil {
return 3
}
C_RTMP_Close(rtmp)
return 0
}
// uint32_t RTMP_GetTime();
// rtmp.c +156
func C_RTMP_GetTime() int32 {
return int32(time.Now().UnixNano() / 1000000)
}
// int RTMPPacket_Alloc(RTMPPacket* p, uint32_t nSize);
// rtmp.c +189
func C_RTMPPacket_Alloc(p *C_RTMPPacket, nSize uint32) {
buf := make([]byte, RTMP_MAX_HEADER_SIZE+nSize)
p.header = buf
p.body = buf[RTMP_MAX_HEADER_SIZE:]
p.nBytesRead = 0
}
// void RTMPPacket_Free(RTMPPacket* p);
// rtmp.c +203
func C_RTMPPacket_Free(p *C_RTMPPacket) {
if p.body != nil {
p.body = nil
}
}
// RTMP* RTMP_IsConnected();
// rtmp.c +317
func C_RTMP_Alloc() *C_RTMP {
return &C_RTMP{}
}
// void RTMP_Init(RTMP *r);
// rtmp.c +329
func C_RTMP_Init(r *C_RTMP) {
*r = C_RTMP{}
r.inChunkSize = RTMP_DEFAULT_CHUNKSIZE
r.outChunkSize = RTMP_DEFAULT_CHUNKSIZE
r.nBufferMS = 30000
r.nClientBW = 2500000
r.nClientBW2 = 2
r.nServerBW = 2500000
r.fAudioCodecs = 3191.0
r.fVideoCodecs = 252.0
r.Link.timeout = 30
r.Link.swfAge = 30
}
// void RTMP_EnableWrite(RTMP *r);
// rtmp.c +351
func C_RTMP_EnableWrite(r *C_RTMP) {
r.Link.protocol |= RTMP_FEATURE_WRITE
}
// int RTMP_IsConnected(RTMP *r);
// rtmp.c +363
func C_RTMP_IsConnected(r *C_RTMP) bool {
return r.Link.conn != nil
}
// void RTMP_SetBufferMS(RTMP *r, int size);
// rtmp.c +381
// DELETED
// void SocksSetup(RTMP *r, C_AVal* sockshost);
// rtmp.c +410
// DELETED
// int RTMP_SetupURL(RTMP *r, char* url);
// rtmp.c +757
// NOTE: code dealing with rtmp over http has been disregarded
func C_RTMP_SetupURL(r *C_RTMP, addr string) (err error) {
r.Link.protocol, r.Link.host, r.Link.port, r.Link.app, r.Link.playpath0, err = C_RTMP_ParseURL(addr)
if err != nil {
return err
}
r.Link.playpath = r.Link.playpath0
if r.Link.tcUrl == "" {
if r.Link.app != "" {
r.Link.tcUrl = fmt.Sprintf("%v://%v:%v/%v",
RTMPProtocolStringsLower[r.Link.protocol], r.Link.host, r.Link.port, r.Link.app)
r.Link.lFlags |= RTMP_LF_FTCU
} else {
r.Link.tcUrl = addr
}
}
if r.Link.port == 0 {
switch {
case (r.Link.protocol & RTMP_FEATURE_SSL) != 0:
r.Link.port = 433
case (r.Link.protocol & RTMP_FEATURE_HTTP) != 0:
r.Link.port = 80
default:
r.Link.port = 1935
}
}
return nil
}
// int RTMP_Connect(RTMP *r, RTMPPacket* cp);
// rtmp.c +1032
func C_RTMP_Connect(r *C_RTMP, cp *C_RTMPPacket) error {
addr, err := net.ResolveTCPAddr("tcp4", r.Link.host+":"+strconv.Itoa(int(r.Link.port)))
if err != nil {
return err
}
r.Link.conn, err = net.DialTCP("tcp4", nil, addr)
if err != nil {
return err
}
if debugMode {
log.Println("... connected, handshaking...")
}
err = C_HandShake(r, 1)
if err != nil {
log.Println("C_RTMP_Connect1: handshake failed!")
return errHandshake
}
if debugMode {
log.Println("... handshaked...")
}
err = C_SendConnectPacket(r, cp)
if err != nil {
log.Println("RTMP connect failed!")
return errConnSend
}
return nil
}
// int RTMP_Connect1(RTMP* r, RTMPPacket* cp);
// rtmp.c +978
// DELETED - subsumed by RTMP_Connect
// int RTMP_ConnectStream(RTMP* r, int seekTime);
// rtmp.c +1099
// Side effects: r.bPlaying is set true upon successful connection
func C_RTMP_ConnectStream(r *C_RTMP, seekTime int32) error {
var packet C_RTMPPacket
if seekTime > 0 {
r.Link.seekTime = seekTime
}
r.mediaChannel = 0
for !r.bPlaying && C_RTMP_IsConnected(r) {
err := C_RTMP_ReadPacket(r, &packet)
if err != nil {
break
}
// TODO: port is ready
if C_RTMPPacket_IsReady(&packet) {
if packet.nBodySize == 0 {
continue
}
if packet.packetType == RTMP_PACKET_TYPE_AUDIO ||
packet.packetType == RTMP_PACKET_TYPE_VIDEO ||
packet.packetType == RTMP_PACKET_TYPE_INFO {
log.Println("C_RTMP_ConnectStream: got packet before play()! Ignoring.")
C_RTMPPacket_Free(&packet)
continue
}
C_RTMP_ClientPacket(r, &packet)
C_RTMPPacket_Free(&packet)
}
}
if !r.bPlaying {
return errConnStream
}
return nil
}
// int RTMP_ClientPacket()
// rtmp.c +1226
// NOTE cases have been commented out that are not currently used by AusOcean
func C_RTMP_ClientPacket(r *C_RTMP, packet *C_RTMPPacket) int32 {
var bHasMediaPacket int32
switch packet.packetType {
case RTMP_PACKET_TYPE_CHUNK_SIZE:
// TODO: port this
C_HandleChangeChunkSize(r, packet)
case RTMP_PACKET_TYPE_BYTES_READ_REPORT:
// TODO: usue new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received: bytes read report", __FUNCTION__);
case RTMP_PACKET_TYPE_CONTROL:
panic("Unsupported packet type RTMP_PACKET_TYPE_CONTROL")
/*
log.Println("RTMP_PACKET_TYPE_CONTROL")
// TODO: port this
C.HandleCtrl(r, packet)
*/
case RTMP_PACKET_TYPE_SERVER_BW:
// TODO: port this
C_HandlServerBW(r, packet)
case RTMP_PACKET_TYPE_CLIENT_BW:
// TODO: port this
C_HandleClientBW(r, packet)
case RTMP_PACKET_TYPE_AUDIO:
panic("Unsupported packet type RTMP_PACKET_TYPE_AUDIO")
case RTMP_PACKET_TYPE_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_VIDEO")
case RTMP_PACKET_TYPE_FLEX_MESSAGE:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLEX_MESSAGE")
case RTMP_PACKET_TYPE_INFO:
panic("Unsupported packet type RTMP_PACKET_TYPE_INFO")
case RTMP_PACKET_TYPE_INVOKE:
log.Println("RTMP_PACKET_TYPE_INVOKE:")
// TODO use new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received: invoke %u bytes", __FUNCTION__,packet.nBodySize);
err := C_HandleInvoke(r, packet.body[:packet.nBodySize])
if err != nil {
// This will never happen with the methods we implement.
log.Println("HasMediaPacket")
bHasMediaPacket = 2
}
case RTMP_PACKET_TYPE_FLASH_VIDEO:
panic("Unsupported packet type RTMP_PACKET_TYPE_FLASH_VIDEO")
default:
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s, unknown packet type received: 0x%02x", __FUNCTION__,packet.packetType);
}
return bHasMediaPacket
}
// int ReadN(RTMP* r, char* buffer, int n);
// rtmp.c +1390
func C_ReadN(r *C_RTMP, buf []byte) error {
err := r.Link.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(r.Link.timeout)))
if err != nil {
return err
}
n, err := io.ReadFull(r.Link.conn, buf)
if err != nil {
if debugMode {
log.Printf("C_ReadN error: %v\n", err)
}
C_RTMP_Close(r)
return err
}
r.nBytesIn += int32(n)
if r.nBytesIn > (r.nBytesInSent + r.nClientBW/10) {
err := C_SendBytesReceived(r)
if err != nil {
return err
}
}
return nil
}
// int WriteN(RTMP* r, const char* buffer, int n);
// rtmp.c +1502
func C_WriteN(r *C_RTMP, buf []byte) error {
//ToDo: consider using a different timeout for writes than for reads
err := r.Link.conn.SetWriteDeadline(time.Now().Add(time.Second * time.Duration(r.Link.timeout)))
if err != nil {
return err
}
_, err = r.Link.conn.Write(buf)
if err != nil {
if debugMode {
log.Printf("C_WriteN, RTMP send error: %v\n", err)
}
C_RTMP_Close(r)
return err
}
return nil
}
// int SendConnectPacket(RTMP* r, RTMPPacket* cp);
// rtmp.c +1579
func C_SendConnectPacket(r *C_RTMP, cp *C_RTMPPacket) error {
if cp != nil {
return C_RTMP_SendPacket(r, cp, 1)
}
var pbuf [4096]byte
packet := C_RTMPPacket{
nChannel: 0x03,
headerType: RTMP_PACKET_SIZE_LARGE,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_connect)
if enc == nil {
return errEncoding
}
r.numInvokes += 1
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_OBJECT
enc = enc[1:]
enc = C_AMF_EncodeNamedString(enc, av_app, r.Link.app)
if enc == nil {
return errEncoding
}
if r.Link.protocol&RTMP_FEATURE_WRITE != 0 {
enc = C_AMF_EncodeNamedString(enc, av_type, av_nonprivate)
if enc == nil {
return errEncoding
}
}
if r.Link.flashVer != "" {
enc = C_AMF_EncodeNamedString(enc, av_flashVer, r.Link.flashVer)
if enc == nil {
return errEncoding
}
}
if r.Link.swfUrl != "" {
enc = C_AMF_EncodeNamedString(enc, av_swfUrl, r.Link.swfUrl)
if enc == nil {
return errEncoding
}
}
if r.Link.tcUrl != "" {
enc = C_AMF_EncodeNamedString(enc, av_tcUrl, r.Link.tcUrl)
if enc == nil {
return errEncoding
}
}
if r.Link.protocol&RTMP_FEATURE_WRITE == 0 {
enc = C_AMF_EncodeNamedBoolean(enc, av_fpad, false)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, av_capabilities, 15)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, av_audioCodecs, r.fAudioCodecs)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, av_videoCodecs, r.fVideoCodecs)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeNamedNumber(enc, av_videoFunction, 1)
if enc == nil {
return errEncoding
}
if r.Link.pageUrl != "" {
enc = C_AMF_EncodeNamedString(enc, av_pageUrl, r.Link.pageUrl)
if enc == nil {
return errEncoding
}
}
}
if r.fEncoding != 0.0 || r.bSendEncoding {
enc = C_AMF_EncodeNamedNumber(enc, av_objectEncoding, r.fEncoding)
if enc == nil {
return errEncoding
}
}
if copy(enc, []byte{0, 0, AMF_OBJECT_END}) != 3 {
return errCopying // TODO: is this even possible?
}
enc = enc[3:]
/* add auth string */
if r.Link.auth != "" {
enc = C_AMF_EncodeBoolean(enc, r.Link.lFlags&RTMP_LF_AUTH != 0)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeString(enc, r.Link.auth)
if enc == nil {
return errEncoding
}
}
for i := range r.Link.extras.o_props {
enc = C_AMF_PropEncode(&r.Link.extras.o_props[i], enc)
if enc == nil {
return errEncoding
}
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 1)
}
// int RTMP_SendCreateStream(RTMP* r);
// rtmp.c +1725
func C_RTMP_SendCreateStream(r *C_RTMP) error {
var pbuf [256]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_createStream)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 1)
}
// int SendReleaseStream(RTMP* r);
// rtmp.c +1816
func C_SendReleaseStream(r *C_RTMP) error {
var pbuf [1024]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_releaseStream)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, r.Link.playpath)
if enc == nil {
return errEncoding
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 0)
}
// int SendFCPublish(RTMP* r);
// rtmp.c +1846
func C_SendFCPublish(r *C_RTMP) error {
var pbuf [1024]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_FCPublish)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, r.Link.playpath)
if enc == nil {
return errEncoding
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 0)
}
// int SendFCUnpublish(RTMP *r);
// rtmp.c +1875
func C_SendFCUnpublish(r *C_RTMP) error {
var pbuf [1024]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_FCUnpublish)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, r.Link.playpath)
if enc == nil {
return errEncoding
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 0)
}
// int SendPublish(RTMP* r);
// rtmp.c +1908
func C_SendPublish(r *C_RTMP) error {
var pbuf [1024]byte
packet := C_RTMPPacket{
nChannel: 0x04, /* source channel (invoke) */
headerType: RTMP_PACKET_SIZE_LARGE,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: r.streamID,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_publish)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeString(enc, r.Link.playpath)
if enc == nil {
return errEncoding
}
enc = C_AMF_EncodeString(enc, av_live)
if enc == nil {
return errEncoding
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 1)
}
// int
// SendDeleteStream(RTMP *r, double dStreamId)
// rtmp.c +1942
func C_SendDeleteStream(r *C_RTMP, dStreamId float64) error {
var pbuf [256]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av_deleteStream)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
enc = C_AMF_EncodeNumber(enc, dStreamId)
if enc == nil {
return errEncoding
}
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
/* no response expected */
return C_RTMP_SendPacket(r, &packet, 0)
}
// int SendBytesReceived(RTMP* r);
// rtmp.c +2080
func C_SendBytesReceived(r *C_RTMP) error {
var pbuf [256]byte
packet := C_RTMPPacket{
nChannel: 0x02, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_MEDIUM,
packetType: RTMP_PACKET_TYPE_BYTES_READ_REPORT,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
r.nBytesInSent = r.nBytesIn
enc = C_AMF_EncodeInt32(enc, r.nBytesIn)
if enc == nil {
return errEncoding
}
packet.nBodySize = 4
return C_RTMP_SendPacket(r, &packet, 0)
}
// int SendCheckBW(RTMP* r);
// rtmp.c +2105
func C_SendCheckBW(r *C_RTMP) error {
var pbuf [256]byte
packet := C_RTMPPacket{
nChannel: 0x03, /* control channel (invoke) */
headerType: RTMP_PACKET_SIZE_LARGE,
packetType: RTMP_PACKET_TYPE_INVOKE,
nTimeStamp: 0,
nInfoField2: 0,
hasAbsTimestamp: false,
header: pbuf[:],
body: pbuf[RTMP_MAX_HEADER_SIZE:],
}
enc := packet.body
enc = C_AMF_EncodeString(enc, av__checkbw)
if enc == nil {
return errEncoding
}
r.numInvokes++
enc = C_AMF_EncodeNumber(enc, float64(r.numInvokes))
if enc == nil {
return errEncoding
}
enc[0] = AMF_NULL
enc = enc[1:]
packet.nBodySize = uint32((len(pbuf) - RTMP_MAX_HEADER_SIZE) - len(enc))
return C_RTMP_SendPacket(r, &packet, 0)
}
// void AV_erase(C_RTMP_METHOD* vals, int* num, int i, int freeit);
// rtmp.c +2393
func C_AV_erase(m []C_RTMP_METHOD, i int) []C_RTMP_METHOD {
copy(m[i:], m[i+1:])
m[len(m)-1] = C_RTMP_METHOD{}
return m[:len(m)-1]
}
// int HandleInvoke(RTMP* r, const char* body, unsigned int nBodySize);
// rtmp.c +2912
// Side effects: r.bPlaying set to true upon av_NetStream_Publish_Start
func C_HandleInvoke(r *C_RTMP, body []byte) error {
if body[0] != 0x02 {
return errInvalidBody
}
var obj C_AMFObject
nRes := C_AMF_Decode(&obj, body, 0)
if nRes < 0 {
return errDecoding
}
// NOTE we don't really need this ?? still functions without it
//C.AMF_Dump(&obj)
//C.AMFProp_GetString(C_AMF_GetProp(&obj, nil, 0), &method)
method := C_AMFProp_GetString(C_AMF_GetProp(&obj, "", 0))
txn := C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 1))
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s, server invoking <%s>", __FUNCTION__, method.av_val);
switch method {
case av__result:
var methodInvoked string
for i, m := range r.methodCalls {
if float64(m.num) == txn {
methodInvoked = m.name
r.methodCalls = C_AV_erase(r.methodCalls, i)
break
}
}
if methodInvoked == "" {
// TODO use new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result id %f without matching request",
//__FUNCTION__, txn);
goto leave
}
// TODO use new logger here
//RTMP_Log(RTMP_LOGDEBUG, "%s, received result for method call <%s>", __FUNCTION__,
//methodInvoked.av_val);
switch methodInvoked {
case av_connect:
if r.Link.token != "" {
panic("No support for link token")
}
if (r.Link.protocol & RTMP_FEATURE_WRITE) != 0 {
C_SendReleaseStream(r)
C_SendFCPublish(r)
} else {
panic("Link protocol has no RTMP_FEATURE_WRITE")
}
C_RTMP_SendCreateStream(r)
if (r.Link.protocol & RTMP_FEATURE_WRITE) == 0 {
panic("Link protocol has no RTMP_FEATURE_WRITE")
}
case av_createStream:
r.streamID = int32(C_AMFProp_GetNumber(C_AMF_GetProp(&obj, "", 3)))
if r.Link.protocol&RTMP_FEATURE_WRITE != 0 {
C_SendPublish(r)
} else {
panic("Link protocol has no RTMP_FEATURE_WRITE")
}
case av_play, av_publish:
panic("Unsupported method av_play/av_publish")
}
//C.free(unsafe.Pointer(methodInvoked.av_val))
case av_onBWDone:
if r.nBWCheckCounter == 0 {
C_SendCheckBW(r)
}
case av_onFCUnsubscribe, av_onFCSubscribe:
panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe")
case av_ping:
panic("Unsupported method av_ping")
case av__onbwcheck:
panic("Unsupported method av_onbwcheck")
case av__onbwdone:
panic("Unsupported method av_onbwdone")
case av_close:
panic("Unsupported method av_close")
case av_onStatus:
var obj2 C_AMFObject
C_AMFProp_GetObject(C_AMF_GetProp(&obj, "", 3), &obj2)
code := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_code, -1))
level := C_AMFProp_GetString(C_AMF_GetProp(&obj2, av_level, -1)) // Not used.
_ = level
// TODO use new logger
// RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val);
switch code {
case av_NetStream_Failed, av_NetStream_Play_Failed,
av_NetStream_Play_StreamNotFound, av_NetConnection_Connect_InvalidApp:
panic("Unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp")
case av_NetStream_Play_Start, av_NetStream_Play_PublishNotify:
panic("Unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify")
case av_NetStream_Publish_Start:
r.bPlaying = true
for i, m := range r.methodCalls {
if m.name == av_publish {
r.methodCalls = C_AV_erase(r.methodCalls, i)
break
}
}
case av_NetStream_Play_Complete, av_NetStream_Play_Stop, av_NetStream_Play_UnpublishNotify:
panic("Unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify")
case av_NetStream_Seek_Notify:
panic("Unsupported method av_netStream_Seek_Notify")
case av_NetStream_Pause_Notify:
panic("Unsupported method av_NetStream_Pause_Notify")
}
case av_playlist_ready:
panic("Unsupported method av_playlist_ready")
default:
panic(fmt.Sprintf("unknown method: %q", method))
}
leave:
C_AMF_Reset(&obj)
// None of the methods we implement will result in a true return.
return nil
}
// void HandleChangeChunkSize(RTMP* r, const RTMPPacket* packet);
// rtmp.c +3345
func C_HandleChangeChunkSize(r *C_RTMP, packet *C_RTMPPacket) {
if packet.nBodySize >= 4 {
//r.inChunkSize = int32(C.AMF_DecodeInt32((*byte)(unsafe.Pointer(packet.body))))
r.inChunkSize = int32(C_AMF_DecodeInt32(packet.body[:4]))
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s, received: chunk size change to %d", __FUNCTION__, r.inChunkSize);
}
}
// void HandleServerBW(RTMP* r, const RTMPPacket* packet);
// rtmp.c +3508
func C_HandlServerBW(r *C_RTMP, packet *C_RTMPPacket) {
r.nServerBW = int32(C_AMF_DecodeInt32(packet.body[:4]))
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s: server BW = %d", __FUNCTION__, r.nServerBW);
}
// void HandleClientBW(RTMP* r, const RTMPPacket* packet);
// rtmp.c +3515
func C_HandleClientBW(r *C_RTMP, packet *C_RTMPPacket) {
r.nClientBW = int32(C_AMF_DecodeInt32(packet.body[:4]))
//r.nClientBW = int32(C.AMF_DecodeInt32((*byte)(unsafe.Pointer(packet.body))))
if packet.nBodySize > 4 {
r.nClientBW2 = packet.body[4]
} else {
r.nClientBW2 = 0xff
}
// TODO use new logger here
// RTMP_Log(RTMP_LOGDEBUG, "%s: client BW = %d %d", __FUNCTION__, r.nClientBW,
//r.nClientBW2);
}
// static int DecodeInt32LE(const char* data);
// rtmp.c +3527
func C_DecodeInt32LE(data []byte) int32 {
return int32(data[3])<<24 | int32(data[2])<<16 | int32(data[1])<<8 | int32(data[0])
}
// int EncodeInt32LE(char* output, int nVal);
// rtmp.c +3537
func C_EncodeInt32LE(dst []byte, v int32) int32 {
binary.LittleEndian.PutUint32(dst, uint32(v))
return 4
}
// int RTMP_ReadPacket(RTMP* r, RTMPPacket* packet);
// rtmp.c +3550
func C_RTMP_ReadPacket(r *C_RTMP, packet *C_RTMPPacket) error {
var hbuf [RTMP_MAX_HEADER_SIZE]byte
header := hbuf[:]
err := C_ReadN(r, header[:1])
if err != nil {
log.Println("C_RTMP_ReadPacket: failed to read RTMP packet header!")
return err
}
packet.headerType = (header[0] & 0xc0) >> 6
packet.nChannel = int32(header[0] & 0x3f)
header = header[1:]
switch {
case packet.nChannel == 0:
err = C_ReadN(r, header[:1])
if err != nil {
log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header 2nd byte.")
return err
}
header = header[1:]
packet.nChannel = int32(header[0]) + 64
case packet.nChannel == 1:
err = C_ReadN(r, header[:2])
if err != nil {
log.Println("C_RTMP_ReadPacket: failed to read RTMP packet 3rd byte")
return err
}
header = header[2:]
packet.nChannel = int32(binary.BigEndian.Uint16(header[:2])) + 64
}
if packet.nChannel >= r.channelsAllocatedIn {
n := packet.nChannel + 10
timestamp := append(r.channelTimestamp, make([]int32, 10)...)
var packets []*C_RTMPPacket
if r.vecChannelsIn == nil {
packets = make([]*C_RTMPPacket, n)
} else {
packets = append(r.vecChannelsIn[:packet.nChannel:packet.nChannel], make([]*C_RTMPPacket, 10)...)
}
r.channelTimestamp = timestamp
r.vecChannelsIn = packets
for i := int(r.channelsAllocatedIn); i < len(r.channelTimestamp); i++ {
r.channelTimestamp[i] = 0
}
for i := int(r.channelsAllocatedIn); i < int(n); i++ {
r.vecChannelsIn[i] = nil
}
r.channelsAllocatedIn = n
}
nSize := packetSize[packet.headerType]
switch {
case nSize == RTMP_LARGE_HEADER_SIZE:
packet.hasAbsTimestamp = true
case nSize < RTMP_LARGE_HEADER_SIZE:
if r.vecChannelsIn[packet.nChannel] != nil {
*packet = *(r.vecChannelsIn[packet.nChannel])
}
}
nSize--
if nSize > 0 {
err = C_ReadN(r, header[:nSize])
if err != nil {
log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header.")
return err
}
}
hSize := len(hbuf) - len(header) + nSize
if nSize >= 3 {
packet.nTimeStamp = C_AMF_DecodeInt24(header[:3])
if nSize >= 6 {
packet.nBodySize = C_AMF_DecodeInt24(header[3:6])
packet.nBytesRead = 0
if nSize > 6 {
packet.packetType = header[6]
if nSize == 11 {
packet.nInfoField2 = C_DecodeInt32LE(header[7:11])
}
}
}
}
extendedTimestamp := packet.nTimeStamp == 0xffffff
if extendedTimestamp {
err = C_ReadN(r, header[nSize:nSize+4])
if err != nil {
log.Println("RTMPRead_Packet: Failed to read extended timestamp")
return err
}
// TODO: port this
packet.nTimeStamp = C_AMF_DecodeInt32(header[nSize : nSize+4])
hSize += 4
}
if packet.nBodySize > 0 && packet.body == nil {
// TODO: port this
C_RTMPPacket_Alloc(packet, packet.nBodySize)
packet.headerType = (hbuf[0] & 0xc0) >> 6
}
nToRead := int32(packet.nBodySize - packet.nBytesRead)
nChunk := r.inChunkSize
if nToRead < nChunk {
nChunk = nToRead
}
if packet.chunk != nil {
packet.chunk.headerSize = int32(hSize)
copy(packet.chunk.header[:], hbuf[:hSize])
packet.chunk.data = packet.body[packet.nBytesRead : packet.nBytesRead+uint32(nChunk)]
}
err = C_ReadN(r, packet.body[packet.nBytesRead:][:nChunk])
if err != nil {
log.Println("C_RTMP_ReadPacket: failed to read RTMP packet body")
return err
}
packet.nBytesRead += uint32(nChunk)
// keep the packet as ref for other packets on this channel
if r.vecChannelsIn[packet.nChannel] == nil {
r.vecChannelsIn[packet.nChannel] = &C_RTMPPacket{}
}
*(r.vecChannelsIn[packet.nChannel]) = *packet
if extendedTimestamp {
r.vecChannelsIn[packet.nChannel].nTimeStamp = 0xffffff
}
// TODO: port this
if C_RTMPPacket_IsReady(packet) {
if !packet.hasAbsTimestamp {
// timestamps seem to always be relative
packet.nTimeStamp += uint32(r.channelTimestamp[packet.nChannel])
}
r.channelTimestamp[packet.nChannel] = int32(packet.nTimeStamp)
r.vecChannelsIn[packet.nChannel].body = nil
r.vecChannelsIn[packet.nChannel].nBytesRead = 0
r.vecChannelsIn[packet.nChannel].hasAbsTimestamp = false
} else {
packet.body = nil /* so it won't be erased on free */
}
return nil
}
// int HandShake(RTMP* r, int FP9HandShake);
// rtmp.c +3744
func C_HandShake(r *C_RTMP, FP9HandShake int32) error {
var clientbuf [RTMP_SIG_SIZE + 1]byte
clientsig := clientbuf[1:]
var serversig [RTMP_SIG_SIZE]byte
clientbuf[0] = 0x03 // not encrypted
binary.BigEndian.PutUint32(clientsig, uint32(C_RTMP_GetTime()))
copy(clientsig[4:8], []byte{0, 0, 0, 0})
for i := 8; i < RTMP_SIG_SIZE; i++ {
clientsig[i] = byte(rand.Intn(256))
}
err := C_WriteN(r, clientbuf[:])
if err != nil {
return err
}
var typ [1]byte
err = C_ReadN(r, typ[:])
if err != nil {
return err
}
if debugMode {
log.Printf("C_HandShake: Type answer: %v\n", typ[0])
}
if typ[0] != clientbuf[0] {
log.Printf("C_HandShake: type mismatch: client sent %v, server sent: %v\n",
clientbuf[0], typ)
}
err = C_ReadN(r, serversig[:])
if err != nil {
return err
}
// decode server response
suptime := binary.BigEndian.Uint32(serversig[:4])
_ = suptime
// RTMP_Log(RTMP_LOGDEBUG, "%s: Server Uptime : %d", __FUNCTION__, suptime)
// RTMP_Log(RTMP_LOGDEBUG, "%s: FMS Version : %d.%d.%d.%d", __FUNCTION__,
// serversig[4], serversig[5], serversig[6], serversig[7])
// 2nd part of handshake
err = C_WriteN(r, serversig[:])
if err != nil {
return err
}
err = C_ReadN(r, serversig[:])
if err != nil {
return err
}
if !bytes.Equal(serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1]) {
log.Printf("Client signature does not match: %q != %q",
serversig[:RTMP_SIG_SIZE], clientbuf[1:RTMP_SIG_SIZE+1])
}
return nil
}
// int RTMP_SendPacket(RTMP* r, RTMPPacket* packet, int queue);
// rtmp.c +3896
func C_RTMP_SendPacket(r *C_RTMP, packet *C_RTMPPacket, queue int) error {
var prevPacket *C_RTMPPacket
var last int
if packet.nChannel >= r.channelsAllocatedOut {
n := int(packet.nChannel + 10)
var packets []*C_RTMPPacket
if r.vecChannelsOut == nil {
packets = make([]*C_RTMPPacket, n)
} else {
packets = append(r.vecChannelsOut[:packet.nChannel:packet.nChannel], make([]*C_RTMPPacket, 10)...)
}
r.vecChannelsOut = packets
for i := int(r.channelsAllocatedOut); i < n; i++ {
r.vecChannelsOut[i] = nil
}
r.channelsAllocatedOut = int32(n)
}
prevPacket = r.vecChannelsOut[packet.nChannel]
if prevPacket != nil && packet.headerType != RTMP_PACKET_SIZE_LARGE {
// compress a bit by using the prev packet's attributes
if prevPacket.nBodySize == packet.nBodySize && prevPacket.packetType == packet.packetType && packet.headerType == RTMP_PACKET_SIZE_MEDIUM {
packet.headerType = RTMP_PACKET_SIZE_SMALL
}
if prevPacket.nTimeStamp == packet.nTimeStamp && packet.headerType == RTMP_PACKET_SIZE_SMALL {
packet.headerType = RTMP_PACKET_SIZE_MINIMUM
}
last = int(prevPacket.nTimeStamp)
}
if packet.headerType > 3 {
log.Printf("Sanity failed! trying to send header of type: 0x%02x.",
packet.headerType)
return errInvalidHeader
}
var headBytes []byte
var origIdx int
if packet.body != nil {
// Span from -packetsize for the type to the start of the body.
headBytes = packet.header
origIdx = RTMP_MAX_HEADER_SIZE - packetSize[packet.headerType]
} else {
// Allocate a new header and allow 6 bytes of movement backward.
var hbuf [RTMP_MAX_HEADER_SIZE]byte
headBytes = hbuf[:]
origIdx = 6
}
var cSize int
switch {
case packet.nChannel > 319:
cSize = 2
case packet.nChannel > 63:
cSize = 1
}
hSize := packetSize[packet.headerType]
if cSize != 0 {
origIdx -= cSize
hSize += cSize
}
var ts uint32
if prevPacket != nil {
ts = uint32(int(packet.nTimeStamp) - last)
}
if ts >= 0xffffff {
origIdx -= 4
hSize += 4
log.Printf("Larger timestamp than 24-bit: 0x%v", ts)
}
headerIdx := origIdx
c := packet.headerType << 6
switch cSize {
case 0:
c |= byte(packet.nChannel)
case 1:
// Do nothing.
case 2:
c |= 1
}
headBytes[headerIdx] = c
headerIdx++
if cSize != 0 {
tmp := packet.nChannel - 64
headBytes[headerIdx] = byte(tmp & 0xff)
headerIdx++
if cSize == 2 {
headBytes[headerIdx] = byte(tmp >> 8)
headerIdx++
}
}
if packetSize[packet.headerType] > 1 {
res := ts
if ts > 0xffffff {
res = 0xffffff
}
C_AMF_EncodeInt24(headBytes[headerIdx:], int32(res))
headerIdx += 3 // 24bits
}
if packetSize[packet.headerType] > 4 {
C_AMF_EncodeInt24(headBytes[headerIdx:], int32(packet.nBodySize))
headerIdx += 3 // 24bits
headBytes[headerIdx] = packet.packetType
headerIdx++
}
if packetSize[packet.headerType] > 8 {
n := int(C_EncodeInt32LE(headBytes[headerIdx:headerIdx+4], packet.nInfoField2))
headerIdx += n
}
if ts >= 0xffffff {
C_AMF_EncodeInt32(headBytes[headerIdx:], int32(ts))
headerIdx += 4 // 32bits
}
nSize := int(packet.nBodySize)
nChunkSize := int(r.outChunkSize)
if debugMode {
log.Printf("C_RTMP_SendPacket: %v->%v, size=%v", r.Link.conn.LocalAddr(), r.Link.conn.RemoteAddr(), nSize)
}
// Send the previously deferred packet if combining it with the next packet would exceed the chunk size.
if r.defered != nil && len(r.defered)+nSize+hSize > nChunkSize {
err := C_WriteN(r, r.defered)
if err != nil {
return err
}
r.defered = nil
}
// TODO(kortschak): Rewrite this horrific peice of premature optimisation.
// NB: RTMP wants packets in chunks which are 128 bytes by default, but the server may request a different size.
for nSize+hSize != 0 {
if r.defered == nil && packet.packetType == RTMP_PACKET_TYPE_AUDIO && nSize < nChunkSize {
r.defered = headBytes[origIdx:][:nSize+hSize]
break
}
if nChunkSize > nSize {
nChunkSize = nSize
}
bytes := headBytes[origIdx:][:nChunkSize+hSize]
if r.defered != nil {
// Prepend the previously deferred packet and write it with the current one.
bytes = append(r.defered, bytes...)
}
err := C_WriteN(r, bytes)
if err != nil {
return err
}
r.defered = nil
nSize -= nChunkSize
origIdx += nChunkSize + hSize
hSize = 0
if nSize > 0 {
origIdx -= 1 + cSize
hSize = 1 + cSize
if ts >= 0xffffff {
origIdx -= 4
hSize += 4
}
headBytes[origIdx] = 0xc0 | c
if cSize != 0 {
tmp := int(packet.nChannel) - 64
headBytes[origIdx+1] = byte(tmp)
if cSize == 2 {
headBytes[origIdx+2] = byte(tmp >> 8)
}
}
if ts >= 0xffffff {
extendedTimestamp := headBytes[origIdx+1+cSize:]
C_AMF_EncodeInt32(extendedTimestamp[:4], int32(ts))
}
}
}
// We invoked a remote method
// TODO: port the const
if packet.packetType == RTMP_PACKET_TYPE_INVOKE {
buf := packet.body[1:]
method := C_AMF_DecodeString(buf)
if debugMode {
log.Printf("invoking %v", method)
}
// keep it in call queue till result arrives
if queue != 0 {
buf = buf[3+len(method):]
txn := int32(C_AMF_DecodeNumber(buf[:8]))
r.methodCalls = append(r.methodCalls, C_RTMP_METHOD{name: method, num: txn})
}
}
if r.vecChannelsOut[packet.nChannel] == nil {
r.vecChannelsOut[packet.nChannel] = &C_RTMPPacket{}
}
*(r.vecChannelsOut[packet.nChannel]) = *packet
return nil
}
// void RTMP_Close(RTMP *r);
// rtmp.c +4168
func C_RTMP_Close(r *C_RTMP) {
C_CloseInternal(r, false)
}
// static void CloseInternal(RTMP *r, int reconnect);
// rtmp.c +4175
func C_CloseInternal(r *C_RTMP, reconnect bool) {
var i int32
if C_RTMP_IsConnected(r) {
if r.streamID > 0 {
i = r.streamID
if r.Link.protocol&RTMP_FEATURE_WRITE != 0 {
C_SendFCUnpublish(r)
}
C_SendDeleteStream(r, float64(i))
}
err := r.Link.conn.Close()
if err != nil && debugMode {
log.Printf("C_RTMPSockBuf_Close error: %v\n", err)
}
}
r.streamID = -1
r.Link.conn = nil
r.nBWCheckCounter = 0
r.nBytesIn = 0
r.nBytesInSent = 0
r.write.nBytesRead = 0
C_RTMPPacket_Free(&r.write)
// NOTE: C frees - not using in our case
for i := 0; i < int(r.channelsAllocatedIn); i++ {
if r.vecChannelsIn[i] != nil {
r.vecChannelsIn[i] = nil
}
}
//C.free(unsafe.Pointer(r.vecChannelsOut))
r.vecChannelsOut = nil
r.channelsAllocatedOut = 0
r.methodCalls = nil //C_AV_clear(r.methodCalls, r.numCalls)
r.methodCalls = r.methodCalls[:0]
r.numInvokes = 0
r.bPlaying = false
r.msgCounter = 0
r.resplen = 0
r.unackd = 0
if ((r.Link.lFlags & RTMP_LF_FTCU) != 0) && !reconnect {
r.Link.app = ""
r.Link.lFlags ^= RTMP_LF_FAPU
}
if !reconnect {
r.Link.playpath0 = ""
}
}
/// int RTMP_Write(RTMP* r, const char* buf, int size);
// rtmp.c +5136
func C_RTMP_Write(r *C_RTMP, buf []byte) error {
// TODO: port RTMPPacket
var pkt = &r.write
var enc []byte
size := len(buf)
var num int
pkt.nChannel = 0x04
pkt.nInfoField2 = r.streamID
for len(buf) != 0 {
if pkt.nBytesRead == 0 {
if size < minDataSize {
return errTinyPacket
}
if buf[0] == 'F' && buf[1] == 'L' && buf[2] == 'V' {
buf = buf[13:]
}
pkt.packetType = buf[0]
buf = buf[1:]
pkt.nBodySize = C_AMF_DecodeInt24(buf[:3])
buf = buf[3:]
pkt.nTimeStamp = C_AMF_DecodeInt24(buf[:3])
buf = buf[3:]
pkt.nTimeStamp |= uint32(buf[0]) << 24
buf = buf[4:]
pkt.headerType = RTMP_PACKET_SIZE_MEDIUM
switch pkt.packetType {
case RTMP_PACKET_TYPE_VIDEO, RTMP_PACKET_TYPE_AUDIO:
if pkt.nTimeStamp == 0 {
pkt.headerType = RTMP_PACKET_SIZE_LARGE
}
case RTMP_PACKET_TYPE_INFO:
pkt.headerType = RTMP_PACKET_SIZE_LARGE
pkt.nBodySize += 16
}
// TODO: Port this
C_RTMPPacket_Alloc(pkt, pkt.nBodySize)
enc = pkt.body[:pkt.nBodySize]
if pkt.packetType == RTMP_PACKET_TYPE_INFO {
enc = C_AMF_EncodeString(enc, setDataFrame)
if enc == nil {
return errEncoding
}
pkt.nBytesRead = uint32(len(pkt.body) - len(enc))
}
} else {
enc = pkt.body[:pkt.nBodySize][pkt.nBytesRead:]
}
num = int(pkt.nBodySize - pkt.nBytesRead)
if num > len(buf) {
num = len(buf)
}
copy(enc[:num], buf[:num])
pkt.nBytesRead += uint32(num)
buf = buf[num:]
if pkt.nBytesRead == pkt.nBodySize {
err := C_RTMP_SendPacket(r, pkt, 0)
C_RTMPPacket_Free(pkt)
pkt.nBytesRead = 0
if err != nil {
return err
}
if len(buf) < 4 {
return nil
}
buf = buf[4:]
}
}
return nil
}
var rtmpErrs = [...]string{
1: "rtmp: not connected",
2: "rtmp: write error",
3: "rtmp: not started",
}
type Err uint
func (e Err) Error() string {
if 0 <= int(e) && int(e) < len(rtmpErrs) {
s := rtmpErrs[e]
if s != "" {
return s
}
}
return "rtmp: " + strconv.Itoa(int(e))
}
Reference in New Issue View Git Blame Copy Permalink