diff --git a/revid/cmd/h264-file-to-flv-rtmp/main b/revid/cmd/h264-file-to-flv-rtmp/main new file mode 100755 index 00000000..b9e7fd78 Binary files /dev/null and b/revid/cmd/h264-file-to-flv-rtmp/main differ diff --git a/rtmp/rtmp.go b/rtmp/rtmp.go index 7a885bd5..c5534eb9 100644 --- a/rtmp/rtmp.go +++ b/rtmp/rtmp.go @@ -30,9 +30,10 @@ package rtmp /* #cgo CFLAGS: -I/usr/local/include/librtmp -#cgo LDFLAGS: -lrtmp -lz -Wl,-rpath=/usr/local/lib +#cgo LDFLAGS: -lrtmp -Wl,-rpath=/usr/local/lib #include +#include #include RTMP* start_session(RTMP* rtmp, char* url, uint connect_timeout); @@ -43,8 +44,15 @@ import "C" import ( "errors" + "log" "strconv" "unsafe" + _"fmt" + "math" +) + +const ( + minDataSize = 11 ) // Session provides an interface for sending flv tags over rtmp. @@ -56,8 +64,7 @@ type Session interface { // session provides parameters required for an rtmp communication session. type session struct { - rtmp *C.RTMP - + rtmp *C.RTMP url string timeout uint } @@ -72,6 +79,18 @@ func NewSession(url string, connectTimeout uint) Session { } } +// 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 setDataFrame = AVC("@setDataFrame") + // Open establishes an rtmp connection with the url passed into the // constructor func (s *session) Open() error { @@ -91,13 +110,128 @@ func (s *session) Write(data []byte) (int, error) { if s.rtmp == nil { return 0, Err(3) } - ret := C.write_frame(s.rtmp, (*C.char)(unsafe.Pointer(&data[0])), C.uint(len(data))) - if ret != 0 { - return 0, Err(ret) + 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, &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 + } + for i := 0; i < num; i++ { + *(*byte)(unsafe.Pointer(enc)) = *(*byte)(unsafe.Pointer(buf)) + } + //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 { @@ -128,3 +262,7 @@ func (e Err) Error() string { } return "rtmp: " + strconv.Itoa(int(e)) } + +func byteSliceToCArr(buf []byte) *C.char { + return (*C.char)(unsafe.Pointer(&buf[0])) +}