export outQueueSize in client/server configs (#248)

om26er · muzzammilshahid · web-flow · commit e36f20c954d5 · 2025-11-05T19:35:52.000+05:00
* export outQueueSize in client/server configs

* add default queue size in peers

* make outque constants for client/router

---------

Co-authored-by: Muzzammil Shahid &lt;muzzammil@xconn.io&gt;
diff --git a/acceptor.go b/acceptor.go
@@ -195,7 +195,12 @@ func UpgradeWebSocket(conn net.Conn, config *WebSocketServerConfig) (Peer, error
 		Server:            true,
 		KeepAliveInterval: config.KeepAliveInterval,
 		KeepAliveTimeout:  config.KeepAliveTimeout,
+		OutQueueSize:      config.OutQueueSize,
 	}
+	if peerConfig.OutQueueSize == 0 {
+		peerConfig.OutQueueSize = RouterOutQueueSizeDefault
+	}
+
 	peer, err := NewWebSocketPeer(conn, peerConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to init reader/writer: %w", err)
@@ -246,8 +251,8 @@ func (r *RawSocketAcceptor) Spec(serializerID SerializerID) (serializers.Seriali
 	return serializer, nil
 }
 
-func (r *RawSocketAcceptor) Accept(conn net.Conn) (BaseSession, error) {
-	peer, serializerID, err := UpgradeRawSocket(conn)
+func (r *RawSocketAcceptor) Accept(conn net.Conn, config *RawSocketServerConfig) (BaseSession, error) {
+	peer, serializerID, err := UpgradeRawSocket(conn, config)
 	if err != nil {
 		return nil, fmt.Errorf("failed to init reader/writer: %w", err)
 	}
@@ -265,7 +270,7 @@ func (r *RawSocketAcceptor) Accept(conn net.Conn) (BaseSession, error) {
 	return Accept(peer, hello, serializer, r.Authenticator)
 }
 
-func UpgradeRawSocket(conn net.Conn) (Peer, transports.Serializer, error) {
+func UpgradeRawSocket(conn net.Conn, config *RawSocketServerConfig) (Peer, transports.Serializer, error) {
 	maxMessageSize := transports.ProtocolMaxMsgSize
 
 	handshakeRequestRaw := make([]byte, 4)
@@ -290,8 +295,17 @@ func UpgradeRawSocket(conn net.Conn) (Peer, transports.Serializer, error) {
 		return nil, 0, err
 	}
 
+	if config == nil {
+		config = DefaultRawSocketServerConfig()
+	}
+
 	peerConfig := RawSocketPeerConfig{
-		Serializer: handshakeResponse.Serializer(),
+		Serializer:   handshakeResponse.Serializer(),
+		OutQueueSize: config.OutQueueSize,
+	}
+
+	if peerConfig.OutQueueSize == 0 {
+		peerConfig.OutQueueSize = RouterOutQueueSizeDefault
 	}
 
 	return NewRawSocketPeer(conn, peerConfig), handshakeRequest.Serializer(), nil
diff --git a/client.go b/client.go
@@ -24,6 +24,7 @@ type Client struct {
 	DialTimeout       time.Duration
 	KeepAliveInterval time.Duration
 	KeepAliveTimeout  time.Duration
+	OutQueueSize      int
 }
 
 func (c *Client) Connect(ctx context.Context, uri string, realm string) (*Session, error) {
@@ -39,6 +40,7 @@ func (c *Client) Connect(ctx context.Context, uri string, realm string) (*Sessio
 			NetDial:           c.NetDial,
 			KeepAliveInterval: c.KeepAliveInterval,
 			KeepAliveTimeout:  c.KeepAliveTimeout,
+			OutQueueSize:      c.OutQueueSize,
 		}
 
 		joiner := &WebSocketJoiner{
@@ -64,6 +66,7 @@ func (c *Client) Connect(ctx context.Context, uri string, realm string) (*Sessio
 			NetDial:           c.NetDial,
 			KeepAliveInterval: c.KeepAliveInterval,
 			KeepAliveTimeout:  c.KeepAliveTimeout,
+			OutQueueSize:      c.OutQueueSize,
 		}
 
 		joiner := &RawSocketJoiner{
@@ -116,13 +119,13 @@ func ConnectCryptosign(ctx context.Context, uri, realm, authid, privateKey strin
 }
 
 func ConnectInMemoryBase(router *Router, realm, authID, authRole string,
-	serializer serializers.Serializer) (BaseSession, error) {
+	serializer serializers.Serializer, outQueueSize int) (BaseSession, error) {
 
 	if serializer == nil {
 		return nil, fmt.Errorf("serializer must not be nil")
 	}
 
-	clientPeer, routerPeer := NewInMemoryPeerPair()
+	clientPeer, routerPeer := NewInMemoryPeerPair(outQueueSize)
 	sessionID := wampproto.GenerateID()
 	routerSession := NewBaseSession(
 		sessionID,
@@ -168,8 +171,9 @@ func ConnectInMemoryBase(router *Router, realm, authID, authRole string,
 func ConnectInMemory(router *Router, realm string) (*Session, error) {
 	authID := fmt.Sprintf("%012x", rand.Uint64())[:12] // #nosec
 	authRole := "trusted"
+	outQueueSize := 16
 
-	base, err := ConnectInMemoryBase(router, realm, authID, authRole, &serializers.MsgPackSerializer{})
+	base, err := ConnectInMemoryBase(router, realm, authID, authRole, &serializers.MsgPackSerializer{}, outQueueSize)
 	if err != nil {
 		return nil, err
 	}
diff --git a/client_test.go b/client_test.go
@@ -37,7 +37,7 @@ func connectInMemory(t *testing.T, router *xconn.Router, realm string,
 	authID := fmt.Sprintf("%012x", rand.Uint64())[:12]
 	authRole := "trusted"
 
-	base, err := xconn.ConnectInMemoryBase(router, realm, authID, authRole, serializer)
+	base, err := xconn.ConnectInMemoryBase(router, realm, authID, authRole, serializer, 0)
 	require.NoError(t, err)
 
 	return xconn.NewSession(base, base.Serializer())
diff --git a/joiner.go b/joiner.go
@@ -92,6 +92,7 @@ func DialWebSocket(ctx context.Context, url *netURL.URL, config *WSDialerConfig)
 		Server:            false,
 		KeepAliveInterval: config.KeepAliveInterval,
 		KeepAliveTimeout:  config.KeepAliveTimeout,
+		OutQueueSize:      config.OutQueueSize,
 	}
 	wsPeer, err := NewWebSocketPeer(conn, peerConfig)
 	if err != nil {
@@ -156,7 +157,10 @@ func DialRawSocket(ctx context.Context, url *netURL.URL, config *RawSocketDialer
 		return nil, err
 	}
 
-	return NewRawSocketPeer(conn, RawSocketPeerConfig{Serializer: config.Serializer}), nil
+	return NewRawSocketPeer(conn, RawSocketPeerConfig{
+		Serializer:   config.Serializer,
+		OutQueueSize: config.OutQueueSize,
+	}), nil
 }
 
 func Join(cl Peer, realm string, serializer serializers.Serializer,
diff --git a/peer.go b/peer.go
@@ -18,6 +18,11 @@ import (
 	"github.com/xconnio/wampproto-go/transports"
 )
 
+const (
+	ClientOutQueueSizeDefault = 16
+	RouterOutQueueSizeDefault = 64
+)
+
 func NewBaseSession(id uint64, realm, authID, authRole string, cl Peer, serializer serializers.Serializer) BaseSession {
 	return &baseSession{
 		id:         id,
@@ -194,14 +199,18 @@ func NewWebSocketPeer(conn net.Conn, peerConfig WSPeerConfig) (Peer, error) {
 		msgOpCode = ws.OpText
 	}
 
+	if peerConfig.OutQueueSize == 0 {
+		peerConfig.OutQueueSize = ClientOutQueueSizeDefault
+	}
+
 	peer := &WebSocketPeer{
 		transportType: TransportWebSocket,
 		protocol:      peerConfig.Protocol,
 		conn:          conn,
 		pingCh:        make(chan struct{}, 1),
 		wsMsgOp:       msgOpCode,
 		server:        peerConfig.Server,
-		writeChan:     make(chan []byte, 64),
+		writeChan:     make(chan []byte, peerConfig.OutQueueSize),
 		ctrlChan:      make(chan wsMsg, 2),
 		doneWriting:   make(chan struct{}),
 		closeChan:     make(chan struct{}),
@@ -435,11 +444,15 @@ type RawSocketPeer struct {
 }
 
 func NewRawSocketPeer(conn net.Conn, peerConfig RawSocketPeerConfig) Peer {
+	if peerConfig.OutQueueSize == 0 {
+		peerConfig.OutQueueSize = ClientOutQueueSizeDefault
+	}
+
 	peer := &RawSocketPeer{
 		transportType: TransportRawSocket,
 		conn:          conn,
 		serializer:    peerConfig.Serializer,
-		writeChan:     make(chan []byte, 64),
+		writeChan:     make(chan []byte, peerConfig.OutQueueSize),
 		ctrlChan:      make(chan rsMsg, 2),
 		doneWriting:   make(chan struct{}),
 	}
@@ -731,26 +744,26 @@ func (l *localPeer) Close() error {
 	return l.conn.Close()
 }
 
-func newLocalPeer(conn, otherSide net.Conn, bufferSize int) *localPeer {
-	if bufferSize <= 0 {
-		bufferSize = 16 // default fallback
+func newLocalPeer(conn, otherSide net.Conn, outQueueSize int) *localPeer {
+	if outQueueSize == 0 {
+		outQueueSize = ClientOutQueueSizeDefault
 	}
 
 	p := &localPeer{
 		conn:        conn,
 		other:       otherSide,
-		writeChan:   make(chan []byte, bufferSize),
+		writeChan:   make(chan []byte, outQueueSize),
 		doneWriting: make(chan struct{}),
 	}
 
 	go p.writer()
 	return p
 }
 
-func NewInMemoryPeerPair() (Peer, Peer) {
+func NewInMemoryPeerPair(outQueueSize int) (Peer, Peer) {
 	conn1, conn2 := net.Pipe()
-	peer1 := newLocalPeer(conn1, conn2, 16)
-	peer2 := newLocalPeer(conn2, conn1, 16)
+	peer1 := newLocalPeer(conn1, conn2, outQueueSize)
+	peer2 := newLocalPeer(conn2, conn1, outQueueSize)
 
 	return peer1, peer2
 }
diff --git a/peer_test.go b/peer_test.go
@@ -10,7 +10,7 @@ import (
 )
 
 func TestInMemoryPeer(t *testing.T) {
-	client, server := xconn.NewInMemoryPeerPair()
+	client, server := xconn.NewInMemoryPeerPair(0)
 
 	clientChan := make(chan []byte)
 	clientCloseChan := make(chan struct{})
diff --git a/router_test.go b/router_test.go
@@ -74,11 +74,11 @@ func TestRouterMetaKillByAuthID(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create sessions with authid "test"
-	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "trusted", &serializers.JSONSerializer{})
+	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "trusted", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session1 := xconn.NewSession(baseSession, baseSession.Serializer())
 
-	baseSession1, err := xconn.ConnectInMemoryBase(router, realmName, "test", "trusted", &serializers.JSONSerializer{})
+	baseSession1, err := xconn.ConnectInMemoryBase(router, realmName, "test", "trusted", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session2 := xconn.NewSession(baseSession1, baseSession.Serializer())
 
@@ -116,11 +116,11 @@ func TestRouterMetaKillByAuthRole(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create sessions with authrole "test"
-	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "test", &serializers.JSONSerializer{})
+	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "test", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session1 := xconn.NewSession(baseSession, baseSession.Serializer())
 
-	baseSession1, err := xconn.ConnectInMemoryBase(router, realmName, "test", "test", &serializers.JSONSerializer{})
+	baseSession1, err := xconn.ConnectInMemoryBase(router, realmName, "test", "test", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session2 := xconn.NewSession(baseSession1, baseSession.Serializer())
 
@@ -198,7 +198,7 @@ func TestRouterMetaSessionCount(t *testing.T) {
 	})
 
 	// Connect second session with role=admin
-	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "admin", &serializers.JSONSerializer{})
+	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "admin", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session2 := xconn.NewSession(baseSession, baseSession.Serializer())
 
@@ -241,7 +241,7 @@ func TestRouterMetaSessionList(t *testing.T) {
 	})
 
 	// Connect second session with role=admin
-	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "admin", &serializers.JSONSerializer{})
+	baseSession, err := xconn.ConnectInMemoryBase(router, realmName, "test", "admin", &serializers.JSONSerializer{}, 0)
 	require.NoError(t, err)
 	session2 := xconn.NewSession(baseSession, baseSession.Serializer())
 
@@ -305,7 +305,7 @@ func TestAuthorization(t *testing.T) {
 
 	createSession := func(role string) *xconn.Session {
 		authID := fmt.Sprintf("%012x", rand.Uint64())[:12] // #nosec
-		baseSession, err := xconn.ConnectInMemoryBase(router, realmName, authID, role, &serializers.JSONSerializer{})
+		baseSession, err := xconn.ConnectInMemoryBase(router, realmName, authID, role, &serializers.JSONSerializer{}, 0)
 		require.NoError(t, err)
 		return xconn.NewSession(baseSession, baseSession.Serializer())
 	}
@@ -487,7 +487,7 @@ func TestCustomAuthorizer(t *testing.T) {
 
 	createSession := func(role string) *xconn.Session {
 		authID := fmt.Sprintf("%012x", rand.Uint64())[:12] // #nosec
-		baseSession, err := xconn.ConnectInMemoryBase(router, realmName, authID, role, &serializers.JSONSerializer{})
+		baseSession, err := xconn.ConnectInMemoryBase(router, realmName, authID, role, &serializers.JSONSerializer{}, 0)
 		require.NoError(t, err)
 		return xconn.NewSession(baseSession, baseSession.Serializer())
 	}
@@ -663,7 +663,7 @@ func TestBlockedLocalClient(t *testing.T) {
 
 	// subscribe to topic 'io.xconn.test'
 	authID := fmt.Sprintf("%012x", rand.Uint64())[:12] // #nosec
-	baseSession, err := xconn.ConnectInMemoryBase(router, "realm1", authID, "trusted", &serializers.MsgPackSerializer{})
+	baseSession, err := xconn.ConnectInMemoryBase(router, "realm1", authID, "trusted", &serializers.MsgPackSerializer{}, 0)
 	require.NoError(t, err)
 
 	// publisher client
diff --git a/server.go b/server.go
@@ -20,6 +20,8 @@ const (
 	ListenerWebSocket ListenerType = iota
 	ListenerRawSocket
 	ListenerUniversalTCP
+
+	ServerOutQueueSizeDefault = 8
 )
 
 type Network string
@@ -36,6 +38,7 @@ type Server struct {
 	throttle          *Throttle
 	keepAliveInterval time.Duration
 	keepAliveTimeout  time.Duration
+	outQueueSize      int
 }
 
 func NewServer(router *Router, authenticator auth.ServerAuthenticator, config *ServerConfig) *Server {
@@ -57,6 +60,7 @@ func NewServer(router *Router, authenticator auth.ServerAuthenticator, config *S
 		server.throttle = config.Throttle
 		server.keepAliveInterval = config.KeepAliveInterval
 		server.keepAliveTimeout = config.KeepAliveTimeout
+		server.outQueueSize = config.OutQueueSize
 	}
 
 	return server
@@ -136,14 +140,19 @@ func (s *Server) HandleClient(conn net.Conn, listener ListenerType) {
 		}
 
 		if magicArray[0] == transports.MAGIC {
-			base, err = s.rsAcceptor.Accept(wrapped)
+			config := DefaultRawSocketServerConfig()
+			config.KeepAliveInterval = s.keepAliveInterval
+			config.KeepAliveTimeout = s.keepAliveTimeout
+			config.OutQueueSize = s.outQueueSize
+			base, err = s.rsAcceptor.Accept(wrapped, config)
 			if err != nil {
 				return
 			}
 		} else {
 			config := DefaultWebSocketServerConfig()
 			config.KeepAliveInterval = s.keepAliveInterval
 			config.KeepAliveTimeout = s.keepAliveTimeout
+			config.OutQueueSize = s.outQueueSize
 			base, err = s.wsAcceptor.Accept(wrapped, s.router, config)
 			if err != nil {
 				return
@@ -153,13 +162,18 @@ func (s *Server) HandleClient(conn net.Conn, listener ListenerType) {
 		config := DefaultWebSocketServerConfig()
 		config.KeepAliveInterval = s.keepAliveInterval
 		config.KeepAliveTimeout = s.keepAliveTimeout
+		config.OutQueueSize = s.outQueueSize
 		base, err = s.wsAcceptor.Accept(conn, s.router, config)
 		if err != nil {
 			log.Debugf("failed to accept websocket connection: %v", err)
 			return
 		}
 	case ListenerRawSocket:
-		base, err = s.rsAcceptor.Accept(conn)
+		config := DefaultRawSocketServerConfig()
+		config.KeepAliveInterval = s.keepAliveInterval
+		config.KeepAliveTimeout = s.keepAliveTimeout
+		config.OutQueueSize = s.outQueueSize
+		base, err = s.rsAcceptor.Accept(conn, config)
 		if err != nil {
 			log.Debugf("failed to accept rawsocket connection: %v", err)
 			return
diff --git a/types.go b/types.go

Original file line number	Diff line number	Diff line change
`@@ -92,6 +92,7 @@ func DialWebSocket(ctx context.Context, url netURL.URL, config WSDialerConfig)`
`92`	`92`	`Server: false,`
`93`	`93`	`KeepAliveInterval: config.KeepAliveInterval,`
`94`	`94`	`KeepAliveTimeout: config.KeepAliveTimeout,`
	`95`	`+ OutQueueSize: config.OutQueueSize,`
`95`	`96`	`}`
`96`	`97`	`wsPeer, err := NewWebSocketPeer(conn, peerConfig)`
`97`	`98`	`if err != nil {`
`@@ -156,7 +157,10 @@ func DialRawSocket(ctx context.Context, url netURL.URL, config RawSocketDialer`
`156`	`157`	`return nil, err`
`157`	`158`	`}`
`158`	`159`
`159`		`- return NewRawSocketPeer(conn, RawSocketPeerConfig{Serializer: config.Serializer}), nil`
	`160`	`+ return NewRawSocketPeer(conn, RawSocketPeerConfig{`
	`161`	`+ Serializer: config.Serializer,`
	`162`	`+ OutQueueSize: config.OutQueueSize,`
	`163`	`+ }), nil`
`160`	`164`	`}`
`161`	`165`
`162`	`166`	`func Join(cl Peer, realm string, serializer serializers.Serializer,`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ import (`
`10`	`10`	`)`
`11`	`11`
`12`	`12`	`func TestInMemoryPeer(t *testing.T) {`
`13`		`- client, server := xconn.NewInMemoryPeerPair()`
	`13`	`+ client, server := xconn.NewInMemoryPeerPair(0)`
`14`	`14`
`15`	`15`	`clientChan := make(chan []byte)`
`16`	`16`	`clientCloseChan := make(chan struct{})`