feat: better

Author: e.s.prilutskiy

connection.go

@@ -13,6 +13,11 @@ import (
 	"time"
 )
 
+const (
+	notificationQueueSize         = 1024
+	notificationQueueDrainTimeout = 5 * time.Second
+)
+
 type anyMessage struct {
 	JSONRPC string           `json:"jsonrpc"`
 	ID      *json.RawMessage `json:"id,omitempty"`
@@ -73,7 +78,7 @@ func NewConnection(handler MethodHandler, peerInput io.Writer, peerOutput io.Rea
 		cancel:            cancel,
 		inboundCtx:        inboundCtx,
 		inboundCancel:     inboundCancel,
-		notificationQueue: make(chan *anyMessage, 1024),
+		notificationQueue: make(chan *anyMessage, notificationQueueSize),
 	}
 	go c.receive()
 	go c.processNotifications()
@@ -160,7 +165,6 @@ func (c *Connection) receive() {
 
 	// Cancel inboundCtx after notifications finish, but ensure we don't leak forever if a
 	// handler blocks waiting for cancellation.
-	const drainTimeout = 5 * time.Second
 	go func() {
 		done := make(chan struct{})
 		go func() {
@@ -169,15 +173,16 @@ func (c *Connection) receive() {
 		}()
 		select {
 		case <-done:
-		case <-time.After(drainTimeout):
+		case <-time.After(notificationQueueDrainTimeout):
 		}
 		c.inboundCancel(cause)
 	}()
 
 	c.loggerOrDefault().Info("peer connection closed")
 }
 
-// processNotifications processes notifications sequentially to maintain order
+// processNotifications processes notifications sequentially to maintain order.
+// It terminates when notificationQueue is closed (e.g. on disconnect in receive()).
 func (c *Connection) processNotifications() {
 	for msg := range c.notificationQueue {
 		c.handleInbound(msg)