Merge branch 'master' into yutopp-hack

.github/workflows/ci.yml

@@ -3,16 +3,16 @@ on: [push, pull_request]
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
-        go: [ '1.19', '1.18' ]
+        go: [ '1.22.0', '1.21.0', '1.20', '1.19' ]
     name: ${{ matrix.go }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
 
       - name: Setup go
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go }}
 
@@ -26,10 +26,10 @@ jobs:
 
       - run: make vet
 
-      - uses: codecov/codecov-action@v1
+      - uses: codecov/codecov-action@v4
         with:
           files: ./coverage.txt
           flags: unittests
           name: codecov-umbrella-${{ matrix.go }}
-          fail_ci_if_error: true
+          token: ${{ secrets.CODECOV_TOKEN }}
           verbose: true

client.go

@@ -9,6 +9,7 @@ package rtmp
 
 import (
 	"context"
+	"crypto/tls"
 	"net"
 
 	"github.com/pkg/errors"
@@ -26,8 +27,11 @@ func WithContextDialer(dialFunc func(context.Context, string, string) (net.Conn,
 
 type DialOption func(*dialOptions)
 
-func Dial(protocol, addr string, config *ConnConfig, opts ...DialOption) (*ClientConn, error) {
+// 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)
 		},
@@ -36,30 +40,44 @@ func Dial(protocol, addr string, config *ConnConfig, opts ...DialOption) (*Clien
 		o(opt)
 	}
 
-	if protocol != "rtmp" {
-		return nil, errors.Errorf("Unknown protocol: %s", protocol)
+	if protocol != "rtmp" && protocol != "rtmps" {
+		return nil, errors.Errorf("unknown protocol: %s", protocol)
 	}
 
-	// TODO: support ctx
-	rwc, err := opt.dialFunc(context.Background(), "tcp", addr)
+	rwc, err := opt.dialFunc(ctx, "tcp", addr)
 	if err != nil {
 		return nil, err
 	}
 
 	return newClientConnWithSetup(rwc, config)
 }
 
+// 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...)
+}
+
+// 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))
 }
 
-func makeValidAddr(addr string) (string, error) {
-	host, port, err := net.SplitHostPort(addr)
-	if err != nil {
-		if err, ok := err.(*net.AddrError); ok && err.Err == "missing port in address" {
-			return makeValidAddr(addr + ":1935") // Default RTMP port
-		}
-		return "", err
+// 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,
 	}
-	return net.JoinHostPort(host, port), nil
+	opts = append([]DialOption{WithContextDialer(dialer.DialContext)}, opts...)
+	return DialContext(ctx, protocol, addr, config, opts...)
+}
+
+// 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))
 }

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()