feat: support server-side part

Author: DMwangnima

pkg/remote/trans/ttstream/server_handler.go

@@ -126,24 +126,21 @@ func (t *svrTransHandler) OnActive(ctx context.Context, conn net.Conn) (context.
 // OnRead control the connection level lifecycle.
 // only when OnRead return, netpoll can close the connection buffer
 func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) (err error) {
+	trans, _ := ctx.Value(serverTransCtxKey{}).(*serverTransport)
+	if trans == nil {
+		err = fmt.Errorf("server transport is nil")
+		return
+	}
 	var wg sync.WaitGroup
 	defer func() {
 		wg.Wait()
-		trans, _ := ctx.Value(serverTransCtxKey{}).(*serverTransport)
-		if trans != nil {
-			trans.WaitClosed()
-		}
+		trans.WaitClosed()
 		if errors.Is(err, io.EOF) {
 			err = nil
 		}
 	}()
 	// connection level goroutine
 	for {
-		trans, _ := ctx.Value(serverTransCtxKey{}).(*serverTransport)
-		if trans == nil {
-			err = fmt.Errorf("server transport is nil")
-			return
-		}
 		var st *serverStream
 		// ReadStream will block until a stream coming or conn return error
 		st, err = trans.ReadStream(ctx)

pkg/remote/trans/ttstream/transport_client.go

@@ -190,14 +190,17 @@ func (t *clientTransport) loopRead() error {
 func (t *clientTransport) WriteFrame(fr *Frame) (err error) {
 	// todo: optimize performance
 	t.mu.Lock()
-	defer t.mu.Unlock()
+	defer func() {
+		t.mu.Unlock()
+		if err != nil {
+			t.Close(err)
+		}
+	}()
 	if err = EncodeFrame(context.Background(), t.writer, fr); err != nil {
-		t.Close(err)
 		return err
 	}
 	recycleFrame(fr)
 	if err = t.writer.Flush(); err != nil {
-		t.Close(err)
 		return err
 	}
 	return nil

pkg/remote/trans/ttstream/transport_server.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"io"
 	"net"
+	"sync"
 	"sync/atomic"
 
 	"github.com/cloudwego/gopkg/bufiox"
@@ -35,6 +36,9 @@ import (
 type serverTransport struct {
 	conn   netpoll.Connection
 	stream atomic.Pointer[serverStream]
+	// mu protects writer
+	// todo: optimize performance
+	mu     sync.Mutex
 	writer *writerBuffer
 	// transport should operate directly on stream
 	scache        []*serverStream                // size is streamCacheSize
@@ -141,6 +145,10 @@ func (t *serverTransport) readFrame(reader bufiox.Reader) error {
 		s = newServerStream(ctx, t, fr.streamFrame)
 		s.cancelFunc = cFunc
 		t.storeStream(s)
+		// todo: 如果能确保一个连接就是被一个 stream 独占的，那么这里不用那么麻烦，直接通知即可
+		// 是否可能存在新的 stream 开始了，但是连接上继续收到了上一个 stream 的情况？
+		// 这里需要确保一个流的生命周期结束后，连接才能被分配给下一个流
+		// 此处需要做防御性编程，确保一个连接只被一个 Stream 占据
 		err = t.spipe.Write(context.Background(), s)
 	} else {
 		// load exist stream
@@ -184,11 +192,22 @@ func (t *serverTransport) loopRead() error {
 
 // WriteFrame is concurrent safe
 func (t *serverTransport) WriteFrame(fr *Frame) (err error) {
+	// todo: optimize performance
+	t.mu.Lock()
+	defer func() {
+		t.mu.Unlock()
+		if err != nil {
+			t.Close(err)
+		}
+	}()
 	if err = EncodeFrame(context.Background(), t.writer, fr); err != nil {
 		return err
 	}
 	recycleFrame(fr)
-	return t.writer.Flush()
+	if err = t.writer.Flush(); err != nil {
+		return err
+	}
+	return nil
 }
 
 func (t *serverTransport) CloseStream(sid int32) (err error) {