Merge pull request #58 from gomaja/feature-negotiated-stream-id-usage

feat: features usage of the exact negotiated stream id limit

Author: wmnsk

client.go

@@ -14,8 +14,7 @@ import (
 )
 
 // Dial establishes a M3UA connection as a client.
-//
-// After successfully established the connection with peer, state-changing
+// After successfully establishing the connection with peer, state-changing
 // signals and heartbeats are automatically handled background in another goroutine.
 func Dial(ctx context.Context, net string, laddr, raddr *sctp.SCTPAddr, cfg *Config) (*Conn, error) {
 	var err error
@@ -42,6 +41,12 @@ func Dial(ctx context.Context, net string, laddr, raddr *sctp.SCTPAddr, cfg *Con
 		return nil, err
 	}
 
+	r, err := conn.sctpConn.GetStatus()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get sctpConn status: %w", err)
+	}
+	conn.maxMessageStreamID = r.Ostreams - 1 // removing 1 for management messages of stream ID 0
+
 	go func() {
 		conn.stateChan <- StateAspDown
 	}()
@@ -50,10 +55,12 @@ func Dial(ctx context.Context, net string, laddr, raddr *sctp.SCTPAddr, cfg *Con
 	select {
 	case _, ok := <-conn.established:
 		if !ok {
+			conn.sctpConn.Close()
 			return nil, ErrFailedToEstablish
 		}
 		return conn, nil
 	case <-time.After(10 * time.Second):
+		conn.sctpConn.Close()
 		return nil, ErrTimeout
 	}
 }

config.go

@@ -25,7 +25,7 @@ func NewHeartbeatInfo(interval, timer time.Duration, data []byte) *HeartbeatInfo
 	}
 }
 
-// Config is a configration that defines a M3UA server.
+// Config is a configuration that defines a M3UA server.
 type Config struct {
 	*HeartbeatInfo
 	AspIdentifier          *params.Param

conn.go

@@ -6,6 +6,7 @@ package m3ua
 
 import (
 	"fmt"
+	"math/rand"
 	"net"
 	"sync"
 	"time"
@@ -24,6 +25,9 @@ const (
 
 // Conn represents a M3UA connection, which satisfies standard net.Conn interface.
 type Conn struct {
+	// maxMessageStreamID is the maximum negotiated sctp stream ID used,
+	// must not be zero, must vary from 1 to maxMessageStreamID
+	maxMessageStreamID uint16
 	// muState is to Lock when updating state
 	muState *sync.RWMutex
 	// mode represents the endpoint works as client or server
@@ -100,7 +104,9 @@ func (c *Conn) ReadPD() (pd *params.ProtocolDataPayload, err error) {
 
 // Write writes data to the connection.
 func (c *Conn) Write(b []byte) (n int, err error) {
-	return c.WriteToStream(b, c.StreamID())
+	stream := c.chooseStreamID()
+
+	return c.WriteToStream(b, stream)
 }
 
 // WriteToStream writes data to the connection and specific stream
@@ -133,7 +139,9 @@ func (c *Conn) WriteToStream(b []byte, streamID uint16) (n int, err error) {
 
 // WritePD writes data with a specific mtp3 protocol data to the connection.
 func (c *Conn) WritePD(protocolData *params.Param) (n int, err error) {
-	return c.WritePDToStream(protocolData, c.StreamID())
+	stream := c.chooseStreamID()
+
+	return c.WritePDToStream(protocolData, stream)
 }
 
 // WritePDToStream writes data with a specific mtp3 protocol data to the connection and specific stream
@@ -238,3 +246,19 @@ func (c *Conn) State() State {
 func (c *Conn) StreamID() uint16 {
 	return c.sctpInfo.Stream
 }
+
+// MaxMessageStreamID returns the maximum negotiated sctp stream ID
+// The streamID for sending a message must start from 1 up to maxMessageStreamID, 0 is reserved for management messages
+func (c *Conn) MaxMessageStreamID() uint16 {
+	return c.maxMessageStreamID
+}
+
+// chooseStreamID generates a random uint16 from 1 to max (inclusive)
+func (c *Conn) chooseStreamID() uint16 {
+	if c.maxMessageStreamID == 1 {
+		return 1
+	}
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	randomNum := uint16(r.Intn(int(c.maxMessageStreamID)))
+	return randomNum + 1
+}

fsm.go

@@ -25,6 +25,23 @@ const (
 	StateSCTPRI
 )
 
+func (s State) String() string {
+	switch s {
+	case StateAspDown:
+		return "AspDown"
+	case StateAspInactive:
+		return "AspInactive"
+	case StateAspActive:
+		return "AspActive"
+	case StateSCTPCDI:
+		return "SCTPCDI"
+	case StateSCTPRI:
+		return "SCTPRI"
+	default:
+		return "Unknown"
+	}
+}
+
 func (c *Conn) handleStateUpdate(current State) error {
 	c.muState.Lock()
 	defer c.muState.Unlock()

go.mod

@@ -4,6 +4,6 @@ go 1.23
 
 require (
 	github.com/google/go-cmp v0.7.0
-	github.com/ishidawataru/sctp v0.0.0-20250425063137-c91d9fd4afaf
+	github.com/ishidawataru/sctp v0.0.0-20250427101207-53eab83c1cf6
 	github.com/pascaldekloe/goe v0.1.1
 )

go.sum

@@ -2,5 +2,7 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/ishidawataru/sctp v0.0.0-20250425063137-c91d9fd4afaf h1:Yk0+ELKOCwbQ3ysGeTYMQYZJMb1rEj7VPTRXHqaEasM=
 github.com/ishidawataru/sctp v0.0.0-20250425063137-c91d9fd4afaf/go.mod h1:co9pwDoBCm1kGxawmb4sPq0cSIOOWNPT4KnHotMP1Zg=
+github.com/ishidawataru/sctp v0.0.0-20250427101207-53eab83c1cf6 h1:BcV9jRUmgOhP6dWHo1awB1QQQjGMRUuS9E4/lmwcbQY=
+github.com/ishidawataru/sctp v0.0.0-20250427101207-53eab83c1cf6/go.mod h1:co9pwDoBCm1kGxawmb4sPq0cSIOOWNPT4KnHotMP1Zg=
 github.com/pascaldekloe/goe v0.1.1 h1:Ah6WQ56rZONR3RW3qWa2NCZ6JAVvSpUcoLBaOmYFt9Q=
 github.com/pascaldekloe/goe v0.1.1/go.mod h1:KSyfaxQOh0HZPjDP1FL/kFtbqYqrALJTaMafFUIccqU=

server.go

@@ -6,7 +6,6 @@ package m3ua
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"sync"
@@ -39,7 +38,7 @@ func Listen(net string, laddr *sctp.SCTPAddr, cfg *Config) (*Listener, error) {
 }
 
 // Accept waits for and returns the next connection to the listener.
-// After successfully established the association with peer, Payload can be read with Read() func.
+// After successfully establishing the association with peer, Payload can be read with Read() func.
 // Other signals are automatically handled background in another goroutine.
 func (l *Listener) Accept(ctx context.Context) (*Conn, error) {
 	conn := &Conn{
@@ -63,9 +62,16 @@ func (l *Listener) Accept(ctx context.Context) (*Conn, error) {
 	var ok bool
 	conn.sctpConn, ok = c.(*sctp.SCTPConn)
 	if !ok {
-		return nil, errors.New("failed to assert conn")
+		c.Close()
+		return nil, fmt.Errorf("failed to assert server connection")
 	}
 
+	r, err := conn.sctpConn.GetStatus()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get sctpConn status: %w", err)
+	}
+	conn.maxMessageStreamID = r.Ostreams - 1 // removing 1 for management messages of stream ID 0
+
 	go func() {
 		conn.stateChan <- StateAspDown
 	}()
@@ -74,10 +80,12 @@ func (l *Listener) Accept(ctx context.Context) (*Conn, error) {
 	select {
 	case _, ok := <-conn.established:
 		if !ok {
+			conn.sctpConn.Close()
 			return nil, ErrFailedToEstablish
 		}
 		return conn, nil
 	case <-time.After(10 * time.Second):
+		conn.sctpConn.Close()
 		return nil, ErrTimeout
 	}
 }