Support client play handler / etc

README.md

@@ -21,7 +21,7 @@ See also [server_demo](https://github.com/yutopp/go-rtmp/tree/master/example/ser
 ## Documentation
 
 - [GoDoc](https://pkg.go.dev/github.com/yutopp/go-rtmp)
-- [REAL-TIME MESSAGING PROTOCOL (RTMP) SPECIFICATION](https://www.adobe.com/devnet/rtmp.html)
+- [Adobe RTMP Specification](https://rtmp.veriskope.com/docs/spec/)
 
 
 ## NOTES

client.go

@@ -8,45 +8,76 @@
 package rtmp
 
 import (
+	"context"
 	"crypto/tls"
 	"net"
 
 	"github.com/pkg/errors"
 )
 
-func Dial(protocol, addr string, config *ConnConfig) (*ClientConn, error) {
-	return DialWithDialer(&net.Dialer{}, protocol, addr, config)
+type dialOptions struct {
+	dialFunc func(ctx context.Context, network, addr string) (net.Conn, error)
 }
 
-func TLSDial(protocol, addr string, config *ConnConfig, tlsConfig *tls.Config) (*ClientConn, error) {
-	return DialWithTLSDialer(&tls.Dialer{
-		NetDialer: &net.Dialer{},
-		Config:    tlsConfig,
-	}, protocol, addr, config)
+func WithContextDialer(dialFunc func(context.Context, string, string) (net.Conn, error)) DialOption {
+	return func(o *dialOptions) {
+		o.dialFunc = dialFunc
+	}
 }
 
-func DialWithDialer(dialer *net.Dialer, protocol, addr string, config *ConnConfig) (*ClientConn, error) {
-	if protocol != "rtmp" {
-		return nil, errors.Errorf("Unknown protocol: %s", protocol)
+type DialOption func(*dialOptions)
+
+// DialContext dials a connection to the specified address.
+// The protocol must be "rtmp" or "rtmps".
+func DialContext(ctx context.Context, protocol, addr string, config *ConnConfig, opts ...DialOption) (*ClientConn, error) {
+	opt := &dialOptions{
+		// default dialer
+		dialFunc: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return (&net.Dialer{}).DialContext(ctx, network, addr)
+		},
+	}
+	for _, o := range opts {
+		o(opt)
 	}
 
-	rwc, err := dialer.Dial("tcp", addr)
+	if protocol != "rtmp" && protocol != "rtmps" {
+		return nil, errors.Errorf("unknown protocol: %s", protocol)
+	}
+
+	rwc, err := opt.dialFunc(ctx, "tcp", addr)
 	if err != nil {
 		return nil, err
 	}
 
 	return newClientConnWithSetup(rwc, config)
 }
 
-func DialWithTLSDialer(dialer *tls.Dialer, protocol, addr string, config *ConnConfig) (*ClientConn, error) {
-	if protocol != "rtmps" {
-		return nil, errors.Errorf("Unknown protocol: %s", protocol)
-	}
+// Dial dials a connection to the specified address.
+func Dial(protocol, addr string, config *ConnConfig, opts ...DialOption) (*ClientConn, error) {
+	return DialContext(context.Background(), protocol, addr, config, opts...)
+}
 
-	rwc, err := dialer.Dial("tcp", addr)
-	if err != nil {
-		return nil, err
+// DialWithDialer dials a connection to the specified address with the specified dialer.
+func DialWithDialer(dialer *net.Dialer, protocol, addr string, config *ConnConfig) (*ClientConn, error) {
+	return Dial(protocol, addr, config, WithContextDialer(dialer.DialContext))
+}
+
+// TLSDialContext dials a connection to the specified address with TLS.
+func TLSDialContext(ctx context.Context, protocol, addr string, config *ConnConfig, tlsConfig *tls.Config, opts ...DialOption) (*ClientConn, error) {
+	dialer := &tls.Dialer{
+		NetDialer: &net.Dialer{},
+		Config:    tlsConfig,
 	}
+	opts = append([]DialOption{WithContextDialer(dialer.DialContext)}, opts...)
+	return DialContext(ctx, protocol, addr, config, opts...)
+}
 
-	return newClientConnWithSetup(rwc, config)
+// TLSDial dials a connection to the specified address with TLS.
+func TLSDial(protocol, addr string, config *ConnConfig, tlsConfig *tls.Config, opts ...DialOption) (*ClientConn, error) {
+	return TLSDialContext(context.Background(), protocol, addr, config, tlsConfig, opts...)
+}
+
+// DialWithTLSDialer dials a connection to the specified address with the specified TLS dialer.
+func DialWithTLSDialer(dialer *tls.Dialer, protocol, addr string, config *ConnConfig) (*ClientConn, error) {
+	return Dial(protocol, addr, config, WithContextDialer(dialer.DialContext))
 }

client_conn.go

@@ -104,6 +104,7 @@ func (cc *ClientConn) CreateStream(body *message.NetConnectionCreateStream, chun
 	if err != nil {
 		return nil, err
 	}
+	newStream.handler.ChangeState(streamStateClientConnected)
 
 	return newStream, nil
 }

client_control_connected_handler.go

@@ -0,0 +1,59 @@
+//
+// Copyright (c) 2023- yutopp ([email protected])
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at  https://www.boost.org/LICENSE_1_0.txt)
+//
+
+package rtmp
+
+import (
+	"github.com/yutopp/go-rtmp/internal"
+	"github.com/yutopp/go-rtmp/message"
+)
+
+// clientControlConnectedHandler Handle control messages from a server in flow of connected.
+//
+//	transitions:
+//	  | _         -> self
+type clientControlConnectedHandler struct {
+	sh *streamHandler
+}
+
+var _ stateHandler = (*clientControlConnectedHandler)(nil)
+
+func (h *clientControlConnectedHandler) onMessage(
+	chunkStreamID int,
+	timestamp uint32,
+	msg message.Message,
+) error {
+	return internal.ErrPassThroughMsg
+}
+
+func (h *clientControlConnectedHandler) onData(
+	chunkStreamID int,
+	timestamp uint32,
+	dataMsg *message.DataMessage,
+	body interface{},
+) error {
+	return internal.ErrPassThroughMsg
+}
+
+func (h *clientControlConnectedHandler) onCommand(
+	chunkStreamID int,
+	timestamp uint32,
+	cmdMsg *message.CommandMessage,
+	body interface{},
+) error {
+	switch cmd := body.(type) {
+	case *message.NetStreamOnStatus:
+		if cmd.InfoObject.Code == message.NetStreamOnStatusCodePlayStart {
+			h.sh.ChangeState(streamStateClientPlay)
+		}
+
+		return internal.ErrPassThroughMsg
+
+	default:
+		return internal.ErrPassThroughMsg
+	}
+}

client_control_not_connected_handler.go

@@ -12,8 +12,6 @@ import (
 	"github.com/yutopp/go-rtmp/message"
 )
 
-var _ stateHandler = (*clientControlNotConnectedHandler)(nil)
-
 // clientControlNotConnectedHandler Handle control messages from a server in flow of connecting.
 //
 //	transitions:
@@ -23,6 +21,8 @@ type clientControlNotConnectedHandler struct {
 	sh *streamHandler
 }
 
+var _ stateHandler = (*clientControlNotConnectedHandler)(nil)
+
 func (h *clientControlNotConnectedHandler) onMessage(
 	chunkStreamID int,
 	timestamp uint32,

client_data_play_handler.go

@@ -0,0 +1,64 @@
+//
+// Copyright (c) 2023- yutopp ([email protected])
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at  https://www.boost.org/LICENSE_1_0.txt)
+//
+
+package rtmp
+
+import (
+	"github.com/yutopp/go-rtmp/internal"
+	"github.com/yutopp/go-rtmp/message"
+)
+
+// clientDataPlayHandler Handle control messages from a server in flow of connected.
+//
+//	transitions:
+//	  | _         -> self
+type clientDataPlayHandler struct {
+	sh *streamHandler
+}
+
+var _ stateHandler = (*clientDataPlayHandler)(nil)
+
+func (h *clientDataPlayHandler) onMessage(
+	chunkStreamID int,
+	timestamp uint32,
+	msg message.Message,
+) error {
+	switch msg := msg.(type) {
+	case *message.AudioMessage:
+		return h.sh.stream.userHandler().OnAudio(timestamp, msg.Payload)
+
+	case *message.VideoMessage:
+		return h.sh.stream.userHandler().OnVideo(timestamp, msg.Payload)
+
+	default:
+		return internal.ErrPassThroughMsg
+	}
+}
+
+func (h *clientDataPlayHandler) onData(
+	chunkStreamID int,
+	timestamp uint32,
+	dataMsg *message.DataMessage,
+	body interface{},
+) error {
+	switch data := body.(type) {
+	case *message.NetStreamSetDataFrame:
+		return h.sh.stream.userHandler().OnSetDataFrame(timestamp, data)
+
+	default:
+		return internal.ErrPassThroughMsg
+	}
+}
+
+func (h *clientDataPlayHandler) onCommand(
+	chunkStreamID int,
+	timestamp uint32,
+	cmdMsg *message.CommandMessage,
+	body interface{},
+) error {
+	return internal.ErrPassThroughMsg
+}

example/client_demo/Makefile

@@ -1,3 +1,3 @@
 PHONY: all
 all:
-	go build -i -v -o client_demo .
+	go build -v -o client_demo .

example/server_demo/Makefile

@@ -1,3 +1,3 @@
 PHONY: all
 all:
-	go build -i -v -o server_demo .
+	go build -v -o server_demo .

example/server_relay_demo/.gitignore

@@ -1 +1,2 @@
 /server_relay_demo
+/movie.mp4

example/server_relay_demo/Makefile

@@ -1,3 +1,3 @@
 PHONY: all
 all:
-	go build -i -v -o server_relay_demo .
+	go build -v -o server_relay_demo .

message/body_decoder.go

@@ -71,6 +71,7 @@
 	"getStreamLength": DecodeBodyGetStreamLength,
 	"ping":            DecodeBodyPing,
 	"closeStream":     DecodeBodyCloseStream,
+	"onStatus":        DecodeBodyOnStatus,
 }
 
 func CmdBodyDecoderFor(name string, transactionID int64) BodyDecoderFunc {
@@ -354,3 +355,22 @@
 
 	return nil
 }
+
+func DecodeBodyOnStatus(_ io.Reader, d AMFDecoder, v *AMFConvertible) error {
+	var commandObject interface{} // maybe nil
+	if err := d.Decode(&commandObject); err != nil {
+		return errors.Wrap(err, "Failed to decode 'OnStatus' args[0]")
+	}
+	var infoObject map[string]interface{}
+	if err := d.Decode(&infoObject); err != nil {
+		return errors.Wrap(err, "Failed to decode 'OnStatus' args[1]")
+	}
+
+	var cmd NetStreamOnStatus
+	if err := cmd.FromArgs(commandObject, infoObject); err != nil {
+		return errors.Wrap(err, "Failed to reconstruct 'OnStatus'")
+	}
+
+	*v = &cmd
+	return nil
+}

message/body_decoder_test.go

@@ -252,6 +252,45 @@ func TestDecodeCmdMessageCloseStream(t *testing.T) {
 	require.Equal(t, &NetStreamCloseStream{}, v)
 }
 
+func TestDecodeCmdMessageOnStatus(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		bin := []byte{
+			// nil
+			0x05,
+			// object start
+			0x03,
+			// key[0]: "level"
+			0x00, 0x05, 0x6c, 0x65, 0x76, 0x65, 0x6c,
+			// value[0]: string status
+			0x02, 0x00, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73,
+			// key[1]: "code",
+			0x00, 0x04, 0x63, 0x6f, 0x64, 0x65,
+			// value[1]: string NetStream.Play.Start
+			0x02, 0x00, 0x14, 0x4e, 0x65, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x50, 0x6c, 0x61, 0x79, 0x2e, 0x53, 0x74, 0x61, 0x72, 0x74,
+			// key[2]: "description",
+			0x00, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e,
+			// value[2]: string abc
+			0x02, 0x00, 0x03, 0x61, 0x62, 0x63,
+			// empty key, object end
+			0x00, 0x00, 0x09,
+		}
+		r := bytes.NewReader(bin)
+		d := amf0.NewDecoder(r)
+
+		var v AMFConvertible
+		err := CmdBodyDecoderFor("onStatus", 42)(r, d, &v)
+		require.Nil(t, err)
+		require.Equal(t, &NetStreamOnStatus{
+			InfoObject: NetStreamOnStatusInfoObject{
+				Level:           NetStreamOnStatusLevelStatus,
+				Code:            NetStreamOnStatusCodePlayStart,
+				Description:     "abc",
+				ExtraProperties: map[string]interface{}{},
+			},
+		}, v)
+	})
+}
+
 func TestDecodeCmdMessageUnknown(t *testing.T) {
 	bin := []byte{
 		// nil