RTMP handshake fix, ReadFull now only reads upto C0C1 instead of waiting on full sequence

rtmp/chunk.go

@@ -1,6 +1,7 @@
 package rtmp
 
 import (
+	"io"
 	"net"
 	"time"
 	"encoding/binary"
@@ -42,18 +43,18 @@ type ChunkBuffers struct {
 // reads the initial variable size header that defines the chunk's format and chunkstream id
 // 5.3.1.1
 func read_basic_header(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (format uint8, csid uint32, err error) {
-	if _, err = conn.Read(chunk_bufs_ptr.fmt_csid_byte); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.fmt_csid_byte); err != nil {
 		return
 	}
 	format = uint8(chunk_bufs_ptr.fmt_csid_byte[0] >> 6) // get first 2 bits for format 0-3
 	switch chunk_bufs_ptr.fmt_csid_byte[0] & 0x3f { // last 6 bits 0-63
 	case 0: // csid 0 is invalid, means true csid is the next byte, + 64 for the 6 bits prior
-		if _, err = conn.Read(chunk_bufs_ptr.csid_true[1:]); err != nil {
+		if _, err = io.ReadFull(conn, chunk_bufs_ptr.csid_true[1:]); err != nil {
 			return
 		}
 		csid = uint32(chunk_bufs_ptr.csid_true[1]) + 64
 	case 1: // csid 1 is invalid, means true csid is in the next 2 bytes, reverse order (little endian) and the 64, reconstruct
-		if _, err = conn.Read(chunk_bufs_ptr.csid_true); err != nil {
+		if _, err = io.ReadFull(conn, chunk_bufs_ptr.csid_true); err != nil {
 			return
 		}
 		csid = uint32(binary.LittleEndian.Uint16(chunk_bufs_ptr.csid_true)) + 64
@@ -65,7 +66,7 @@ func read_basic_header(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (format uint
 
 func read_time(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (time uint32, extended_time bool, err error) {
 	chunk_bufs_ptr.time[0] = 0
-	if _, err = conn.Read(chunk_bufs_ptr.time[1:]); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.time[1:]); err != nil {
 		return
 	}
 	time = binary.BigEndian.Uint32(chunk_bufs_ptr.time)
@@ -76,7 +77,7 @@ func read_time(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (time uint32, extend
 }
 
 func read_msg_len(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_len uint32, err error) {
-	if _, err = conn.Read(chunk_bufs_ptr.msg_len[1:]); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.msg_len[1:]); err != nil {
 		return
 	}
 	msg_len = binary.BigEndian.Uint32(chunk_bufs_ptr.msg_len)
@@ -84,7 +85,7 @@ func read_msg_len(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_len uint32,
 }
 
 func read_msg_typeid(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_type uint8, err error) {
-	if _, err = conn.Read(chunk_bufs_ptr.msg_typeid); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.msg_typeid); err != nil {
 		return
 	}
 	msg_type = chunk_bufs_ptr.msg_typeid[0]
@@ -92,7 +93,7 @@ func read_msg_typeid(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_type uint
 }
 
 func read_msg_streamid(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_streamid uint32, err error) {
-	if _, err = conn.Read(chunk_bufs_ptr.msg_streamid); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.msg_streamid); err != nil {
 		return
 	}
 	msg_streamid = binary.LittleEndian.Uint32(chunk_bufs_ptr.msg_streamid)
@@ -100,7 +101,7 @@ func read_msg_streamid(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (msg_streami
 }
 
 func read_time_extd(conn net.Conn, chunk_bufs_ptr *ChunkBuffers) (time uint32, err error) {
-	if _, err = conn.Read(chunk_bufs_ptr.time); err != nil {
+	if _, err = io.ReadFull(conn, chunk_bufs_ptr.time); err != nil {
 		return
 	}
 	time = binary.BigEndian.Uint32(chunk_bufs_ptr.time)
@@ -225,7 +226,7 @@ func read_chunk_data(conn net.Conn, msg_ptr *Message, chnk_size uint32) (uint32,
 	} else {
 		buffer_end = chnk_size + msg_ptr.curr_bytes_read
 	}
-	n, err := conn.Read(msg_ptr.data[msg_ptr.curr_bytes_read:buffer_end])
+	n, err := io.ReadFull(conn, msg_ptr.data[msg_ptr.curr_bytes_read:buffer_end])
 	if err != nil {
 		return 0, err
 	}

rtmp/handshake.go

@@ -1,6 +1,7 @@
 package rtmp
 
 import (
+	"io"
 	"net"
 	"time"
 	"encoding/binary"
@@ -17,7 +18,7 @@ func DoHandshake(conn net.Conn) (hs_success bool) {
 	// force handshake to finish in under 15 seconds (aribtrary) or throw an error
 	conn.SetDeadline(time.Now().Add(15 * time.Second))
 
-	if _, err := conn.Read(C0C1C2); err != nil || C0C1C2[0] != 3 {
+	if _, err := io.ReadFull(conn, C0C1C2[:1+1536]); err != nil || C0C1C2[0] != 3 {
 		return
 	}
 	copy(C0C1C2[1:1536], S0S1S2[1+1536:])
@@ -26,7 +27,7 @@ func DoHandshake(conn net.Conn) (hs_success bool) {
 	if _, err := conn.Write(S0S1S2); err != nil { // specs say only send S0S1 and wait for C2 before sending S2, obs doesn't care apparently
 		return
 	}
-	if _, err := conn.Read(C0C1C2[1+1536:]); err != nil {
+	if _, err := io.ReadFull(conn, C0C1C2[1+1536:]); err != nil {
 		return
 	}
 	hs_success = true

transcoder/transcoder.go

@@ -8,26 +8,11 @@ import (
 
 func NewTranscoder(stream_key string) (io.WriteCloser, error) {
 	base_dir, _ := os.UserHomeDir()
-	command := "ffmpeg"
+	command := "hls-transcoder-2"
 	command_args := []string{
-		"-probesize", "500",
+		"4",
-		"-i", "pipe:0",
+		"/vid/" + stream_key + "/",
-		"-c:a", "libopus",
+		"4",
-		"-c:v", "vp9", 
-			"-quality", "realtime", 
-			"-speed", "7", 
-			"-tile-columns", "4", 
-			"-frame-parallel", "1", 
-			"-threads", "8",
-		"-b:v", "1M",
-		"-g", "30",
-		"-hls_time", "6",
-		"-hls_list_size", "4",
-		"-hls_base_url", "/vid/" + stream_key + "/",
-		"-hls_segment_type", "fmp4",
-		"-hls_flags", "delete_segments",
-		"-hls_flags", "+program_date_time",
-		"stream.m3u8",
 	}
 	transcoder := exec.Command(command, command_args...)
 	transcoder.Dir = base_dir + "/live/" + stream_key + "/"