conn.go

@@ -105,6 +105,10 @@ func newConn(rwc io.ReadWriteCloser, config *ConnConfig) *Conn {
 	return conn
 }
 
+func (c *Conn) GetChunkStreamer() *ChunkStreamer {
+	return c.streamer
+}
+
 func (c *Conn) Close() error {
 	c.m.Lock()
 	defer c.m.Unlock()

message/body_decoder.go

@@ -200,7 +200,12 @@ func DecodeBodyPublish(_ io.Reader, d AMFDecoder, v *AMFConvertible) error {
 	}
 	var publishingType string
 	if err := d.Decode(&publishingType); err != nil {
-		return errors.Wrap(err, "Failed to decode 'publish' args[2]")
+		// value is optional
+		if errors.Is(err, io.EOF) {
+			publishingType = "live"
+		} else {
+			return errors.Wrap(err, "Failed to decode 'publish' args[2]")
+		}
 	}
 
 	var cmd NetStreamPublish

message/body_decoder_test.go

@@ -112,6 +112,25 @@ func TestDecodeCmdMessagePublish(t *testing.T) {
 	}, v)
 }
 
+func TestDecodeCmdMessagePublishWithoutPublishingType(t *testing.T) {
+	bin := []byte{
+		// nil
+		0x05,
+		// string: abc
+		0x02, 0x00, 0x03, 0x61, 0x62, 0x63,
+	}
+	r := bytes.NewReader(bin)
+	d := amf0.NewDecoder(r)
+
+	var v AMFConvertible
+	err := CmdBodyDecoderFor("publish", 42)(r, d, &v)
+	require.Nil(t, err)
+	require.Equal(t, &NetStreamPublish{
+		PublishingName: "abc",
+		PublishingType: "live",
+	}, v)
+}
+
 func TestDecodeCmdMessagePlay(t *testing.T) {
 	bin := []byte{
 		// nil