Add async fabric session pump

agents/rap-node-agent/internal/mesh/client.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/example/remote-access-platform/agents/rap-node-agent/internal/fabricproto"
@@ -31,6 +32,24 @@ type FabricSessionClient struct {
 	conn       *websocket.Conn
 	timeout    time.Duration
 	maxPayload int
+	readMu     sync.Mutex
+	writeMu    sync.Mutex
+}
+
+type FabricSessionPumpOptions struct {
+	OutboundBuffer int
+	InboundBuffer  int
+	ErrorBuffer    int
+}
+
+type FabricSessionPump struct {
+	session  *FabricSessionClient
+	outbound chan fabricproto.Frame
+	inbound  chan fabricproto.Frame
+	errors   chan error
+	done     chan struct{}
+	cancel   context.CancelFunc
+	closeMu  sync.Once
 }
 
 func NewClient(baseURL string) Client {
@@ -192,6 +211,8 @@ func (c *FabricSessionClient) WriteFrame(ctx context.Context, frame fabricproto.
 	if err != nil {
 		return err
 	}
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
 	c.applyWriteDeadline(ctx)
 	return c.conn.WriteMessage(websocket.BinaryMessage, payload)
 }
@@ -200,6 +221,8 @@ func (c *FabricSessionClient) ReadFrame(ctx context.Context) (fabricproto.Frame,
 	if c == nil || c.conn == nil {
 		return fabricproto.Frame{}, fmt.Errorf("fabric session client is closed")
 	}
+	c.readMu.Lock()
+	defer c.readMu.Unlock()
 	c.applyReadDeadline(ctx)
 	messageType, responsePayload, err := c.conn.ReadMessage()
 	if err != nil {
@@ -218,6 +241,118 @@ func (c *FabricSessionClient) RoundTrip(ctx context.Context, frame fabricproto.F
 	return c.ReadFrame(ctx)
 }
 
+func (c *FabricSessionClient) StartPump(ctx context.Context, opts FabricSessionPumpOptions) *FabricSessionPump {
+	if opts.OutboundBuffer <= 0 {
+		opts.OutboundBuffer = 64
+	}
+	if opts.InboundBuffer <= 0 {
+		opts.InboundBuffer = 64
+	}
+	if opts.ErrorBuffer <= 0 {
+		opts.ErrorBuffer = 8
+	}
+	pumpCtx, cancel := context.WithCancel(ctx)
+	pump := &FabricSessionPump{
+		session:  c,
+		outbound: make(chan fabricproto.Frame, opts.OutboundBuffer),
+		inbound:  make(chan fabricproto.Frame, opts.InboundBuffer),
+		errors:   make(chan error, opts.ErrorBuffer),
+		done:     make(chan struct{}),
+		cancel:   cancel,
+	}
+	go pump.writeLoop(pumpCtx)
+	go pump.readLoop(pumpCtx)
+	return pump
+}
+
+func (p *FabricSessionPump) Send(ctx context.Context, frame fabricproto.Frame) error {
+	if p == nil {
+		return fmt.Errorf("fabric session pump is nil")
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-p.done:
+		return fmt.Errorf("fabric session pump is closed")
+	case p.outbound <- frame:
+		return nil
+	}
+}
+
+func (p *FabricSessionPump) Frames() <-chan fabricproto.Frame {
+	if p == nil {
+		return nil
+	}
+	return p.inbound
+}
+
+func (p *FabricSessionPump) Errors() <-chan error {
+	if p == nil {
+		return nil
+	}
+	return p.errors
+}
+
+func (p *FabricSessionPump) Close() error {
+	if p == nil {
+		return nil
+	}
+	var err error
+	p.closeMu.Do(func() {
+		close(p.done)
+		p.cancel()
+		err = p.session.Close()
+	})
+	return err
+}
+
+func (p *FabricSessionPump) writeLoop(ctx context.Context) {
+	defer p.Close()
+	for {
+		select {
+		case <-ctx.Done():
+			p.reportError(ctx.Err())
+			return
+		case <-p.done:
+			return
+		case frame := <-p.outbound:
+			if err := p.session.WriteFrame(ctx, frame); err != nil {
+				p.reportError(err)
+				return
+			}
+		}
+	}
+}
+
+func (p *FabricSessionPump) readLoop(ctx context.Context) {
+	defer p.Close()
+	for {
+		frame, err := p.session.ReadFrame(ctx)
+		if err != nil {
+			p.reportError(err)
+			return
+		}
+		select {
+		case <-ctx.Done():
+			p.reportError(ctx.Err())
+			return
+		case <-p.done:
+			return
+		case p.inbound <- frame:
+		}
+	}
+}
+
+func (p *FabricSessionPump) reportError(err error) {
+	if err == nil {
+		return
+	}
+	select {
+	case p.errors <- err:
+	default:
+	}
+}
+
 func (c *FabricSessionClient) applyReadDeadline(ctx context.Context) {
 	if deadline, ok := ctx.Deadline(); ok {
 		_ = c.conn.SetReadDeadline(deadline)