update ftcp

listener/ftcp/listener.go

@@ -2,9 +2,8 @@ package ftcp
 
 import (
 	"net"
-	"sync"
-	"sync/atomic"
 
+	"github.com/go-gost/core/common/util/udp"
 	"github.com/go-gost/core/listener"
 	"github.com/go-gost/core/logger"
 	md "github.com/go-gost/core/metadata"
@@ -18,13 +17,10 @@ func init() {
 }
 
 type ftcpListener struct {
-	conn     net.PacketConn
-	connChan chan net.Conn
-	errChan  chan error
-	connPool connPool
-	logger   logger.Logger
-	md       metadata
-	options  listener.Options
+	ln      net.Listener
+	logger  logger.Logger
+	md      metadata
+	options listener.Options
 }
 
 func NewListener(opts ...listener.Option) listener.Listener {
@@ -43,117 +39,36 @@ func (l *ftcpListener) Init(md md.Metadata) (err error) {
 		return
 	}
 
-	l.conn, err = tcpraw.Listen("tcp", l.options.Addr)
+	laddr, err := net.ResolveTCPAddr("tcp", l.options.Addr)
 	if err != nil {
 		return
 	}
 
-	l.connChan = make(chan net.Conn, l.md.connQueueSize)
-	l.errChan = make(chan error, 1)
-
-	go l.listenLoop()
+	var conn net.PacketConn
+	conn, err = tcpraw.Listen("tcp", l.options.Addr)
+	if err != nil {
+		return
+	}
+	conn = metrics.WrapPacketConn(l.options.Service, conn)
 
+	l.ln = udp.NewListener(
+		conn,
+		laddr,
+		l.md.backlog,
+		l.md.readQueueSize, l.md.readBufferSize,
+		l.md.ttl,
+		l.logger)
 	return
 }
 
 func (l *ftcpListener) Accept() (conn net.Conn, err error) {
-	var ok bool
-	select {
-	case conn = <-l.connChan:
-		conn = metrics.WrapConn(l.options.Service, conn)
-	case err, ok = <-l.errChan:
-		if !ok {
-			err = listener.ErrClosed
-		}
-	}
-	return
-}
-
-func (l *ftcpListener) Close() error {
-	err := l.conn.Close()
-	l.connPool.Range(func(k any, v *serverConn) bool {
-		v.Close()
-		return true
-	})
-	return err
+	return l.ln.Accept()
 }
 
 func (l *ftcpListener) Addr() net.Addr {
-	return l.conn.LocalAddr()
+	return l.ln.Addr()
 }
 
-func (l *ftcpListener) listenLoop() {
-	for {
-		b := make([]byte, l.md.readBufferSize)
-
-		n, raddr, err := l.conn.ReadFrom(b)
-		if err != nil {
-			l.logger.Error("accept:", err)
-			l.errChan <- err
-			close(l.errChan)
-			return
-		}
-
-		conn, ok := l.connPool.Get(raddr.String())
-		if !ok {
-			conn = newServerConn(l.conn, raddr,
-				&serverConnConfig{
-					ttl:   l.md.ttl,
-					qsize: l.md.readQueueSize,
-					onClose: func() {
-						l.connPool.Delete(raddr.String())
-					},
-				})
-
-			select {
-			case l.connChan <- conn:
-				l.connPool.Set(raddr.String(), conn)
-			default:
-				conn.Close()
-				l.logger.Error("connection queue is full")
-			}
-		}
-
-		if err := conn.send(b[:n]); err != nil {
-			l.logger.Warn("data discarded:", err)
-		}
-		l.logger.Debug("recv", n)
-	}
-}
-
-func (l *ftcpListener) parseMetadata(md md.Metadata) (err error) {
-	return
-}
-
-type connPool struct {
-	size int64
-	m    sync.Map
-}
-
-func (p *connPool) Get(key any) (conn *serverConn, ok bool) {
-	v, ok := p.m.Load(key)
-	if ok {
-		conn, ok = v.(*serverConn)
-	}
-	return
-}
-
-func (p *connPool) Set(key any, conn *serverConn) {
-	p.m.Store(key, conn)
-	atomic.AddInt64(&p.size, 1)
-}
-
-func (p *connPool) Delete(key any) {
-	p.m.Delete(key)
-	atomic.AddInt64(&p.size, -1)
-}
-
-func (p *connPool) Range(f func(key any, value *serverConn) bool) {
-	p.m.Range(func(k, v any) bool {
-		return f(k, v.(*serverConn))
-	})
-}
-
-func (p *connPool) Size() int64 {
-	return atomic.LoadInt64(&p.size)
+func (l *ftcpListener) Close() error {
+	return l.ln.Close()
 }

listener/ftcp/metadata.go

@@ -1,16 +1,16 @@
 package ftcp
 
-import "time"
+import (
+	"time"
+
+	mdata "github.com/go-gost/core/metadata"
+)
 
 const (
 	defaultTTL            = 60 * time.Second
 	defaultReadBufferSize = 1024
 	defaultReadQueueSize  = 128
-	defaultConnQueueSize  = 128
-)
-
-const (
-	addr = "addr"
+	defaultBacklog        = 128
 )
 
 type metadata struct {
@@ -18,5 +18,35 @@ type metadata struct {
 
 	readBufferSize int
 	readQueueSize  int
-	connQueueSize  int
+	backlog        int
+}
+
+func (l *ftcpListener) parseMetadata(md mdata.Metadata) (err error) {
+	const (
+		ttl            = "ttl"
+		readBufferSize = "readBufferSize"
+		readQueueSize  = "readQueueSize"
+		backlog        = "backlog"
+	)
+
+	l.md.ttl = mdata.GetDuration(md, ttl)
+	if l.md.ttl <= 0 {
+		l.md.ttl = defaultTTL
+	}
+	l.md.readBufferSize = mdata.GetInt(md, readBufferSize)
+	if l.md.readBufferSize <= 0 {
+		l.md.readBufferSize = defaultReadBufferSize
+	}
+
+	l.md.readQueueSize = mdata.GetInt(md, readQueueSize)
+	if l.md.readQueueSize <= 0 {
+		l.md.readQueueSize = defaultReadQueueSize
+	}
+
+	l.md.backlog = mdata.GetInt(md, backlog)
+	if l.md.backlog <= 0 {
+		l.md.backlog = defaultBacklog
+	}
+
+	return
 }