/* NAME rtmp.go DESCRIPTION See Readme.md AUTHOR 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. */ package rtmp /* #cgo CFLAGS: -I/usr/local/include/librtmp #cgo LDFLAGS: -lrtmp -Wl,-rpath=/usr/local/lib #include #include #include RTMP* start_session(RTMP* rtmp, char* url, uint connect_timeout); int write_frame(RTMP* rtmp, char* data, uint data_length); int end_session(RTMP* rtmp); */ import "C" import ( "errors" "log" "strconv" "unsafe" _"fmt" "math" ) const ( minDataSize = 11 ) // Session provides an interface for sending flv tags over rtmp. type Session interface { Open() error Write([]byte) (int, error) Close() error } // session provides parameters required for an rtmp communication session. type session struct { rtmp *C.RTMP url string timeout uint } var _ Session = (*session)(nil) // NewSession returns a new session. func NewSession(url string, connectTimeout uint) Session { return &session{ url: url, timeout: connectTimeout, } } // C.AVal is in amf.h // See #define AVC(str) {str, sizeof(str)-1} in amf.h func AVC(str string) C.AVal { var aval C.AVal aval.av_val = C.CString(str) aval.av_len = C.int(len(str)) return aval } // av_setDataFrame is a static const global in rtmp.c var av_setDataFrame = AVC("@setDataFrame") // Open establishes an rtmp connection with the url passed into the // constructor func (s *session) Open() error { if s.rtmp != nil { return errors.New("rtmp: attempt to start already running session") } var err error s.rtmp, err = C.start_session(s.rtmp, C.CString(s.url), C.uint(s.timeout)) if s.rtmp == nil { return err } return nil } // Write writes a frame (flv tag) to the rtmp connection func (s *session) Write(data []byte) (int, error) { if s.rtmp == nil { return 0, Err(3) } if C.RTMP_IsConnected(s.rtmp) <= 0 { return 0, Err(1) } //if C.RTMP_Write(s.rtmp,(*C.char)(unsafe.Pointer(&data[0])),C.int(len(data))) <= 0 { if rtmpWrite(s.rtmp, data) <= 0 { return 0, Err(2) } return len(data), nil } const ( RTMP_PACKET_SIZE_LARGE = 0 RTMP_PACKET_SIZE_MEDIUM = 1 RTMP_PACKET_SIZE_SMALL = 2 RTMP_PACKET_TYPE_INFO = 0x12 RTMP_PACKET_TYPE_AUDIO = 0x08 RTMP_PACKET_TYPE_VIDEO = 0x09 ) func rtmpWrite(r *C.RTMP, data []byte) int { buf := (*C.char)(unsafe.Pointer(&data[0])) var pkt = &r.m_write var pend, enc *C.char size := len(data) s2 := size var ret, num int pkt.m_nChannel = 0x04 pkt.m_nInfoField2 = r.m_stream_id for s2 > 0 || s2 < 0{ if pkt.m_nBytesRead == 0 { if size < minDataSize { log.Printf("size: %d\n", size) log.Printf("too small \n") return 0 } buf0 := *(*byte)(unsafe.Pointer(buf)) buf1 := *(*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(1))) buf2 := *(*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(2))) if buf0 == 'F' && buf1 == 'L' && buf2 == 'V' { buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(13))) s2 -= 13 } pkt.m_packetType = C.uchar(*(*byte)(unsafe.Pointer(buf))) buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(1))) pkt.m_nBodySize = C.AMF_DecodeInt24(buf) buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(3))) pkt.m_nTimeStamp = C.AMF_DecodeInt24(buf) buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(3))) pkt.m_nTimeStamp |= C.uint(*((*byte)(unsafe.Pointer(buf)))) << 24 buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(4))) s2 -= 11 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 } if int(C.RTMPPacket_Alloc(pkt, pkt.m_nBodySize)) == 0 { log.Println("Failed to allocate packet") return 0 } enc = pkt.m_body pend = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(enc)) + uintptr(pkt.m_nBodySize))) if pkt.m_packetType == RTMP_PACKET_TYPE_INFO { enc = C.AMF_EncodeString(enc, pend, &av_setDataFrame) pkt.m_nBytesRead = C.uint(math.Abs(float64(uintptr(unsafe.Pointer(enc)) - uintptr(unsafe.Pointer(pkt.m_body))))) } } else { enc = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(pkt.m_body)) + uintptr(pkt.m_nBytesRead))) } num = int(pkt.m_nBodySize - pkt.m_nBytesRead) if num > s2 { num = s2 } C.memcpy(unsafe.Pointer(enc),unsafe.Pointer(buf),C.ulong(num)) pkt.m_nBytesRead += C.uint(num) s2 -= num buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(num))) if pkt.m_nBytesRead == pkt.m_nBodySize { ret = int(C.RTMP_SendPacket(r, pkt, 0)) C.RTMPPacket_Free(pkt) pkt.m_nBytesRead = 0 if ret == 0 { return -1 } buf = (*C.char)(unsafe.Pointer(uintptr(unsafe.Pointer(buf)) + uintptr(4))) s2 -= 4 if s2 < 0 { break } } } return size + s2 } /* func sendPacket(r *C.RTMP, pkt *C.RTMPPacket, queue int) int { const prevPacket *C.RTMPPacket last := 0 var nSize, hSize, cSize int var } */ // Close terminates the rtmp connection func (s *session) Close() error { if s.rtmp == nil { return Err(3) } ret := C.end_session(s.rtmp) s.rtmp = nil if ret != 0 { return Err(ret) } return nil } var rtmpErrs = [...]string{ 1: "rtmp: not connected", 2: "rtmp: write error", 3: "rtmp: not started", } type Err uint func (e Err) Error() string { if 0 <= int(e) && int(e) < len(rtmpErrs) { s := rtmpErrs[e] if s != "" { return s } } return "rtmp: " + strconv.Itoa(int(e)) } func byteSliceToCArr(buf []byte) *C.char { return (*C.char)(unsafe.Pointer(&buf[0])) }