/* 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 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" "strconv" "unsafe" "log" "bitbucket.org/ausocean/av/tools" ) 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 } func byteSliceToCArr(buf []byte) *C.char { return (*C.char)(unsafe.Pointer(&buf[0])) } func (s *session) rtmpWrite(r *C.RTMP, buf []byte) int { var pkt *C.RTMPPacket = &r.m_write var pend, enc []byte size := len(buf) s2 := size var ret, num int pkt.m_nChannel = 0x04 pkt.m_nInfoField2 = r.m_stream_id for s2 > 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 += 13 buf = buf[13:] s2 -= 13 } buf = buf[1:] pkt.m_packetType = C.uchar(buf[0]) pkt.m_nBodySize = C.AMF_DecodeInt24(byteSliceToCArr(buf)) buf = buf[3:] pkt.m_nTimeStamp = C.AMF_DecodeInt24(byteSliceToCArr(buf)) buf = buf[3:] buf = buf[1:] pkt.m_nTimeStamp |= C.uint( buf[0] << 24 ) buf = buf[3:] s2 -= 11 if ((pkt.m_packetType == C.RTMP_PACKET_TYPE_AUDIO || pkt.m_packetType == C.RTMP_PACKET_TYPE_VIDEO) && pkt.m_nTimeStamp == 0) || pkt.m_packetType == C.RTMP_PACKET_TYPE_INFO { pkt.m_headerType = C.RTMP_PACKET_SIZE_LARGE if pkt.m_packetType == C.RTMP_PACKET_TYPE_INFO { pkt.m_nBodySize += 16 } } else { pkt.m_headerType = C.RTMP_PACKET_SIZE_MEDIUM } // C: if (!RTMPPacket_Alloc(pkt, pkt->m_nBodySize)) if int( C.RTMPPacket_Alloc(pkt, pkt.m_nBodySize)) == 0 { log.Println("Failed to allocated packet") return 0 } // Note this is pointer arithmatic enc = pkt.m_body // pend = enc + pkt.m_nBodySize pend = (*C.char)(unsafe.Pointer( uintptr(unsafe.Pointer(enc)) + ( unsafe.Sizeof(C.char(0x00)) * (uintptr(pkt.m_nBodySize)) ) )) if pkt.m_packetType == C.RTMP_PACKET_TYPE_INFO { enc = C.AMF_EncodeString(enc, pend, &av_setDataFrame ) // C-code: pkt.m_nBytesRead enc - pkt.m_body pkt.m_nBytesRead = (C.uint)((uintptr(unsafe.Pointer(enc)) - uintptr(unsafe.Pointer(pkt.m_body)))/unsafe.Sizeof(C.char(0x00))) } } else { //C-code: enc = pkt.m_body + pkt.m_nBytesRead enc = (*C.char)(unsafe.Pointer( uintptr(unsafe.Pointer(pkt.m_body)) + ( unsafe.Sizeof(C.char(0x00)) * (uintptr(pkt.m_nBytesRead)) ) )) } num = int(pkt.m_nBodySize - pkt.m_nBytesRead) if num > s2 { num = s2 } // C.memcpy(enc, buf, num) // Need to create byte slice from c array enc encSlice := C.GoBytes(unsafe.Pointer(enc),C.int(num)) copy(encSlice,buf) pkt.m_nBytesRead += C.uint(num) s2 -= num //buf += num buf = buf[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 !tools.IntToBool(ret) { return -1 } // buf += 4 buf = buf[4:] s2 -= 4 if s2 < 0 { break } } } return size+s2 } func (s *session) writeFrame(data []byte) uint { if C.RTMP_IsConnected(s.rtmp) <= 0 { return 1 } // This is where C.RTMP_Write would be used if s.rtmpWrite(s.rtmp, data) <= 0 { return 2 } return 0 } // 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) } ret := s.writeFrame(data) if ret != 0 { return 0, Err(ret) } return len(data), nil } // 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)) }