/* NAME rtmp.go DESCRIPTION See Readme.md AUTHORS Saxon Nelson-Milton Dan Kortschak 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" "log" "math/rand" "strconv" "strings" "time" "unsafe" ) const _Gi = 1 << 30 func pp2b(b, e *byte) []byte { if b == nil { return nil } base := unsafe.Pointer(b) len := uintptr(unsafe.Pointer(e)) - uintptr(base) return (*[_Gi]byte)(base)[:len] } func pl2b(b *byte, l int) []byte { if b == nil { return nil } base := unsafe.Pointer(b) return (*[_Gi]byte)(base)[:l] } func bAddr(buf []byte) *byte { if len(buf) == 0 { return nil } return &buf[0] } const ( minDataSize = 11 debugMode = false 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", } ) func startSession(rtmp *C_RTMP, u string, timeout uint32) (*C_RTMP, error) { connect_timeout := int32(timeout) rtmp = C_RTMP_Alloc() C_RTMP_Init(rtmp) rtmp.Link.timeout = connect_timeout if !C_RTMP_SetupURL(rtmp, u) { C_RTMP_Close(rtmp) //C.RTMP_Free(rtmp) return nil, errors.New("rtmp startSession: Failed to setup URL!") } C_RTMP_EnableWrite(rtmp) C_RTMP_SetBufferMS(rtmp, 3600*1000) if !C_RTMP_Connect(rtmp, nil) { C_RTMP_Close(rtmp) //C.RTMP_Free(rtmp) return nil, errors.New("rtmp startSession: Failed to connect!") } // TODO: port this if !C_RTMP_ConnectStream(rtmp, 0) { C_RTMP_Close(rtmp) //C.RTMP_Free(rtmp) return nil, errors.New("rtmp startSession: Failed to connect stream!") } return rtmp, nil } func endSession(rtmp *C_RTMP) uint32 { if rtmp == nil { return 3 } C_RTMP_Close(rtmp) //C.RTMP_Free(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) (ok bool) { buf := make([]byte, RTMP_MAX_HEADER_SIZE+nSize) p.m_header = buf[:RTMP_MAX_HEADER_SIZE] p.m_body = buf[RTMP_MAX_HEADER_SIZE:] p.m_nBytesRead = 0 return true } // void RTMPPacket_Free(RTMPPacket* p); // rtmp.c +203 func C_RTMPPacket_Free(p *C_RTMPPacket) { if p.m_body != nil { p.m_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.m_sb.sb_socket = -1 r.m_inChunkSize = RTMP_DEFAULT_CHUNKSIZE r.m_outChunkSize = RTMP_DEFAULT_CHUNKSIZE r.m_nBufferMS = 30000 r.m_nClientBW = 2500000 r.m_nClientBW2 = 2 r.m_nServerBW = 2500000 r.m_fAudioCodecs = 3191.0 r.m_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) int32 { if r.m_sb.sb_socket != -1 { return 1 } return 0 } // void RTMP_SetBufferMS(RTMP *r, int size); // rtmp.c +381 func C_RTMP_SetBufferMS(r *C_RTMP, size int32) { r.m_nBufferMS = size } // void SocksSetup(RTMP *r, C_AVal* sockshost); // rtmp.c +410 func C_SocksSetup(r *C_RTMP, sockshost string) { if sockshost != "" { p := strings.SplitN(sockshost, ":", 2) r.Link.sockshost = p[0] r.Link.socksport = 1080 if len(p) != 1 { port, err := strconv.Atoi(p[1]) if err != nil { port = 1080 log.Println("C_SocksSetup: bad string conversion!") } r.Link.socksport = uint16(port) } } else { r.Link.sockshost = "" r.Link.socksport = 0 } } // 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) (ok bool) { r.Link.protocol, r.Link.hostname, r.Link.port, r.Link.app, r.Link.playpath0, ok = C_RTMP_ParseURL(addr) if !ok { return false } 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.hostname, r.Link.port, r.Link.app) r.Link.lFlags |= RTMP_LF_FTCU } else { r.Link.tcUrl = addr } } C_SocksSetup(r, r.Link.sockshost) 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 true } // int RTMP_Connect1(RTMP* r, RTMPPacket* cp); // rtmp.c +978 func C_RTMP_Connect1(r *C_RTMP, cp *C_RTMPPacket) (ok bool) { if debugMode { log.Println("... connected, handshaking...") } if !C_HandShake(r, 1) { log.Println("C_RTMP_Connect1: handshake failed!") return false } if debugMode { log.Println("... handshaked...") } if !C_SendConnectPacket(r, cp) { log.Println("RTMP connect failed!") return false } return true } // int RTMP_ConnectStream(RTMP* r, int seekTime); // rtmp.c +1099 func C_RTMP_ConnectStream(r *C_RTMP, seekTime int32) (playing bool) { var packet C_RTMPPacket if seekTime > 0 { r.Link.seekTime = seekTime } r.m_mediaChannel = 0 // TODO: read packet for !r.m_bPlaying && C_RTMP_IsConnected(r) != 0 && C_RTMP_ReadPacket(r, &packet) { // TODO: port is ready if C_RTMPPacket_IsReady(&packet) { if packet.m_nBodySize == 0 { continue } if packet.m_packetType == RTMP_PACKET_TYPE_AUDIO || packet.m_packetType == RTMP_PACKET_TYPE_VIDEO || packet.m_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) } } return r.m_bPlaying } // 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.m_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.m_nBodySize); if C_HandleInvoke(r, packet.m_body[:packet.m_nBodySize]) { // 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.m_packetType); } return bHasMediaPacket } // int ReadN(RTMP* r, char* buffer, int n); // rtmp.c +1390 func C_ReadN(r *C_RTMP, buf []byte) int { nOriginalSize := len(buf) r.m_sb.sb_timedout = false for len(buf) != 0 { nBytes := 0 var nRead int avail := int(r.m_sb.sb_size) if avail == 0 { if C_RTMPSockBuf_Fill(&r.m_sb) < 1 { if !r.m_sb.sb_timedout { return 0 } } avail = int(r.m_sb.sb_size) } if len(buf) < avail { nRead = len(buf) } else { nRead = avail } if nRead > 0 { copy(buf, r.m_sb.sb_buf[r.m_sb.sb_start:][:nRead]) r.m_sb.sb_start += nRead r.m_sb.sb_size -= nRead nBytes = nRead r.m_nBytesIn += int32(nRead) if r.m_bSendCounter && r.m_nBytesIn > (r.m_nBytesInSent+r.m_nClientBW/10) { if !C_SendBytesReceived(r) { return 0 } } } if nBytes == 0 { log.Println("RTMP socket closed by peer") C_RTMP_Close(r) break } buf = buf[nBytes:] } return nOriginalSize - len(buf) } // int WriteN(RTMP* r, const char* buffer, int n); // rtmp.c +1502 func C_WriteN(r *C_RTMP, buf []byte) (ok bool) { for len(buf) != 0 { nBytes := int(C_RTMPSockBuf_Send(&r.m_sb, buf)) if nBytes < 0 { if debugMode { log.Println("C_WriteN, RTMP send error") } C_RTMP_Close(r) return false } if nBytes == 0 { break } buf = buf[nBytes:] } // !ok here is equivalent to io.ErrShortWrite. return len(buf) == 0 } // int SendConnectPacket(RTMP* r, RTMPPacket* cp); // rtmp.c +1579 func C_SendConnectPacket(r *C_RTMP, cp *C_RTMPPacket) (ok bool) { if cp != nil { return C_RTMP_SendPacket(r, cp, 1) } var pbuf [4096]byte packet := C_RTMPPacket{ m_nChannel: 0x03, m_headerType: RTMP_PACKET_SIZE_LARGE, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_connect) r.m_numInvokes += 1 enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_OBJECT enc = enc[1:] enc = C_AMF_EncodeNamedString(enc, av_app, r.Link.app) if enc == nil { return false } if r.Link.protocol&RTMP_FEATURE_WRITE != 0 { enc = C_AMF_EncodeNamedString(enc, av_type, av_nonprivate) if enc == nil { return false } } if r.Link.flashVer != "" { enc = C_AMF_EncodeNamedString(enc, av_flashVer, r.Link.flashVer) if enc == nil { return false } } if r.Link.swfUrl != "" { enc = C_AMF_EncodeNamedString(enc, av_swfUrl, r.Link.swfUrl) if enc == nil { return false } } if r.Link.tcUrl != "" { enc = C_AMF_EncodeNamedString(enc, av_tcUrl, r.Link.tcUrl) if enc == nil { return false } } if r.Link.protocol&RTMP_FEATURE_WRITE == 0 { enc = C_AMF_EncodeNamedBoolean(enc, av_fpad, false) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, av_capabilities, 15) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, av_audioCodecs, r.m_fAudioCodecs) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, av_videoCodecs, r.m_fVideoCodecs) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, av_videoFunction, 1) if enc == nil { return false } if r.Link.pageUrl != "" { enc = C_AMF_EncodeNamedString(enc, av_pageUrl, r.Link.pageUrl) if enc == nil { return false } } } if r.m_fEncoding != 0.0 || r.m_bSendEncoding { enc = C_AMF_EncodeNamedNumber(enc, av_objectEncoding, r.m_fEncoding) if enc == nil { return false } } if copy(enc, []byte{0, 0, AMF_OBJECT_END}) != 3 { return false } 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 false } enc = C_AMF_EncodeString(enc, r.Link.auth) if enc == nil { return false } } for i := range r.Link.extras.o_props { enc = C_AMF_PropEncode(&r.Link.extras.o_props[i], enc) if enc == nil { return false } } packet.m_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) (ok bool) { var pbuf [256]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_createStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] packet.m_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) (ok bool) { var pbuf [1024]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_releaseStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] enc = C_AMF_EncodeString(enc, r.Link.playpath) if enc == nil { return false } packet.m_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) (ok bool) { var pbuf [1024]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_FCPublish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] enc = C_AMF_EncodeString(enc, r.Link.playpath) if enc == nil { return false } packet.m_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) (ok bool) { var pbuf [1024]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_FCUnpublish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] enc = C_AMF_EncodeString(enc, r.Link.playpath) if enc == nil { return false } packet.m_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) (ok bool) { var pbuf [1024]byte packet := C_RTMPPacket{ m_nChannel: 0x04, /* source channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_LARGE, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: r.m_stream_id, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_publish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] enc = C_AMF_EncodeString(enc, r.Link.playpath) if enc == nil { return false } enc = C_AMF_EncodeString(enc, av_live) if enc == nil { return false } packet.m_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) (ok bool) { var pbuf [256]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av_deleteStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] enc = C_AMF_EncodeNumber(enc, dStreamId) packet.m_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) (ok bool) { var pbuf [256]byte packet := C_RTMPPacket{ m_nChannel: 0x02, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_MEDIUM, m_packetType: RTMP_PACKET_TYPE_BYTES_READ_REPORT, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body r.m_nBytesInSent = r.m_nBytesIn C_AMF_EncodeInt32(enc, r.m_nBytesIn) packet.m_nBodySize = 4 return C_RTMP_SendPacket(r, &packet, 0) } // int SendCheckBW(RTMP* r); // rtmp.c +2105 func C_SendCheckBW(r *C_RTMP) (ok bool) { var pbuf [256]byte packet := C_RTMPPacket{ m_nChannel: 0x03, /* control channel (invoke) */ m_headerType: RTMP_PACKET_SIZE_LARGE, m_packetType: RTMP_PACKET_TYPE_INVOKE, m_nTimeStamp: 0, m_nInfoField2: 0, m_hasAbsTimestamp: false, m_header: pbuf[:RTMP_MAX_HEADER_SIZE], m_body: pbuf[RTMP_MAX_HEADER_SIZE:], } enc := packet.m_body enc = C_AMF_EncodeString(enc, av__checkbw) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, float64(r.m_numInvokes)) enc[0] = AMF_NULL enc = enc[1:] packet.m_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 func C_HandleInvoke(r *C_RTMP, body []byte) (ok bool) { if body[0] != 0x02 { // TODO use new logger here //RTMP_Log(RTMP_LOGWARNING, "%s, Sanity failed. no string method in invoke packet", //__FUNCTION__); return false } var obj C_AMFObject nRes := C_AMF_Decode(&obj, body, 0) if nRes < 0 { // TODO use new logger here //RTMP_Log(RTMP_LOGERROR, "%s, error decoding invoke packet", __FUNCTION__); return false } // 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.m_methodCalls { if float64(m.num) == txn { methodInvoked = m.name r.m_methodCalls = C_AV_erase(r.m_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.m_stream_id = 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.m_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.m_bPlaying = true for i, m := range r.m_methodCalls { if m.name == av_publish { r.m_methodCalls = C_AV_erase(r.m_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 ok } // void HandleChangeChunkSize(RTMP* r, const RTMPPacket* packet); // rtmp.c +3345 func C_HandleChangeChunkSize(r *C_RTMP, packet *C_RTMPPacket) { if packet.m_nBodySize >= 4 { //r.m_inChunkSize = int32(C.AMF_DecodeInt32((*byte)(unsafe.Pointer(packet.m_body)))) r.m_inChunkSize = int32(C_AMF_DecodeInt32(packet.m_body[:4])) // TODO use new logger here // RTMP_Log(RTMP_LOGDEBUG, "%s, received: chunk size change to %d", __FUNCTION__, r.m_inChunkSize); } } // void HandleServerBW(RTMP* r, const RTMPPacket* packet); // rtmp.c +3508 func C_HandlServerBW(r *C_RTMP, packet *C_RTMPPacket) { r.m_nServerBW = int32(C_AMF_DecodeInt32(packet.m_body[:4])) // TODO use new logger here // RTMP_Log(RTMP_LOGDEBUG, "%s: server BW = %d", __FUNCTION__, r.m_nServerBW); } // void HandleClientBW(RTMP* r, const RTMPPacket* packet); // rtmp.c +3515 func C_HandleClientBW(r *C_RTMP, packet *C_RTMPPacket) { r.m_nClientBW = int32(C_AMF_DecodeInt32(packet.m_body[:4])) //r.m_nClientBW = int32(C.AMF_DecodeInt32((*byte)(unsafe.Pointer(packet.m_body)))) if packet.m_nBodySize > 4 { r.m_nClientBW2 = packet.m_body[4] } else { r.m_nClientBW2 = 0xff } // TODO use new logger here // RTMP_Log(RTMP_LOGDEBUG, "%s: client BW = %d %d", __FUNCTION__, r.m_nClientBW, //r.m_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) (ok bool) { var hbuf [RTMP_MAX_HEADER_SIZE]byte header := hbuf[:] if C_ReadN(r, header[:1]) != 1 { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet header!") return false } packet.m_headerType = (header[0] & 0xc0) >> 6 packet.m_nChannel = int32(header[0] & 0x3f) header = header[1:] switch { case packet.m_nChannel == 0: if C_ReadN(r, header[:1]) != 1 { log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header 2nd byte.") return false } header = header[1:] packet.m_nChannel = int32(header[0]) + 64 case packet.m_nChannel == 1: if C_ReadN(r, header[:2]) != 2 { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet 3rd byte") return false } header = header[2:] packet.m_nChannel = int32(binary.BigEndian.Uint16(header[:2])) + 64 } if packet.m_nChannel >= r.m_channelsAllocatedIn { n := packet.m_nChannel + 10 timestamp := append(r.m_channelTimestamp, make([]int32, 10)...) var packets []*C_RTMPPacket if r.m_vecChannelsIn == nil { packets = make([]*C_RTMPPacket, n) } else { packets = append(r.m_vecChannelsIn[:packet.m_nChannel:packet.m_nChannel], make([]*C_RTMPPacket, 10)...) } r.m_channelTimestamp = timestamp r.m_vecChannelsIn = packets for i := int(r.m_channelsAllocatedIn); i < len(r.m_channelTimestamp); i++ { r.m_channelTimestamp[i] = 0 } for i := int(r.m_channelsAllocatedIn); i < int(n); i++ { r.m_vecChannelsIn[i] = nil } r.m_channelsAllocatedIn = n } nSize := packetSize[packet.m_headerType] switch { case nSize == RTMP_LARGE_HEADER_SIZE: packet.m_hasAbsTimestamp = true case nSize < RTMP_LARGE_HEADER_SIZE: if r.m_vecChannelsIn[packet.m_nChannel] != nil { *packet = *(r.m_vecChannelsIn[packet.m_nChannel]) } } nSize-- if nSize > 0 && C_ReadN(r, header[:nSize]) != int(nSize) { log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header.") return false } hSize := len(hbuf) - len(header) + nSize if nSize >= 3 { packet.m_nTimeStamp = C_AMF_DecodeInt24(header[:3]) if nSize >= 6 { packet.m_nBodySize = C_AMF_DecodeInt24(header[3:6]) packet.m_nBytesRead = 0 if nSize > 6 { packet.m_packetType = header[6] if nSize == 11 { packet.m_nInfoField2 = C_DecodeInt32LE(header[7:11]) } } } } extendedTimestamp := packet.m_nTimeStamp == 0xffffff if extendedTimestamp { if C_ReadN(r, header[nSize:nSize+4]) != 4 { log.Println("RTMPRead_Packet: Failed to read extended timestamp") return false } // TODO: port this packet.m_nTimeStamp = C_AMF_DecodeInt32(header[nSize : nSize+4]) hSize += 4 } if packet.m_nBodySize > 0 && packet.m_body == nil { // TODO: port this if !C_RTMPPacket_Alloc(packet, packet.m_nBodySize) { log.Println("RTMPRead_Packet: failed to allocate packet") return false } packet.m_headerType = (hbuf[0] & 0xc0) >> 6 } nToRead := int32(packet.m_nBodySize - packet.m_nBytesRead) nChunk := r.m_inChunkSize if nToRead < nChunk { nChunk = nToRead } if packet.m_chunk != nil { packet.m_chunk.c_headerSize = int32(hSize) copy(packet.m_chunk.c_header[:], hbuf[:hSize]) packet.m_chunk.c_chunk = packet.m_body[packet.m_nBytesRead : packet.m_nBytesRead+uint32(nChunk)] } if C_ReadN(r, packet.m_body[packet.m_nBytesRead:][:nChunk]) != int(nChunk) { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet body") return false } packet.m_nBytesRead += uint32(nChunk) // keep the packet as ref for other packets on this channel if r.m_vecChannelsIn[packet.m_nChannel] == nil { r.m_vecChannelsIn[packet.m_nChannel] = &C_RTMPPacket{} } *(r.m_vecChannelsIn[packet.m_nChannel]) = *packet if extendedTimestamp { r.m_vecChannelsIn[packet.m_nChannel].m_nTimeStamp = 0xffffff } // TODO: port this if C_RTMPPacket_IsReady(packet) { if !packet.m_hasAbsTimestamp { // timestamps seem to always be relative packet.m_nTimeStamp += uint32(r.m_channelTimestamp[packet.m_nChannel]) } r.m_channelTimestamp[packet.m_nChannel] = int32(packet.m_nTimeStamp) r.m_vecChannelsIn[packet.m_nChannel].m_body = nil r.m_vecChannelsIn[packet.m_nChannel].m_nBytesRead = 0 r.m_vecChannelsIn[packet.m_nChannel].m_hasAbsTimestamp = false } else { packet.m_body = nil /* so it won't be erased on free */ } return true } // int HandShake(RTMP* r, int FP9HandShake); // rtmp.c +3744 func C_HandShake(r *C_RTMP, FP9HandShake int32) (ok bool) { 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)) } if !C_WriteN(r, clientbuf[:]) { return false } var typ [1]byte if C_ReadN(r, typ[:]) != 1 { return false } 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) } if C_ReadN(r, serversig[:]) != RTMP_SIG_SIZE { return false } // 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 if !C_WriteN(r, serversig[:]) { return false } if C_ReadN(r, serversig[:]) != RTMP_SIG_SIZE { return false } 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 true } // int RTMP_SendPacket(RTMP* r, RTMPPacket* packet, int queue); // rtmp.c +3896 func C_RTMP_SendPacket(r *C_RTMP, packet *C_RTMPPacket, queue int) (ok bool) { var prevPacket *C_RTMPPacket var last int if packet.m_nChannel >= r.m_channelsAllocatedOut { n := int(packet.m_nChannel + 10) var packets []*C_RTMPPacket if r.m_vecChannelsOut == nil { packets = make([]*C_RTMPPacket, n) } else { packets = append(r.m_vecChannelsOut[:packet.m_nChannel:packet.m_nChannel], make([]*C_RTMPPacket, 10)...) } r.m_vecChannelsOut = packets for i := int(r.m_channelsAllocatedOut); i < n; i++ { r.m_vecChannelsOut[i] = nil } r.m_channelsAllocatedOut = int32(n) } prevPacket = r.m_vecChannelsOut[packet.m_nChannel] if prevPacket != nil && packet.m_headerType != RTMP_PACKET_SIZE_LARGE { // compress a bit by using the prev packet's attributes if prevPacket.m_nBodySize == packet.m_nBodySize && prevPacket.m_packetType == packet.m_packetType && packet.m_headerType == RTMP_PACKET_SIZE_MEDIUM { packet.m_headerType = RTMP_PACKET_SIZE_SMALL } if prevPacket.m_nTimeStamp == packet.m_nTimeStamp && packet.m_headerType == RTMP_PACKET_SIZE_SMALL { packet.m_headerType = RTMP_PACKET_SIZE_MINIMUM } last = int(prevPacket.m_nTimeStamp) } if packet.m_headerType > 3 { log.Printf("Sanity failed! trying to send header of type: 0x%02x.", packet.m_headerType) return false } nSize := packetSize[int(packet.m_headerType)] hSize := nSize cSize := 0 t := uint32(int(packet.m_nTimeStamp) - last) var header, hend *byte if packet.m_body != nil { // Span from -nSize to the start of the body. header = decBytePtr(&packet.m_body[0], nSize) hend = &packet.m_body[0] } else { // Allocate a new header and allow 6 bytes of movement backward. var hbuf [RTMP_MAX_HEADER_SIZE]byte header = incBytePtr(&hbuf[0], 6) hend = incBytePtr(&hbuf[0], RTMP_MAX_HEADER_SIZE) } switch { case packet.m_nChannel > 319: cSize = 2 case packet.m_nChannel > 63: cSize = 1 } if cSize != 0 { header = decBytePtr(header, cSize) hSize += cSize } if t >= 0xffffff { header = decBytePtr(header, 4) hSize += 4 log.Printf("Larger timestamp than 24-bit: 0x%v", t) } hptr := header c := packet.m_headerType << 6 switch cSize { case 0: c |= byte(packet.m_nChannel) case 1: case 2: c |= byte(1) } *hptr = c hptr = incBytePtr(hptr, 1) if cSize != 0 { tmp := packet.m_nChannel - 64 *hptr = byte(tmp & 0xff) hptr = incBytePtr(hptr, 1) if cSize == 2 { *hptr = byte(tmp >> 8) hptr = incBytePtr(hptr, 1) } } if nSize > 1 { res := t if t > 0xffffff { res = 0xffffff } hptr = bAddr(C_AMF_EncodeInt24(pp2b(hptr, hend), int32(res))) } if nSize > 4 { hptr = bAddr(C_AMF_EncodeInt24(pp2b(hptr, hend), int32(packet.m_nBodySize))) *hptr = packet.m_packetType hptr = incBytePtr(hptr, 1) } if nSize > 8 { hptr = incBytePtr(hptr, int(C_EncodeInt32LE(pl2b(hptr, 4), packet.m_nInfoField2))) } if t >= 0xffffff { hptr = bAddr(C_AMF_EncodeInt32(pp2b(hptr, hend), int32(t))) } nSize = int(packet.m_nBodySize) buffer := &packet.m_body[0] nChunkSize := int(r.m_outChunkSize) if debugMode { log.Printf("C_RTMP_SendPacket: fd=%v, size=%v", r.m_sb.sb_socket, nSize) } var tbuf, toff *byte for (nSize + hSize) != 0 { if nSize < nChunkSize { nChunkSize = nSize } if tbuf != nil { copy(pl2b(toff, nChunkSize+hSize), pl2b(header, nChunkSize+hSize)) toff = incBytePtr(toff, nChunkSize+hSize) } else { if !C_WriteN(r, pl2b(header, nChunkSize+hSize)) { return false } } nSize -= nChunkSize buffer = incBytePtr(buffer, nChunkSize) hSize = 0 if nSize > 0 { header = decBytePtr(buffer, 1) hSize = 1 if cSize != 0 { header = decBytePtr(header, cSize) hSize += cSize } if t >= 0xffffff { header = decBytePtr(header, 4) hSize += 4 } *header = 0xc0 | c if cSize != 0 { tmp := int(packet.m_nChannel) - 64 pl2b(incBytePtr(header, 1), 2)[1] = byte(tmp) if cSize == 2 { pl2b(incBytePtr(header, 1), 3)[2] = byte(tmp >> 8) } } if t >= 0xffffff { extendedTimestamp := incBytePtr(header, 1+cSize) C_AMF_EncodeInt32(pl2b(extendedTimestamp, 4), int32(t)) } } } if tbuf != nil { ok := C_WriteN(r, pp2b(tbuf, toff)) tbuf = nil if !ok { return false } } // We invoked a remote method // TODO: port the const if packet.m_packetType == RTMP_PACKET_TYPE_INVOKE { buf := packet.m_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.m_methodCalls = append(r.m_methodCalls, C_RTMP_METHOD{name: method, num: txn}) } } if r.m_vecChannelsOut[packet.m_nChannel] == nil { r.m_vecChannelsOut[packet.m_nChannel] = &C_RTMPPacket{} } *(r.m_vecChannelsOut[packet.m_nChannel]) = *packet return true } // 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) != 0 { if r.m_stream_id > 0 { i = r.m_stream_id if r.Link.protocol&RTMP_FEATURE_WRITE != 0 { C_SendFCUnpublish(r) } C_SendDeleteStream(r, float64(i)) } C_RTMPSockBuf_Close(&r.m_sb) } r.m_stream_id = -1 r.m_sb.sb_socket = -1 r.m_nBWCheckCounter = 0 r.m_nBytesIn = 0 r.m_nBytesInSent = 0 r.m_write.m_nBytesRead = 0 C_RTMPPacket_Free(&r.m_write) // NOTE: C frees - not using in our case for i := 0; i < int(r.m_channelsAllocatedIn); i++ { if r.m_vecChannelsIn[i] != nil { r.m_vecChannelsIn[i] = nil } } //C.free(unsafe.Pointer(r.m_vecChannelsOut)) r.m_vecChannelsOut = nil r.m_channelsAllocatedOut = 0 r.m_methodCalls = nil //C_AV_clear(r.m_methodCalls, r.m_numCalls) r.m_methodCalls = r.m_methodCalls[:0] r.m_numInvokes = 0 r.m_bPlaying = false r.m_sb.sb_size = 0 r.m_msgCounter = 0 r.m_resplen = 0 r.m_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) int { // TODO: port RTMPPacket var pkt = &r.m_write var enc []byte size := len(buf) var num int pkt.m_nChannel = 0x04 pkt.m_nInfoField2 = r.m_stream_id for len(buf) != 0 { if pkt.m_nBytesRead == 0 { if size < minDataSize { log.Printf("size: %d\n", size) log.Printf("too small \n") return 0 } if buf[0] == 'F' && buf[1] == 'L' && buf[2] == 'V' { buf = buf[13:] } pkt.m_packetType = buf[0] buf = buf[1:] pkt.m_nBodySize = C_AMF_DecodeInt24(buf[:3]) buf = buf[3:] pkt.m_nTimeStamp = C_AMF_DecodeInt24(buf[:3]) buf = buf[3:] pkt.m_nTimeStamp |= uint32(buf[0]) << 24 buf = buf[4:] if ((pkt.m_packetType == RTMP_PACKET_TYPE_AUDIO || pkt.m_packetType == RTMP_PACKET_TYPE_VIDEO) && pkt.m_nTimeStamp == 0) || pkt.m_packetType == RTMP_PACKET_TYPE_INFO { pkt.m_headerType = RTMP_PACKET_SIZE_LARGE if pkt.m_packetType == RTMP_PACKET_TYPE_INFO { pkt.m_nBodySize += 16 } } else { pkt.m_headerType = RTMP_PACKET_SIZE_MEDIUM } // TODO: Port this if !C_RTMPPacket_Alloc(pkt, pkt.m_nBodySize) { log.Println("Failed to allocate packet") return 0 } enc = pkt.m_body[:pkt.m_nBodySize] if pkt.m_packetType == RTMP_PACKET_TYPE_INFO { enc = C_AMF_EncodeString(enc, setDataFrame) pkt.m_nBytesRead = uint32(len(pkt.m_body) - len(enc)) } } else { enc = pkt.m_body[:pkt.m_nBodySize][pkt.m_nBytesRead:] } num = int(pkt.m_nBodySize - pkt.m_nBytesRead) if num > len(buf) { num = len(buf) } copy(enc[:num], buf[:num]) pkt.m_nBytesRead += uint32(num) buf = buf[num:] if pkt.m_nBytesRead == pkt.m_nBodySize { // TODO: Port this ok := C_RTMP_SendPacket(r, pkt, 0) // TODO: Port this C_RTMPPacket_Free(pkt) pkt.m_nBytesRead = 0 if !ok { return -1 } if len(buf) < 4 { return size + (len(buf) - 4) } buf = buf[4:] } } return size } // incBytePtr returns an unsafe.Pointer to a byte that is inc positive positions // from the passed ptr func incBytePtr(ptr *byte, inc int) *byte { return (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(ptr)) + uintptr(inc))) } // decBytePtr returns an unsafe.Pointer to a byte that is dec negative positions // from ptr func decBytePtr(ptr *byte, dec int) *byte { return (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(ptr)) - uintptr(dec))) } 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)) }