/* 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 /* #include #include #include #include #include typedef struct sockaddr_in sockaddr_in; typedef struct sockaddr sockaddr; */ import "C" import ( "bytes" "errors" "fmt" "log" "math/rand" "strconv" "time" "unsafe" "github.com/chamaken/cgolmnl/inet" ) const _Gi = 1 << 30 const ( minDataSize = 11 debugMode = false length = 512 ) // av_setDataFrame is a static const global in rtmp.c var ( AV_empty = AVC("") setDataFrame = AVC("@setDataFrame") av_connect = AVC("connect") av_app = AVC("app") av_type = AVC("type") av_nonprivate = AVC("nonprivate") av_flashVer = AVC("flashVer") av_swfUrl = AVC("swfUrl") av_tcUrl = AVC("tcUrl") av_fpad = AVC("fpad") av_capabilities = AVC("capabilities") av_audioCodecs = AVC("audioCodecs") av_videoCodecs = AVC("videoCodecs") av_videoFunction = AVC("videoFunction") av_pageUrl = AVC("pageUrl") av_objectEncoding = AVC("objectEncoding") av__result = AVC("_result") av_secureToken = AVC("secureToken") av_createStream = AVC("createStream") av_releaseStream = AVC("releaseStream") av_play = AVC("play") av_publish = AVC("publish") av_onBWDone = AVC("onBWDone") av_onFCSubscribe = AVC("onFCSubscribe") av_onFCUnsubscribe = AVC("onFCUnsubscribe") av__onbwcheck = AVC("_onbwcheck") av__onbwdone = AVC("_onbwdone") av_ping = AVC("ping") av__checkbw = AVC("_checkbw") av_close = AVC("close") av_onStatus = AVC("onStatus") av_code = AVC("code") av_level = AVC("level") av_NetStream_Failed = AVC("NetStream.Failed") av_NetStream_Play_Failed = AVC("NetStream.Play.Failed") av_NetConnection_Connect_InvalidApp = AVC("NetConnection.Connect.InvalidApp") av_NetStream_Play_StreamNotFound = AVC("NetStream.Play.StreamNotFound") av_NetStream_Play_Start = AVC("NetStream.Play.Start") av_NetStream_Play_PublishNotify = AVC("NetStream.Play.PublishNotify") av_NetStream_Publish_Start = AVC("NetStream.Publish.Start") av_NetStream_Play_Complete = AVC("NetStream.Play.Complete") av_NetStream_Play_Stop = AVC("NetStream.Play.Stop") av_NetStream_Play_UnpublishNotify = AVC("NetStream.Play.UnpublishNotify") av_NetStream_Seek_Notify = AVC("NetStream.Seek.Notify") av_NetStream_Pause_Notify = AVC("NetStream.Pause.Notify") av_playlist_ready = AVC("playlist_ready") av_set_playlist = AVC("set_playlist") av_FCPublish = AVC("FCPublish") av_FCUnpublish = AVC("FCUnpublish") av_live = AVC("live") av_deleteStream = AVC("deleteStream") ) 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) { var ptr *byte // TODO: port C.SIZE_MAX // also work out how to deal with the overfloat /* if int64(nSize) > (C.SIZE_MAX - RTMP_MAX_HEADER_SIZE) { return 0 } */ ptr = (*byte)(calloc(1, uintptr(nSize+RTMP_MAX_HEADER_SIZE))) if ptr == nil { return false } p.m_body = (*byte)(incBytePtr(unsafe.Pointer(ptr), 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 = int32(size) } // void SocksSetup(RTMP *r, C_AVal* sockshost); // rtmp.c +410 func C_SocksSetup(r *C_RTMP, sockshost *C_AVal) { if sockshost.av_len != 0 { socksport := strchr((*byte)(unsafe.Pointer(sockshost.av_val)), ':') hostname := strdup((*byte)(unsafe.Pointer(sockshost.av_val))) if unsafe.Pointer(socksport) != nil { (*[_Gi]byte)(unsafe.Pointer(hostname))[uintptr(decBytePtr(unsafe.Pointer(socksport), int(uintptr(unsafe.Pointer(sockshost.av_val)))))] = '\000' r.Link.sockshost.av_val = (*byte)(unsafe.Pointer(hostname)) r.Link.sockshost.av_len = int32(strlen(hostname)) value, err := strconv.Atoi(string((*[_Gi]byte)(unsafe.Pointer(uintptr( unsafe.Pointer(socksport)) + uintptr(1)))[:strlen((*byte)(unsafe.Pointer( uintptr(unsafe.Pointer(socksport))+uintptr(1))))+1])) if err != nil { log.Println("C_SocksSetup: bad string conversion!") } if uintptr(unsafe.Pointer(socksport)) == 0 { value = 1080 } r.Link.socksport = uint16(value) } } else { r.Link.sockshost.av_val = nil r.Link.sockshost.av_len = 0 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, u string) (ok bool) { url := goStrToCStr(u) var length int32 var port uint32 port = 0 length = strlen(url) // TODO: port this ret := int32(C_RTMP_ParseURL((*byte)(unsafe.Pointer(url)), (*int32)( unsafe.Pointer(&r.Link.protocol)), &r.Link.hostname, (*uint32)( unsafe.Pointer(&port)), &r.Link.playpath0, &r.Link.app)) if ret == 0 { return false } r.Link.port = uint16(port) r.Link.playpath = r.Link.playpath0 if r.Link.tcUrl.av_len == 0 { r.Link.tcUrl.av_val = (*byte)(unsafe.Pointer(url)) if r.Link.app.av_len != 0 { if int(uintptr(unsafe.Pointer(r.Link.app.av_val))) < int(uintptr(incBytePtr(unsafe.Pointer(url), int(length)))) { r.Link.tcUrl.av_len = int32(int(r.Link.app.av_len) + int(uintptr(decBytePtr(unsafe.Pointer(r.Link.app.av_val), int(uintptr(unsafe.Pointer(url))))))) } else { length = int32(r.Link.hostname.av_len) + int32(r.Link.app.av_len) + int32(len("rtmpte://:65535/\x00")) r.Link.tcUrl.av_val = (*byte)(malloc(uintptr(uintptr(length)))) hostname := string((*[_Gi]byte)(unsafe.Pointer(r.Link.hostname.av_val))[:r.Link.hostname.av_len]) app := string((*[_Gi]byte)(unsafe.Pointer(r.Link.app.av_val))[:r.Link.app.av_len]) fString := fmt.Sprintf("%v://%v:%v/%v", RTMPProtocolStringsLower[r.Link.protocol], hostname, r.Link.port, app) r.Link.tcUrl.av_val = (*byte)(bToUP(goStrToCStr(fString))) r.Link.tcUrl.av_len = int32(strLen(RTMPProtocolStringsLower[r.Link.protocol]) + strLen(string("://")) + strLen(hostname) + strLen(string(":")) + strLen(strconv.Itoa(int(r.Link.port))) + strLen(string("/")) + strLen(app)) r.Link.lFlags |= RTMP_LF_FTCU } } else { r.Link.tcUrl.av_len = int32(strlen(url)) } } 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 add_addr_info(struct sockaddr_in *service, AVal *host, int port) // rtmp.c +869 func C_add_addr_info(service *C.sockaddr_in, host *C_AVal, port C.int) (ok bool) { var hostname *byte if (*[1 << 20]byte)(unsafe.Pointer(host.av_val))[host.av_len] != 0 { name := make([]byte, host.av_len+1) copy(name, (*[1 << 20]byte)(unsafe.Pointer(host.av_val))[:host.av_len]) name[len(name)-1] = 0 hostname = (*byte)(unsafe.Pointer(&name[0])) } else { hostname = host.av_val } service.sin_addr.s_addr = C.inet_addr((*C.char)(unsafe.Pointer(hostname))) if service.sin_addr.s_addr == C.INADDR_NONE { host := C.gethostbyname((*C.char)(unsafe.Pointer(hostname))) if host == nil || *host.h_addr_list == nil { //RTMP_Log(RTMP_LOGERROR, "Problem accessing the DNS. (addr: %s)", hostname) return false } service.sin_addr = *(*C.struct_in_addr)(unsafe.Pointer(*host.h_addr_list)) } service.sin_port = C.ushort(inet.Htons(uint16(port))) return true } // int RTMP_Connect0(RTMP *r, struct sockaddr* service); // rtmp.c +906 func C_RTMP_Connect0(r *C_RTMP, service *C.sockaddr) (ok bool) { on := 1 r.m_sb.sb_timedout = 0 r.m_pausing = 0 r.m_fDuration = 0 r.m_sb.sb_socket = int32(C.socket(C.AF_INET, C.SOCK_STREAM, C.IPPROTO_TCP)) if r.m_sb.sb_socket != -1 { if C.connect(C.int(r.m_sb.sb_socket), service, C.socklen_t(unsafe.Sizeof(*service))) < 0 { log.Println("C_RTMP_Connect0, failed to connect socket.") } if r.Link.socksport != 0 { if debugMode { log.Println("C_RTMP_Connect0: socks negotiation.") } if !C_SocksNegotiate(r) { log.Println("C_RTMP_Connect0: SOCKS negotiation failed.") return false } } } else { log.Println("C_RTMP_Connect0: failed to create socket.") return false } { var tv int32 SET_RCVTIMEO(&tv, int32(r.Link.timeout)) if C.setsockopt(C.int(r.m_sb.sb_socket), C.SOL_SOCKET, C.SO_RCVTIMEO, unsafe.Pointer(&tv), C.socklen_t(unsafe.Sizeof(tv))) != 0 { log.Println("C_RTMP_Connect0: Setting socket timeout failed") } } C.setsockopt(C.int(r.m_sb.sb_socket), C.IPPROTO_TCP, C.TCP_NODELAY, unsafe.Pointer(&on), C.socklen_t(unsafe.Sizeof(on))) 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_Connect(RTMP *r, RTMPPacket* cp); // rtmp.c +1032 func C_RTMP_Connect(r *C_RTMP, cp *C_RTMPPacket) (ok bool) { // TODO: port this var service C.sockaddr_in if r.Link.hostname.av_len == 0 { return false } memset((*byte)(unsafe.Pointer(&service)), 0, int(unsafe.Sizeof(service))) // TODO: port this service.sin_family = C.AF_INET if r.Link.socksport != 0 { if !C_add_addr_info(&service, (*C_AVal)(unsafe.Pointer(&r.Link.sockshost)), C.int(r.Link.socksport)) { return false } } else { // connect directly if !C_add_addr_info(&service, (*C_AVal)(unsafe.Pointer(&r.Link.hostname)), C.int(r.Link.port)) { return false } } if !C_RTMP_Connect0(r, (*C.sockaddr)(unsafe.Pointer(&service))) { return false } r.m_bSendCounter = 1 return C_RTMP_Connect1(r, cp) } // int SocksNegotiate(RTMP* r); // rtmp.c +1062 func C_SocksNegotiate(r *C_RTMP) (ok bool) { var addr int32 var service C.sockaddr_in memset((*byte)(unsafe.Pointer(&service)), 0, int(unsafe.Sizeof(service))) C_add_addr_info(&service, (*C_AVal)(unsafe.Pointer(&r.Link.hostname)), C.int(r.Link.port)) addr = int32(inet.Htonl(uint32(service.sin_addr.s_addr))) { packet := []byte{ 4, 1, byte((r.Link.port >> 8) & 0xFF), byte((r.Link.port) & 0xFF), byte(addr>>24) & 0xFF, byte(addr>>16) & 0xFF, byte(addr>>8) & 0xFF, byte(addr) & 0xFF, 0, } C_WriteN(r, unsafe.Pointer(&packet[0]), int(unsafe.Sizeof(packet))) if C_ReadN(r, &packet[0], 8) != 8 { return false } if packet[0] == 0 && packet[1] == 90 { return true } else { // TODO: use new logger here log.Println("C_SocksNegotitate: SOCKS returned error code!") return false } } } // 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 memset((*byte)(unsafe.Pointer(&packet)), 0, int(unsafe.Sizeof(packet))) if seekTime > 0 { r.Link.seekTime = int32(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) != 0 { 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, (*byte)(unsafe.Pointer(packet.m_body)), uint32(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, buffer *byte, n int) int { nOriginalSize := n var avail int var ptr *byte r.m_sb.sb_timedout = 0 ptr = buffer for n > 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 == 0 { return 0 } } avail = int(r.m_sb.sb_size) } if n < avail { nRead = n } else { nRead = avail } if nRead > 0 { memmove(unsafe.Pointer(ptr), unsafe.Pointer(r.m_sb.sb_start), uintptr(nRead)) r.m_sb.sb_start = (*byte)(incBytePtr(unsafe.Pointer(r.m_sb.sb_start), nRead)) r.m_sb.sb_size -= int32(nRead) nBytes = nRead r.m_nBytesIn += int32(nRead) if r.m_bSendCounter != 0 && 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 } n -= nBytes ptr = (*byte)(incBytePtr(unsafe.Pointer(ptr), nBytes)) } return nOriginalSize - n } // int WriteN(RTMP* r, const char* buffer, int n); // rtmp.c +1502 func C_WriteN(r *C_RTMP, buffer unsafe.Pointer, n int) (ok bool) { ptr := buffer for n > 0 { var nBytes int nBytes = int(C_RTMPSockBuf_Send(&r.m_sb, (*byte)(ptr), int32(n))) if nBytes < 0 { if debugMode { log.Println("C_WriteN, RTMP send error") } // TODO: port this C_RTMP_Close(r) n = 1 break } if nBytes == 0 { break } n -= nBytes ptr = incBytePtr(ptr, nBytes) } // !ok here is equivalent to io.ErrShortWrite. return n == 0 } // int SendConnectPacket(RTMP* r, RTMPPacket* cp); // rtmp.c +1579 func C_SendConnectPacket(r *C_RTMP, cp *C_RTMPPacket) (ok bool) { var packet C_RTMPPacket var pbuf [4096]byte pend := (*byte)(unsafe.Pointer(incBytePtr(unsafe.Pointer(&pbuf[0]), int(unsafe.Sizeof(pbuf))))) var enc *byte if cp != nil { return C_RTMP_SendPacket(r, cp, 1) } packet.m_nChannel = 0x03 packet.m_headerType = RTMP_PACKET_SIZE_LARGE packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_connect) r.m_numInvokes += 1 enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) (*[_Gi]byte)(unsafe.Pointer(enc))[0] = AMF_OBJECT enc = (*byte)(unsafe.Pointer(incBytePtr(unsafe.Pointer(enc), 1))) enc = C_AMF_EncodeNamedString(enc, pend, &av_app, &r.Link.app) if enc == nil { return false } if r.Link.protocol&RTMP_FEATURE_WRITE != 0 { enc = C_AMF_EncodeNamedString(enc, pend, &av_type, &av_nonprivate) if enc == nil { return false } } if r.Link.flashVer.av_len != 0 { enc = C_AMF_EncodeNamedString(enc, pend, &av_flashVer, &r.Link.flashVer) if enc == nil { return false } } if r.Link.swfUrl.av_len != 0 { enc = C_AMF_EncodeNamedString(enc, pend, &av_swfUrl, &r.Link.swfUrl) if enc == nil { return false } } if r.Link.tcUrl.av_len != 0 { enc = C_AMF_EncodeNamedString(enc, pend, &av_tcUrl, &r.Link.tcUrl) if enc == nil { return false } } if r.Link.protocol&RTMP_FEATURE_WRITE == 0 { enc = C_AMF_EncodeNamedBoolean(enc, pend, &av_fpad, 0) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, pend, &av_capabilities, 15.0) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, pend, &av_audioCodecs, float64(r.m_fAudioCodecs)) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, pend, &av_videoCodecs, float64(r.m_fVideoCodecs)) if enc == nil { return false } enc = C_AMF_EncodeNamedNumber(enc, pend, &av_videoFunction, 1.0) if enc == nil { return false } if r.Link.pageUrl.av_len != 0 { enc = C_AMF_EncodeNamedString(enc, pend, &av_pageUrl, &r.Link.pageUrl) if enc == nil { return false } } } if r.m_fEncoding != 0.0 || r.m_bSendEncoding != 0 { enc = C_AMF_EncodeNamedNumber(enc, pend, &av_objectEncoding, float64(r.m_fEncoding)) if enc == nil { return false } } if int(uintptr(incBytePtr(unsafe.Pointer(enc), 3))) >= int(uintptr( unsafe.Pointer(pend))) { return false } (*[_Gi]byte)(unsafe.Pointer(enc))[0] = 0 enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) (*[_Gi]byte)(unsafe.Pointer(enc))[0] = 0 enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) (*[_Gi]byte)(unsafe.Pointer(enc))[0] = AMF_OBJECT_END enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) /* add auth string */ if r.Link.auth.av_len != 0 { enc = C_AMF_EncodeBoolean(enc, pend, int(r.Link.lFlags&RTMP_LF_AUTH)) if enc == nil { return false } enc = C_AMF_EncodeString(enc, (*byte)(pend), &r.Link.auth) if enc == nil { return false } } if r.Link.extras.o_num != 0 { for i := 0; i < int(r.Link.extras.o_num); i++ { enc = C_AMF_PropEncode((*C_AMFObjectProperty)(incPtr(unsafe.Pointer( &r.Link.extras.o_props), int(unsafe.Sizeof(r.Link.extras.o_props)), i)), enc, pend) if enc == nil { return false } } } packet.m_nBodySize = uint32(uintptr(decBytePtr(unsafe.Pointer(enc), int(uintptr(unsafe.Pointer(packet.m_body)))))) 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 packet C_RTMPPacket var pbuf [256]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_createStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) return C_RTMP_SendPacket(r, &packet, 1) } // int SendReleaseStream(RTMP* r); // rtmp.c +1816 func C_SendReleaseStream(r *C_RTMP) (ok bool) { var packet C_RTMPPacket var pbuf [1024]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_releaseStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) enc = C_AMF_EncodeString(enc, pend, &r.Link.playpath) if enc == nil { return false } packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) return C_RTMP_SendPacket(r, &packet, 0) } // int SendFCPublish(RTMP* r); // rtmp.c +1846 func C_SendFCPublish(r *C_RTMP) (ok bool) { var packet C_RTMPPacket var pbuf [1024]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_FCPublish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) enc = C_AMF_EncodeString(enc, pend, &r.Link.playpath) if enc == nil { return false } packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) return C_RTMP_SendPacket(r, &packet, 0) } // int SendFCUnpublish(RTMP *r); // rtmp.c +1875 func C_SendFCUnpublish(r *C_RTMP) (ok bool) { var packet C_RTMPPacket var pbuf [1024]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_FCUnpublish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) enc = C_AMF_EncodeString(enc, pend, &r.Link.playpath) if enc == nil { return false } packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) return C_RTMP_SendPacket(r, &packet, 0) } // int SendPublish(RTMP* r); // rtmp.c +1908 func C_SendPublish(r *C_RTMP) (ok bool) { var packet C_RTMPPacket var pbuf [1024]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x04 /* source channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_LARGE packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = int32(r.m_stream_id) packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_publish) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) enc = C_AMF_EncodeString(enc, pend, &r.Link.playpath) if enc == nil { return false } enc = C_AMF_EncodeString(enc, pend, &av_live) if enc == nil { return false } packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) 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 packet C_RTMPPacket var pbuf [256]byte pend := (*byte)(incBytePtr(unsafe.Pointer(&pbuf[0]), 256)) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av_deleteStream) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) enc = C_AMF_EncodeNumber(enc, pend, dStreamId) packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) /* 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 packet C_RTMPPacket var pbuf [256]byte pend := (*byte)(incBytePtr(unsafe.Pointer(&pbuf[0]), 256)) packet.m_nChannel = 0x02 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_MEDIUM packet.m_packetType = RTMP_PACKET_TYPE_BYTES_READ_REPORT packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] packet.m_nBodySize = 4 C_AMF_EncodeInt32((*byte)(unsafe.Pointer(packet.m_body)), pend, int32(r.m_nBytesIn)) r.m_nBytesInSent = r.m_nBytesIn return C_RTMP_SendPacket(r, &packet, 0) } // int SendCheckBW(RTMP* r); // rtmp.c +2105 func C_SendCheckBW(r *C_RTMP) (ok bool) { var packet C_RTMPPacket var pbuf [256]byte var pend *byte = (*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&pbuf[0])) + unsafe.Sizeof(pbuf))) var enc *byte packet.m_nChannel = 0x03 /* control channel (invoke) */ packet.m_headerType = RTMP_PACKET_SIZE_LARGE packet.m_packetType = RTMP_PACKET_TYPE_INVOKE packet.m_nTimeStamp = 0 packet.m_nInfoField2 = 0 packet.m_hasAbsTimestamp = 0 packet.m_body = &pbuf[RTMP_MAX_HEADER_SIZE] enc = (*byte)(unsafe.Pointer(packet.m_body)) enc = C_AMF_EncodeString(enc, pend, &av__checkbw) r.m_numInvokes++ enc = C_AMF_EncodeNumber(enc, pend, float64(r.m_numInvokes)) *enc = AMF_NULL enc = (*byte)(incBytePtr(unsafe.Pointer(enc), 1)) packet.m_nBodySize = uint32(uintptr(unsafe.Pointer(enc)) - uintptr( unsafe.Pointer(packet.m_body))) 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(vals *C_RTMP_METHOD, num *int32, i, freeit int32) { if freeit != 0 { //C.free(unsafe.Pointer((*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(vals), int(i), //int(unsafe.Sizeof(*vals))))).name.av_val)) } (*num)-- for ; i < *num; i++ { *(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(vals), int(i), int(unsafe.Sizeof(*vals)))) = *(*C_RTMP_METHOD)(incPtr( unsafe.Pointer(vals), int(i+1), int(unsafe.Sizeof(*vals)))) } (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(vals), int(i), int(unsafe.Sizeof(*vals))))).name.av_val = nil (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(vals), int(i), int(unsafe.Sizeof(*vals))))).name.av_len = 0 (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(vals), int(i), int(unsafe.Sizeof(*vals))))).num = 0 } // void AV_queue(RTMP_METHOD** vals, int* num, C_AVal* av, int txn); // rtmp.c +2414 func C_AV_queue(vals **C_RTMP_METHOD, num *int32, av *C_AVal, txn int32) { if (*num & 0x0f) == 0 { *vals = (*C_RTMP_METHOD)(C.realloc(unsafe.Pointer(*vals), C.size_t((*num+16)* int32(unsafe.Sizeof(*(*vals)))))) } tmp := malloc(uintptr(av.av_len + 1)) //tmp := allocate(uintptr(av.av_len + 1)) memmove(tmp, unsafe.Pointer(av.av_val), uintptr(av.av_len)) (*[_Gi]byte)(tmp)[av.av_len] = '\000' (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(*vals), int(*num), int(unsafe.Sizeof(*(*vals)))))).num = int32(txn) (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(*vals), int(*num), int(unsafe.Sizeof(*(*vals)))))).name.av_len = av.av_len (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(*vals), int(*num), int(unsafe.Sizeof(*(*vals)))))).name.av_val = (*byte)(tmp) (*num)++ } // int HandleInvoke(RTMP* r, const char* body, unsigned int nBodySize); // rtmp.c +2912 func C_HandleInvoke(r *C_RTMP, body *byte, nBodySize uint32) (ok bool) { var obj C_AMFObject var method C_AVal var txn float64 var nRes int32 if *body != 0x02 { // TODO use new logger here //RTMP_Log(RTMP_LOGWARNING, "%s, Sanity failed. no string method in invoke packet", //__FUNCTION__); return false } nRes = C_AMF_Decode(&obj, body, int32(nBodySize), 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) C_AMFProp_GetString(C_AMF_GetProp(&obj, nil, 0), &method) txn = float64(C_AMFProp_GetNumber(C_AMF_GetProp(&obj, nil, 1))) // TODO use new logger here // RTMP_Log(RTMP_LOGDEBUG, "%s, server invoking <%s>", __FUNCTION__, method.av_val); switch { case C_AVMATCH(&method, &av__result) != 0: { var methodInvoked C_AVal var i int32 for i = 0; i < int32(r.m_numCalls); i++ { if float64((*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(r.m_methodCalls), int(i), int(unsafe.Sizeof(*r.m_methodCalls))))).num) == txn { methodInvoked = (*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(r.m_methodCalls), int(i), int(unsafe.Sizeof(*r.m_methodCalls))))).name C_AV_erase(r.m_methodCalls, (*int32)(&r.m_numCalls), int32(i), 0) break } } if methodInvoked.av_val == nil { // 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 { case C_AVMATCH(&methodInvoked, &av_connect) != 0: if r.Link.token.av_len != 0 { 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 C_AVMATCH(&methodInvoked, &av_createStream) != 0: r.m_stream_id = int32(C_AMFProp_GetNumber(C_AMF_GetProp(&obj, nil, 3))) if (r.Link.protocol & RTMP_FEATURE_WRITE) != 0 { C_SendPublish(r) } else { panic("Link protocol has no RTMP_FEATURE_WRITE") } case C_AVMATCH(&methodInvoked, &av_play) != 0 || C_AVMATCH(&methodInvoked, &av_publish) != 0: panic("Unsupported method av_play/av_publish") } //C.free(unsafe.Pointer(methodInvoked.av_val)) } case C_AVMATCH(&method, &av_onBWDone) != 0: if r.m_nBWCheckCounter == 0 { C_SendCheckBW(r) } case C_AVMATCH(&method, &av_onFCUnsubscribe) != 0 || C_AVMATCH(&method, &av_onFCSubscribe) != 0: panic("Unsupported method av_onFCUnsubscribe/av_onFCSubscribe") case C_AVMATCH(&method, &av_ping) != 0: panic("Unsupported method av_ping") case C_AVMATCH(&method, &av__onbwcheck) != 0: panic("Unsupported method av_onbwcheck") case C_AVMATCH(&method, &av__onbwdone) != 0: panic("Unsupported method av_onbwdone") case C_AVMATCH(&method, &av_close) != 0: panic("Unsupported method av_close") case C_AVMATCH(&method, &av_onStatus) != 0: var obj2 C_AMFObject var code, level C_AVal C_AMFProp_GetObject(C_AMF_GetProp(&obj, nil, 3), &obj2) C_AMFProp_GetString(C_AMF_GetProp(&obj2, &av_code, -1), &code) C_AMFProp_GetString(C_AMF_GetProp(&obj2, &av_level, -1), &level) // TODO use new logger // RTMP_Log(RTMP_LOGDEBUG, "%s, onStatus: %s", __FUNCTION__, code.av_val); switch { case C_AVMATCH(&code, &av_NetStream_Failed) != 0 || C_AVMATCH(&code, &av_NetStream_Play_Failed) != 0 || C_AVMATCH(&code, &av_NetStream_Play_StreamNotFound) != 0 || C_AVMATCH(&code, &av_NetConnection_Connect_InvalidApp) != 0: panic("Unsupported method av_NetStream/av_NetStream_Play_Failed/av_netSTream_Play_StreamNotFound/av_netConnection_Connect_invalidApp") case C_AVMATCH(&code, &av_NetStream_Play_Start) != 0 || C_AVMATCH(&code, &av_NetStream_Play_PublishNotify) != 0: panic("Unsupported method av_NetStream_Play_Start/av_NetStream_Play_PublishNotify") case C_AVMATCH(&code, &av_NetStream_Publish_Start) != 0: var i int32 r.m_bPlaying = true for i = 0; i < int32(r.m_numCalls); i++ { if C_AVMATCH(&(*(*C_RTMP_METHOD)(incPtr(unsafe.Pointer(r.m_methodCalls), int(i), int(unsafe.Sizeof(*r.m_methodCalls))))).name, &av_publish) != 0 { C_AV_erase(r.m_methodCalls, (*int32)(&r.m_numCalls), int32(i), 1) break } } case C_AVMATCH(&code, &av_NetStream_Play_Complete) != 0 || C_AVMATCH(&code, &av_NetStream_Play_Stop) != 0 || C_AVMATCH(&code, &av_NetStream_Play_UnpublishNotify) != 0: panic("Unsupported method av_NetStream_Play_Complete/av_NetStream_Play_Stop/av_NetStream_Play_UnpublishNotify") case C_AVMATCH(&code, &av_NetStream_Seek_Notify) != 0: panic("Unsupported method av_netStream_Seek_Notify") case C_AVMATCH(&code, &av_NetStream_Pause_Notify) != 0: panic("Unsupported method av_NetStream_Pause_Notify") } case C_AVMATCH(&method, &av_playlist_ready) != 0: panic("Unsupported method av_playlist_ready") default: } 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((*byte)(unsafe.Pointer(packet.m_body)))) // 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((*byte)(unsafe.Pointer(packet.m_body)))) // 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((*byte)(unsafe.Pointer(packet.m_body)))) //r.m_nClientBW = int32(C.AMF_DecodeInt32((*byte)(unsafe.Pointer(packet.m_body)))) if packet.m_nBodySize > 4 { r.m_nClientBW2 = (uint8)((*[_Gi]byte)(unsafe.Pointer(packet.m_body))[4]) } else { r.m_nClientBW2 = 255 } // 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 { var c *uint8 = (*uint8)(data) return int32(((*[_Gi]byte)(unsafe.Pointer(c))[3] << 24) | ((*[_Gi]byte)(unsafe.Pointer(c))[2] << 16) | ((*[_Gi]byte)(unsafe.Pointer(c))[1] << 8) | *c) } // int EncodeInt32LE(char* output, int nVal); // rtmp.c +3537 func C_EncodeInt32LE(output *byte, nVal int32) int32 { *output = byte(nVal) nVal >>= 8 (*[_Gi]byte)(unsafe.Pointer(output))[1] = byte(nVal) nVal >>= 8 (*[_Gi]byte)(unsafe.Pointer(output))[2] = byte(nVal) nVal >>= 8 (*[_Gi]byte)(unsafe.Pointer(output))[3] = byte(nVal) 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]uint8 memset((*byte)(&hbuf[0]), 0, RTMP_MAX_HEADER_SIZE) var header *byte header = (*byte)(unsafe.Pointer(&hbuf[0])) var nSize, hSize, nToRead, nChunk int32 var extendedTimestamp int32 if C_ReadN(r, (*byte)(&hbuf[0]), 1) == 0 { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet header!") return false } packet.m_headerType = uint8((hbuf[0] & 0xc0) >> 6) packet.m_nChannel = int32(hbuf[0] & 0x3f) header = (*byte)(incBytePtr(unsafe.Pointer(header), 1)) switch { case packet.m_nChannel == 0: if C_ReadN(r, (*byte)(&hbuf[1]), 1) != 1 { log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header 2nd byte.") return false } packet.m_nChannel = int32(hbuf[1]) packet.m_nChannel += 64 header = (*byte)(incBytePtr(unsafe.Pointer(header), 1)) case packet.m_nChannel == 1: var tmp int32 if C_ReadN(r, (*byte)(&hbuf[1]), 2) != 2 { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet 3rd byte") return false } tmp = int32((hbuf[2] << 8) + hbuf[1]) packet.m_nChannel = int32(tmp + 64) header = (*byte)(incBytePtr(unsafe.Pointer(header), 2)) } nSize = int32(packetSize[packet.m_headerType]) if packet.m_nChannel >= r.m_channelsAllocatedIn { var n int32 = int32(packet.m_nChannel + 10) timestamp := (*int32)(C.realloc(unsafe.Pointer(r.m_channelTimestamp), C.size_t(int32(unsafe.Sizeof(n))*n))) var packetPtr *C_RTMPPacket packets := (**C_RTMPPacket)(C.realloc(unsafe.Pointer(r.m_vecChannelsIn), C.size_t(int32(unsafe.Sizeof(packetPtr))*n))) if timestamp == nil { //C.free(unsafe.Pointer(r.m_channelTimestamp)) } if packets == nil { //C.free(unsafe.Pointer(r.m_vecChannelsIn)) } r.m_channelTimestamp = (*int32)(timestamp) r.m_vecChannelsIn = packets if timestamp == nil || packets == nil { r.m_channelsAllocatedIn = 0 return false } memset((*byte)(incPtr(unsafe.Pointer(r.m_channelTimestamp), int(r.m_channelsAllocatedIn), int(unsafe.Sizeof(*r.m_channelTimestamp)))), 0, int(4*int32((n-int32(r.m_channelsAllocatedIn))))) memset((*byte)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int( r.m_channelsAllocatedIn), int(unsafe.Sizeof(*r.m_vecChannelsIn)))), 0, int(int32(unsafe.Sizeof(*packets))*(n-int32(r.m_channelsAllocatedIn)))) r.m_channelsAllocatedIn = int32(n) } switch { case nSize == RTMP_LARGE_HEADER_SIZE: packet.m_hasAbsTimestamp = 1 case nSize < RTMP_LARGE_HEADER_SIZE: var tmpPacketPtr *C_RTMPPacket if *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPacketPtr)))) != nil { var tmpPacket C_RTMPPacket memmove(unsafe.Pointer(packet), unsafe.Pointer( *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPacketPtr))))), unsafe.Sizeof(tmpPacket)) } } nSize-- if nSize > 0 && C_ReadN(r, header, int(nSize)) != int(nSize) { log.Println("C_RTMP_ReadPacket: failed to read rtmp packet header.") return false } hSize = int32(uintptr(incBytePtr(decBytePtr(unsafe.Pointer(header), int(uintptr( unsafe.Pointer(&hbuf[0])))), int(nSize)))) if nSize >= 3 { packet.m_nTimeStamp = uint32(C_AMF_DecodeInt24(header)) if nSize >= 6 { packet.m_nBodySize = uint32(C_AMF_DecodeInt24((*byte)(incBytePtr( unsafe.Pointer(header), 3)))) packet.m_nBytesRead = 0 if nSize > 6 { packet.m_packetType = uint8((*[_Gi]byte)(unsafe.Pointer(header))[6]) if nSize == 11 { // TODO: port this packet.m_nInfoField2 = int32(C_DecodeInt32LE((*byte)(incBytePtr( unsafe.Pointer(header), 7)))) } } } } extendedTimestamp = 0 if packet.m_nTimeStamp == 0xffffff { extendedTimestamp = 1 } if extendedTimestamp != 0 { if C_ReadN(r, (*byte)(incBytePtr(unsafe.Pointer(header), int(nSize))), 4) != 4 { log.Println("RTMPRead_Packet: Failed to read extended timestamp") return false } // TODO: port this packet.m_nTimeStamp = uint32(C_AMF_DecodeInt32((*byte)(incBytePtr( unsafe.Pointer(header), int(nSize))))) hSize += 4 } if packet.m_nBodySize > 0 && packet.m_body == nil { // TODO: port this if !C_RTMPPacket_Alloc(packet, uint32(packet.m_nBodySize)) { log.Println("RTMPRead_Packet: failed to allocate packet") return false } packet.m_headerType = uint8((hbuf[0] & 0xc0) >> 6) } nToRead = int32(packet.m_nBodySize - packet.m_nBytesRead) nChunk = int32(r.m_inChunkSize) if nToRead < nChunk { nChunk = nToRead } if packet.m_chunk != nil { packet.m_chunk.c_headerSize = int32(hSize) memmove(unsafe.Pointer(&packet.m_chunk.c_header[0]), unsafe.Pointer(&hbuf[0]), uintptr(hSize)) packet.m_chunk.c_chunk = (*byte)(incBytePtr(unsafe.Pointer(packet.m_body), int(packet.m_nBytesRead))) packet.m_chunk.c_chunkSize = int32(nChunk) } if C_ReadN(r, (*byte)(incBytePtr(unsafe.Pointer(packet.m_body), int(packet.m_nBytesRead))), int(nChunk)) != int(nChunk) { log.Println("C_RTMP_ReadPacket: failed to read RTMP packet body") return false } packet.m_nBytesRead += uint32(nChunk) var tmpPktPtr *C_RTMPPacket // keep the packet as ref for other packets on this channel if *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPktPtr)))) == nil { var tmpPkt C_RTMPPacket *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPktPtr)))) = (*C_RTMPPacket)(malloc(uintptr( unsafe.Sizeof(tmpPkt)))) } memmove(unsafe.Pointer(*(**C_RTMPPacket)(incPtr(unsafe.Pointer( r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPktPtr))))), unsafe.Pointer(packet), unsafe.Sizeof(tmpPktPtr)) if extendedTimestamp != 0 { (*(**C_RTMPPacket)(incPtr(unsafe.Pointer( r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(tmpPktPtr))))).m_nTimeStamp = 0xffffff } // TODO: port this if C_RTMPPacket_IsReady(packet) != 0 { if packet.m_hasAbsTimestamp == 0 { // timestamps seem to always be relative packet.m_nTimeStamp += *(*uint32)(incPtr(unsafe.Pointer(r.m_channelTimestamp), int(packet.m_nChannel), 4)) } *(*uint32)(incPtr(unsafe.Pointer(r.m_channelTimestamp), int(packet.m_nChannel), 4)) = packet.m_nTimeStamp (*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(*r.m_vecChannelsIn))))).m_body = nil (*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(*r.m_vecChannelsIn))))).m_nBytesRead = 0 (*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), int(packet.m_nChannel), int(unsafe.Sizeof(*r.m_vecChannelsIn))))).m_hasAbsTimestamp = 0 } 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 uptime, suptime uint32 var typ byte //clientbuf := make([]byte, RTMP_SIG_SIZE+1) var clientbuf [RTMP_SIG_SIZE + 1]byte clientsig := (*byte)(incBytePtr(unsafe.Pointer(&clientbuf[0]), 1)) //serversig := make([]byte, RTMP_SIG_SIZE) var serversig [RTMP_SIG_SIZE]byte clientbuf[0] = 0x03 // not encrypted // TODO: port rtmp_getTime uptime = inet.Htonl(uint32(C_RTMP_GetTime())) memmove(unsafe.Pointer(clientsig), unsafe.Pointer(&uptime), 4) memset((*byte)(sliceToPtr((*[_Gi]byte)(unsafe.Pointer(clientsig))[4:])), 0, 4) for i := 8; i < RTMP_SIG_SIZE; i++ { (*[_Gi]byte)(unsafe.Pointer(clientsig))[i] = byte(rand.Intn(256)) } if !C_WriteN(r, unsafe.Pointer(&clientbuf[0]), RTMP_SIG_SIZE+1) { return false } if C_ReadN(r, (*byte)(unsafe.Pointer(&typ)), 1) != 1 { return false } if debugMode { log.Printf("C_HandShake: Type answer: %v\n", typ) } if typ != clientbuf[0] { log.Printf("C_HandShake: type mismatch: client sent %v, server sent: %v\n", clientbuf[0], typ) } if C_ReadN(r, (*byte)(unsafe.Pointer(&serversig[0])), RTMP_SIG_SIZE) != RTMP_SIG_SIZE { return false } // decode server response memmove(unsafe.Pointer(&suptime), unsafe.Pointer(&serversig[0]), 4) suptime = inet.Ntohl(suptime) // 2nd part of handshake if !C_WriteN(r, unsafe.Pointer(&serversig[0]), RTMP_SIG_SIZE) { return false } if C_ReadN(r, (*byte)(unsafe.Pointer(&serversig[0])), RTMP_SIG_SIZE) != 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 last := 0 var nSize, hSize, cSize, nChunkSize int var header, hptr, hend, buffer, tbuf, toff unsafe.Pointer var goHbuf [RTMP_MAX_HEADER_SIZE]byte var hbuf = unsafe.Pointer(&goHbuf[0]) var c byte var t uint32 var packets unsafe.Pointer if packet.m_nChannel >= r.m_channelsAllocatedOut { n := int(packet.m_nChannel + 10) packets = C.realloc(unsafe.Pointer(r.m_vecChannelsOut), C.size_t( unsafe.Sizeof(packet)*uintptr(n))) if uintptr(packets) == uintptr(0) { //C.free(unsafe.Pointer(r.m_vecChannelsOut)) r.m_vecChannelsOut = nil r.m_channelsAllocatedOut = 0 return false } r.m_vecChannelsOut = (**C_RTMPPacket)(packets) memset((*byte)(incPtr(unsafe.Pointer(r.m_vecChannelsOut), int(r.m_channelsAllocatedOut), int(unsafe.Sizeof(packet)))), 0, int(unsafe.Sizeof(packet)* uintptr(n-int(r.m_channelsAllocatedOut)))) r.m_channelsAllocatedOut = int32(n) } prevPacket = *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsOut), int(packet.m_nChannel), int(unsafe.Sizeof(packet)))) 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) if packet.m_body != nil { header = decBytePtr(unsafe.Pointer(packet.m_body), nSize) hend = unsafe.Pointer(packet.m_body) } else { header = incBytePtr(hbuf, 6) // TODO: be cautious about this sizeof - make sure it works how you think it // does. C code used sizeof(hbuf) where hbuf is a *char hend = incBytePtr(hbuf, 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 = byte(packet.m_headerType) << 6 switch cSize { case 0: c |= byte(packet.m_nChannel) case 1: case 2: c |= byte(1) } *(*byte)(hptr) = c hptr = incBytePtr(hptr, 1) if cSize != 0 { tmp := packet.m_nChannel - 64 *(*byte)(hptr) = byte(tmp & 0xff) hptr = incBytePtr(hptr, 1) if cSize == 2 { *(*byte)(hptr) = byte(tmp >> 8) hptr = incBytePtr(hptr, 1) } } if nSize > 1 { res := t if t > 0xffffff { res = 0xffffff } hptr = unsafe.Pointer(C_AMF_EncodeInt24((*byte)(hptr), (*byte)(hend), int32(res))) } if nSize > 4 { hptr = unsafe.Pointer(C_AMF_EncodeInt24((*byte)(hptr), (*byte)(hend), (int32(packet.m_nBodySize)))) *(*byte)(hptr) = byte(packet.m_packetType) hptr = incBytePtr(hptr, 1) } if nSize > 8 { // TODO: port this hptr = incBytePtr(hptr, int(C_EncodeInt32LE((*byte)(hptr), int32(packet.m_nInfoField2)))) } if t >= 0xffffff { hptr = unsafe.Pointer(C_AMF_EncodeInt32((*byte)(hptr), (*byte)(hend), (int32)(t))) } nSize = int(packet.m_nBodySize) buffer = unsafe.Pointer(packet.m_body) nChunkSize = int(r.m_outChunkSize) if debugMode { log.Printf("C_RTMP_SendPacket: fd=%v, size=%v", r.m_sb.sb_socket, nSize) } for (nSize + hSize) != 0 { if nSize < nChunkSize { nChunkSize = nSize } if tbuf != nil { copy((*[_Gi]byte)(toff)[:nChunkSize+hSize], (*[_Gi]byte)(header)[:nChunkSize+hSize]) toff = incBytePtr(toff, nChunkSize+hSize) } else { // TODO: port this if !C_WriteN(r, 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 } *(*byte)(header) = byte(0xc0 | c) if cSize != 0 { tmp := int(packet.m_nChannel) - 64 (*[_Gi]byte)(header)[1] = byte(tmp) if cSize == 2 { (*[_Gi]byte)(header)[2] = byte(tmp >> 8) } } if t >= 0xffffff { extendedTimestamp := incBytePtr(header, 1+cSize) C_AMF_EncodeInt32((*byte)(extendedTimestamp), (*byte)(incBytePtr(extendedTimestamp, 4)), (int32)(t)) } } } if tbuf != nil { ok := C_WriteN(r, tbuf, int(uintptr(decBytePtr(toff, int(uintptr(unsafe.Pointer(tbuf))))))) ////C.free(tbuf) tbuf = nil if !ok { return false } } // We invoked a remote method // TODO: port the const if packet.m_packetType == RTMP_PACKET_TYPE_INVOKE { // TODO: port C_AVal var method C_AVal var ptr unsafe.Pointer ptr = incBytePtr(unsafe.Pointer(packet.m_body), 1) C_AMF_DecodeString((*byte)(ptr), &method) if debugMode { log.Printf("Invoking %v", method.av_val) } // keep it in call queue till result arrives if queue != 0 { var txn int ptr = incBytePtr(ptr, 3+int(method.av_len)) txn = int(C_AMF_DecodeNumber((*byte)(ptr))) C_AV_queue(&r.m_methodCalls, (*int32)(unsafe.Pointer(&r.m_numCalls)), &method, int32(txn)) } } if *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsOut), int(packet.m_nChannel), int(unsafe.Sizeof(packet)))) == nil { *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsOut), int(packet.m_nChannel), int(unsafe.Sizeof(packet)))) = (*C_RTMPPacket)(malloc(uintptr(unsafe.Sizeof(*packet)))) } memmove(unsafe.Pointer(*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsOut), int(packet.m_nChannel), int(unsafe.Sizeof(packet))))), unsafe.Pointer(packet), uintptr(unsafe.Sizeof(*packet))) return true } // void RTMP_Close(RTMP *r); // rtmp.c +4168 func C_RTMP_Close(r *C_RTMP) { C_CloseInternal(r, 0) } // static void CloseInternal(RTMP *r, int reconnect); // rtmp.c +4175 func C_CloseInternal(r *C_RTMP, reconnect int32) { var i int32 if C_RTMP_IsConnected(r) != 0 { if r.m_stream_id > 0 { i = int32(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 if r.m_read.flags&RTMP_READ_HEADER != 0 { //C.free(unsafe.Pointer(r.m_read.buf)) r.m_read.buf = nil } r.m_read.dataType = 0 r.m_read.flags = 0 r.m_read.status = 0 r.m_read.nResumeTS = 0 r.m_read.nIgnoredFrameCounter = 0 r.m_read.nIgnoredFlvFrameCounter = 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 *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), i, int(unsafe.Sizeof(&r.m_write)))) != nil { //C.RTMPPacket_Free(*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), i, //int(unsafe.Sizeof(&r.m_write))))) //C.free(unsafe.Pointer(*(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), //i, int(unsafe.Sizeof(&r.m_write)))))) *(**C_RTMPPacket)(incPtr(unsafe.Pointer(r.m_vecChannelsIn), i, int(unsafe.Sizeof(&r.m_write)))) = 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 = nil r.m_numCalls = 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 == 0) { //C.free(unsafe.Pointer(r.Link.app.av_val)) r.Link.app.av_val = nil r.Link.lFlags ^= RTMP_LF_FAPU } if reconnect == 0 { //C.free(unsafe.Pointer(r.Link.playpath0.av_val)) r.Link.playpath0.av_val = nil } } // int RTMPSockBuf_Fill(RTMPSockBuf* sb); // rtmp.c +4253 func C_RTMPSockBuf_Fill(sb *C_RTMPSockBuf) int { var nBytes int if sb.sb_size == 0 { sb.sb_start = &sb.sb_buf[0] } for { nBytes = int(unsafe.Sizeof(sb.sb_buf)) - 1 - int(sb.sb_size) - int(uintptr(unsafe.Pointer(sb.sb_start))-uintptr(unsafe.Pointer( &sb.sb_buf[0]))) // TODO: figure out what to do with recv nBytes = int(C.recv(C.int(sb.sb_socket), unsafe.Pointer(uintptr(unsafe.Pointer( sb.sb_start))+uintptr(int(sb.sb_size))), C.size_t(nBytes), 0)) if nBytes != -1 { sb.sb_size += int32(nBytes) } else { log.Println("C_RTMPSockBuf_Fill: recv error!") } break } return nBytes } // int RTMPSockBuf_Send(RTMPSockBuf* sb, const char* buf, int len); // rtmp.c +4297 // TODO replace send with golang net connection send func C_RTMPSockBuf_Send(sb *C_RTMPSockBuf, buf *byte, l int32) int32 { return int32(C.send(C.int(sb.sb_socket), unsafe.Pointer(buf), C.size_t(l), 0)) } // int // RTMPSockBuf_Close(RTMPSockBuf *sb) // rtmp.c +4369 func C_RTMPSockBuf_Close(sb *C_RTMPSockBuf) int32 { if sb.sb_socket != -1 { return int32(C.close(C.int(sb.sb_socket))) } return 0 } // 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 pend, enc []byte size := len(buf) var num int pkt.m_nChannel = 0x04 pkt.m_nInfoField2 = int32(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 = uint8(buf[0]) buf = buf[1:] pkt.m_nBodySize = uint32(C_AMF_DecodeInt24(&buf[0])) buf = buf[3:] pkt.m_nTimeStamp = uint32(C_AMF_DecodeInt24(&buf[0])) 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, uint32(pkt.m_nBodySize)) { log.Println("Failed to allocate packet") return 0 } enc = (*[_Gi]byte)(unsafe.Pointer(pkt.m_body))[:pkt.m_nBodySize] pend = enc[pkt.m_nBodySize:] if pkt.m_packetType == RTMP_PACKET_TYPE_INFO { enc = (*[_Gi]byte)(unsafe.Pointer(C_AMF_EncodeString((*byte)(unsafe.Pointer(&enc[0])), (*byte)(unsafe.Pointer(&pend[0])), &setDataFrame)))[:pkt.m_nBodySize] // TODO: work out what to do with this pkt.m_nBytesRead = uint32(float64(uintptr(unsafe.Pointer(&enc[0])) - uintptr(unsafe.Pointer(pkt.m_body)))) } } else { enc = ((*[_Gi]byte)(unsafe.Pointer(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 } func memmove(to, from unsafe.Pointer, n uintptr) { if to != nil && from != nil && n != 0 { copy((*[_Gi]byte)(to)[:n], (*[_Gi]byte)(from)[:n]) } } // TODO: write test for this func func memcmp(a, b unsafe.Pointer, size int) int { for i := 0; i < size; i++ { aValue := (*[_Gi]byte)(a)[i] bValue := (*[_Gi]byte)(b)[i] if aValue != bValue { if aValue < bValue { return -1 } else { return 1 } } } return 0 } func memset(ptr *byte, val byte, num int) { for i := 0; i < num; i++ { (*[_Gi]byte)(unsafe.Pointer(ptr))[i] = val } } func strLen(str string) int { return len(str) } // wrapper for converting byte pointer to unsafe.Pointer func bToUP(b *byte) unsafe.Pointer { return unsafe.Pointer(b) } // Creates a new C style string from a go string func goStrToCStr(str string) *byte { l := len(str) slice := make([]byte, l+1) ptr := unsafe.Pointer(&[]byte(str)[0]) for i := 0; i < l; i++ { slice[i] = (*[_Gi]byte)(ptr)[i] } slice[l] = '\x00' return &slice[0] } // TODO: need a test in rtmp_test.go func cStrToGoStr(cStr *byte) string { return string((*[_Gi]byte)(unsafe.Pointer(cStr))[:strlen(cStr)]) } // Duplicates a string given as a byte pointer func strdup(str *byte) *byte { length := strlen(str) newMem := make([]byte, length+1) oldMem := (*[_Gi]byte)(unsafe.Pointer(str))[:length+1] copy(newMem, oldMem) return &newMem[0] } // Gets the length of the string found at str - length is number of chars // between start and terminating null char. Returns -1 if a null char is not // found before a count of 1000 func strlen(str *byte) int32 { for i := 0; i < 1000; i++ { if (*[_Gi]byte)(unsafe.Pointer(str))[i] == '\000' { return int32(i) } } return int32(-1) } // Returns the pointer where the first occurance of val is located in a string // which is terminated by a null char. Returns nil if null char is not found // before a count of 10000 func strchr(str *byte, val byte) *byte { for i := 0; i < 1000; i++ { if (*[_Gi]byte)(unsafe.Pointer(str))[i] == val { return &(*[_Gi]byte)(unsafe.Pointer(str))[i] } if (*[_Gi]byte)(unsafe.Pointer(str))[i] == '\000' { break } } return nil } // TODO: need a test in rtmp_test.go // Porting: http://www.ai.mit.edu/projects/im/cam8/cam8/working/CAMlib/tcl/compat/strstr.c func strstr(str *byte, substring *byte) *byte { var a, b *byte /* First scan quickly through the two strings looking for a * single-character match. When it's found, then compare the * rest of the substring. */ b = substring if *b == 0 { return str } for *str != 0 { str = (*byte)(incBytePtr(unsafe.Pointer(str), 1)) if *str != *b { continue } a = str for { if *b == 0 { return str } tmp1 := a a = (*byte)(incBytePtr(unsafe.Pointer(a), 1)) tmp2 := b b = (*byte)(incBytePtr(unsafe.Pointer(b), 1)) if *tmp1 != *tmp2 { break } } b = substring } return nil } // Creates mem of the size noOfBytes. returns as unsafe pointer func malloc(nOfBytes uintptr) unsafe.Pointer { mem := make([]byte, int(nOfBytes)) return unsafe.Pointer(&mem[0]) } func calloc(val byte, noOfBytes uintptr) unsafe.Pointer { mem := malloc(noOfBytes) memset((*byte)(mem), val, int(noOfBytes)) return mem } // incBytePtr returns an unsafe.Pointer to a byte that is inc positive positions // from the passed ptr func incBytePtr(ptr unsafe.Pointer, inc int) unsafe.Pointer { return incPtr(ptr, inc, 1) } // incPtr attempts to replicate C like pointer arithmatic functionality func incPtr(ptr unsafe.Pointer, inc, typeSize int) unsafe.Pointer { return unsafe.Pointer(uintptr(ptr) + uintptr(inc*typeSize)) } // incPtr attempts to replicate C like pointer arithmatic functionality func decPtr(ptr unsafe.Pointer, dec, typeSize int) unsafe.Pointer { return unsafe.Pointer(uintptr(ptr) - uintptr(dec*typeSize)) } // decBytePtr returns an unsafe.Pointer to a byte that is dec negative positions // from ptr func decBytePtr(ptr unsafe.Pointer, dec int) unsafe.Pointer { return decPtr(ptr, dec, 1) } // sliceToPtr get's the address of the first data element and returns as unsafe // pointer func sliceToPtr(data []byte) unsafe.Pointer { if len(data) == 0 { return nil } return unsafe.Pointer(&data[0]) } 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)) }