RTMP server enhancement to support OpenIPC cameras

pkg/flv/producer.go

@@ -140,23 +140,29 @@ func (c *Producer) probe() error {
 	// 1. Empty video/audio flag
 	// 2. MedaData without stereo key for AAC
 	// 3. Audio header after Video keyframe tag
-	waitType := []byte{TagData}
-	timeout := time.Now().Add(core.ProbeTimeout)
 
-	for len(waitType) != 0 && time.Now().Before(timeout) {
+	// OpenIPC camera sends:
+	// 1. Empty video/audio flag
+	// 2. No MetaData packet
+	// 3. Sends a video packet in more than 3 seconds
+	waitVideo := true
+	waitAudio := true
+	timeout := time.Now().Add(time.Second * 5)
+
+	for (waitVideo || waitAudio) && time.Now().Before(timeout) {
 		pkt, err := c.readPacket()
 		if err != nil {
 			return err
 		}
 
-		if i := bytes.IndexByte(waitType, pkt.PayloadType); i < 0 {
+		//log.Printf("%d %0.20s", pkt.PayloadType, pkt.Payload)
-			continue
-		} else {
-			waitType = append(waitType[:i], waitType[i+1:]...)
-		}
 
 		switch pkt.PayloadType {
 		case TagAudio:
+			if !waitAudio {
+				continue
+			}
+
 			_ = pkt.Payload[1] // bounds
 
 			codecID := pkt.Payload[0] >> 4 // SoundFormat
@@ -179,8 +185,13 @@ func (c *Producer) probe() error {
 				Codecs:    []*core.Codec{codec},
 			}
 			c.Medias = append(c.Medias, media)
+			waitAudio = false
 
 		case TagVideo:
+			if !waitVideo {
+				continue
+			}
+
 			var codec *core.Codec
 
 			if isExHeader(pkt.Payload) {
@@ -213,19 +224,20 @@ func (c *Producer) probe() error {
 				Codecs:    []*core.Codec{codec},
 			}
 			c.Medias = append(c.Medias, media)
+			waitVideo = false
 
 		case TagData:
 			if !bytes.Contains(pkt.Payload, []byte("onMetaData")) {
-				waitType = append(waitType, TagData)
+				continue
 			}
 			// Dahua cameras doesn't send videocodecid
-			if bytes.Contains(pkt.Payload, []byte("videocodecid")) ||
+			if !bytes.Contains(pkt.Payload, []byte("videocodecid")) &&
-				bytes.Contains(pkt.Payload, []byte("width")) ||
+				!bytes.Contains(pkt.Payload, []byte("width")) &&
-				bytes.Contains(pkt.Payload, []byte("framerate")) {
+				!bytes.Contains(pkt.Payload, []byte("framerate")) {
-				waitType = append(waitType, TagVideo)
+				waitVideo = false
 			}
-			if bytes.Contains(pkt.Payload, []byte("audiocodecid")) {
+			if !bytes.Contains(pkt.Payload, []byte("audiocodecid")) {
-				waitType = append(waitType, TagAudio)
+				waitAudio = false
 			}
 		}
 	}

pkg/rtmp/server.go

@@ -117,10 +117,6 @@ func (c *Conn) acceptCommand(b []byte) error {
 			}
 		}
 
-		if c.App == "" {
-			return fmt.Errorf("rtmp: read command %x", b)
-		}
-
 		payload := amf.EncodeItems(
 			"_result", tID,
 			map[string]any{"fmsVer": "FMS/3,0,1,123"},
@@ -129,9 +125,16 @@ func (c *Conn) acceptCommand(b []byte) error {
 		return c.writeMessage(3, TypeCommand, 0, payload)
 
 	case CommandReleaseStream:
+		// if app is empty - will use key as app
+		if c.App == "" && len(items) == 4 {
+			c.App, _ = items[3].(string)
+		}
+
 		payload := amf.EncodeItems("_result", tID, nil)
 		return c.writeMessage(3, TypeCommand, 0, payload)
 
+	case CommandFCPublish: // no response
+
 	case CommandCreateStream:
 		payload := amf.EncodeItems("_result", tID, nil, 1)
 		return c.writeMessage(3, TypeCommand, 0, payload)
@@ -140,8 +143,6 @@ func (c *Conn) acceptCommand(b []byte) error {
 		c.Intent = cmd
 		c.streamID = 1
 
-	case CommandFCPublish: // no response
-
 	default:
 		println("rtmp: unknown command: " + cmd)
 	}