feat(spmux): 自 go-hua 迁入单端口多路复用并解耦日志依赖

- go-hua/log 已弃用: *log.Logger 参数改为两方法接口 spmux.Logger (*logrus.Logger 天然满足), 传 nil 静默(logErrorf/logDebugf 判空) - 协议识别改用 zogo/httpprobe; 测试与例程同步适配 - 附测试、examples/spmux 例程(含 Logger 适配演示)与包 README
2026-09-20 12:37:10 +08:00
parent f120ab24ed
commit 96c63fb153
6 changed files with 557 additions and 0 deletions
@@ -0,0 +1,89 @@
+// spmux 包示例：单端口多路复用——同一端口上按报文特征把流量分发给 HTTP / HTTPS / TCP 三个 Listener
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	"git.zeroonesoft.cn/golib/zogo/spmux"
+)
+
+// stdoutLogger 适配 spmux.Logger 接口（*logrus.Logger 天然满足，无需适配）
+type stdoutLogger struct{}
+
+func (stdoutLogger) Debugf(format string, a ...any) { log.Printf("DEBUG "+format, a...) }
+func (stdoutLogger) Errorf(format string, a ...any) { log.Printf("ERROR "+format, a...) }
+
+func main() {
+	// 找一个空闲端口
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		panic(err)
+	}
+	port := l.Addr().(*net.TCPAddr).Port
+	l.Close()
+
+	// 创建端口复用器（logger 实现 spmux.Logger 即可，也可传 nil 静默）
+	pm := spmux.NewPortMux(port, stdoutLogger{})
+	if err := pm.Start(); err != nil {
+		panic(err)
+	}
+
+	// HTTP 流量交给 HttpListener
+	go http.Serve(pm.HttpListener(), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintln(w, "hello from http over spmux")
+	}))
+
+	// 其他 TCP 流量交给 TcpListener（echo 服务，回显内容并补换行）
+	tcpL := pm.TcpListener()
+	go func() {
+		for {
+			c, err := tcpL.Accept()
+			if err != nil {
+				return
+			}
+			go func(cc net.Conn) {
+				defer cc.Close()
+				buf := make([]byte, 1024)
+				for {
+					n, err := cc.Read(buf)
+					if n > 0 {
+						cc.Write(append(buf[:n:n], '\n'))
+					}
+					if err != nil {
+						return
+					}
+				}
+			}(c)
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// HTTP 请求会被识别并路由到 HttpListener
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/", port))
+	if err != nil {
+		panic(err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	fmt.Print("HTTP 响应:", string(body))
+
+	// 普通数据会被路由到 TcpListener
+	raw, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", port))
+	if err != nil {
+		panic(err)
+	}
+	fmt.Fprint(raw, "ping")
+	line, _ := bufio.NewReader(raw).ReadString('\n')
+	fmt.Print("TCP 响应:", line)
+	raw.Close()
+
+	_ = os.Stdout
+}
@@ -0,0 +1,47 @@
+# spmux
+
+单端口多路复用：同一端口上按报文特征（首部字节，基于 `zogo/httpprobe`）把流量
+分发给 HTTP / HTTPS / TCP 三个 `net.Listener`，适合"网页管理端 + 自有 TCP 协议"
+共存的单端口部署。
+
+> 迁移自 go-hua/spmux。原依赖的 go-hua/log 已弃用，日志改为两方法接口
+> `Logger`（`*logrus.Logger` 天然满足，可传 nil 静默）；协议识别改用 httpprobe。
+
+## 用法
+
+```go
+import (
+    "net"
+    "net/http"
+
+    "git.zeroonesoft.cn/golib/zogo/spmux"
+    "git.zeroonesoft.cn/golib/zogo/logger" // 可选：logrus 实例满足 Logger
+)
+
+pm := spmux.NewPortMux(8443, logger.Logrus) // 或 nil
+pm.Start()
+
+// HTTP(S) 流量分别交给两个 Listener（TLS 由消费方自行包裹）
+go http.Serve(pm.HttpListener(), mux)
+ln := pm.HttpsListener()
+
+// 其余流量走 TCP 自有协议
+go func() {
+    for {
+        c, err := ln.Accept() // 注意用 TcpListener 时见下方 Deprecated 说明
+        ...
+    }
+}()
+```
+
+完整可运行例程：[examples/spmux/main.go](../examples/spmux/main.go)
+
+## 注意
+
+- 协议识别需要先窥探首包（`FirstPeekSize` 起，最多读到 8 字节），消费方收到的
+  连接是 `PortConn`（已回放窥探字节），业务读取完整无感。
+- `TcpListener` 已标 Deprecated：RFB 等"服务端先发数据"的协议无法识别为 TCP；
+  这类端口不要与其他协议混用。
+- 分发等待消费方 Accept 的超时为 `AcceptTimeout`（100 秒），超时关闭连接。
+- 自 go-hua 迁移的消费方：`NewPortMux(port, *log.Logger)` →
+  `NewPortMux(port, spmux.Logger)`（接口），适配方式见例程 `stdoutLogger`。
@@ -0,0 +1,72 @@
+package spmux
+
+import (
+	"bufio"
+	"net"
+	"time"
+)
+
+// PortConn 复用器分发出来的连接，协议识别时窥探的数据保留在缓冲区中
+type PortConn struct {
+	conn   net.Conn
+	reader *bufio.Reader
+}
+
+// newPortConn 包装原始连接
+func newPortConn(conn net.Conn) *PortConn {
+	c := &PortConn{
+		conn: conn,
+	}
+	c.reader = bufio.NewReaderSize(conn, 0x1000)
+	return c
+}
+
+// Peek 窥探数据
+func (p *PortConn) Peek(n int) ([]byte, error) {
+	return p.reader.Peek(n)
+}
+
+// Buffered 从当前缓冲区读取的字节数
+func (p *PortConn) Buffered() int {
+	return p.reader.Buffered()
+}
+
+// Read 读取数据（先读出缓冲区中已窥探的数据）
+func (p *PortConn) Read(b []byte) (n int, err error) {
+	return p.reader.Read(b)
+}
+
+// Write 写入数据
+func (p *PortConn) Write(b []byte) (n int, err error) {
+	return p.conn.Write(b)
+}
+
+// Close 关闭连接
+func (p *PortConn) Close() error {
+	return p.conn.Close()
+}
+
+// LocalAddr 返回本地网络地址
+func (p *PortConn) LocalAddr() net.Addr {
+	return p.conn.LocalAddr()
+}
+
+// RemoteAddr 返回对端网络地址
+func (p *PortConn) RemoteAddr() net.Addr {
+	return p.conn.RemoteAddr()
+}
+
+// SetDeadline 设置读写超时时间
+func (p *PortConn) SetDeadline(t time.Time) error {
+	return p.conn.SetDeadline(t)
+}
+
+// SetReadDeadline 设置读超时时间
+func (p *PortConn) SetReadDeadline(t time.Time) error {
+	return p.conn.SetReadDeadline(t)
+}
+
+// SetWriteDeadline 设置写超时时间
+func (p *PortConn) SetWriteDeadline(t time.Time) error {
+	return p.conn.SetWriteDeadline(t)
+}
@@ -0,0 +1,53 @@
+package spmux
+
+import (
+	"errors"
+	"net"
+	"sync/atomic"
+)
+
+// PortListener 协议分发后的监听器，从连接通道接收连接
+type PortListener struct {
+	ch      chan *PortConn
+	addr    net.Addr
+	isClose int32
+}
+
+// NewPortListener 基于连接通道创建监听器
+func NewPortListener(connCh chan *PortConn, addr net.Addr) *PortListener {
+	return &PortListener{
+		ch:      connCh,
+		addr:    addr,
+		isClose: 0,
+	}
+}
+
+// Accept 等待并返回下一个分发过来的连接
+func (pListener *PortListener) Accept() (net.Conn, error) {
+	if atomic.LoadInt32(&pListener.isClose) == 1 {
+		return nil, errors.New("监听已经关闭")
+	}
+	if pListener.ch == nil {
+		return nil, errors.New("监听已经关闭")
+	}
+	conn := <-pListener.ch
+	if conn != nil {
+		return conn, nil
+	}
+	return nil, errors.New("the listener has closed")
+}
+
+// Close 关闭监听器
+func (pListener *PortListener) Close() error {
+	//close
+	if atomic.LoadInt32(&pListener.isClose) == 1 {
+		return errors.New("监听已经关闭")
+	}
+	atomic.StoreInt32(&pListener.isClose, 1)
+	return nil
+}
+
+// Addr 返回监听地址
+func (pListener *PortListener) Addr() net.Addr {
+	return pListener.addr
+}
@@ -0,0 +1,168 @@
+// Package spmux 单端口多路复用：同一端口上按报文特征把流量分发给 HTTP / HTTPS / TCP 三个 Listener。
+//
+// 迁移自 go-hua/spmux：原依赖的 go-hua/log 已弃用，日志改为两方法接口 Logger
+// （*logrus.Logger 天然满足），传 nil 时静默；协议识别改用 httpprobe 包。
+package spmux
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"git.zeroonesoft.cn/golib/zogo/httpprobe"
+)
+
+// Logger spmux 内部日志接口：仅需 Debugf / Errorf 两个方法。
+// *logrus.Logger（zogo/logger 初始化的实例）天然满足该接口。
+type Logger interface {
+	Debugf(format string, a ...any)
+	Errorf(format string, a ...any)
+}
+
+// AcceptTimeout 分发连接时等待消费方接收的超时（秒）
+const AcceptTimeout = 100
+
+// ChannelBufferSize 连接分发通道缓冲大小
+const ChannelBufferSize = 1
+
+// FirstPeekSize 协议识别时窥探的首包字节数
+const FirstPeekSize = 1
+
+// PortMux 单端口多路复用器，按协议特征将连接分发到 HTTP/HTTPS/TCP 三个通道
+type PortMux struct {
+	listener  net.Listener
+	logger    Logger
+	once      sync.Once
+	port      int
+	tcpConn   chan *PortConn
+	httpConn  chan *PortConn
+	httpsConn chan *PortConn
+}
+
+// NewPortMux 创建端口复用器，logger 用于内部错误输出，可为 nil（静默）
+func NewPortMux(port int, logger Logger) *PortMux {
+	p := &PortMux{
+		port:      port,
+		logger:    logger,
+		tcpConn:   make(chan *PortConn),
+		httpConn:  make(chan *PortConn),
+		httpsConn: make(chan *PortConn),
+	}
+	return p
+}
+
+// Start 开始监听指定端口并后台分发连接
+func (p *PortMux) Start() error {
+	addr := fmt.Sprintf("0.0.0.0:%d", p.port)
+	tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
+	if err != nil {
+		return err
+	}
+	p.listener, err = net.ListenTCP("tcp", tcpAddr)
+	if err != nil {
+		return err
+	}
+	go p.listen()
+	return nil
+}
+
+// listen 接受连接循环
+func (p *PortMux) listen() {
+	for {
+		conn, err := p.listener.Accept()
+		if err != nil {
+			p.logErrorf("接受连接失败 端口:%d 错误:%s", p.port, err.Error())
+			p.Close()
+			break
+		}
+		p.logDebugf("新连接 地址:%s", conn.RemoteAddr().String())
+		go p.process(conn)
+	}
+}
+
+// process 识别单个连接的协议并分发到对应通道
+func (p *PortMux) process(conn net.Conn) {
+	newConn := newPortConn(conn)
+	buf, err := newConn.Peek(FirstPeekSize)
+	if err != nil {
+		conn.Close()
+		return
+	}
+
+	n := newConn.Buffered()
+	if n >= 8 {
+		buf, err = newConn.Peek(n)
+		if err != nil {
+			conn.Close()
+			return
+		}
+	}
+
+	if n >= 8 && httpprobe.IsHttp(buf) {
+		timer := time.NewTimer(AcceptTimeout)
+		select {
+		case <-timer.C:
+			conn.Close()
+			p.logErrorf("接受超时 端口:%d 协议:%s", p.port, "http")
+		case p.httpConn <- newConn:
+		}
+	} else if n >= 8 && httpprobe.IsHttps(buf) {
+		timer := time.NewTimer(AcceptTimeout)
+		select {
+		case <-timer.C:
+			conn.Close()
+			p.logErrorf("接受超时 端口:%d 协议:%s", p.port, "https")
+		case p.httpsConn <- newConn:
+		}
+	} else {
+		timer := time.NewTimer(AcceptTimeout)
+		select {
+		case <-timer.C:
+			conn.Close()
+			p.logErrorf("接受超时 端口:%d 协议:%s", p.port, "tcp")
+		case p.tcpConn <- newConn:
+		}
+	}
+}
+
+// Close 关闭监听与所有分发通道（幂等）
+func (p *PortMux) Close() (err error) {
+	p.once.Do(func() {
+		close(p.tcpConn)
+		close(p.httpsConn)
+		close(p.httpConn)
+		err = p.listener.Close()
+	})
+	return
+}
+
+// TcpListener TCP监听
+// Deprecated: 一些网络通讯(RFB协议)是由服务器先发送数据的，会导致无法识别成TCP
+func (p *PortMux) TcpListener() net.Listener {
+	return NewPortListener(p.tcpConn, p.listener.Addr())
+}
+
+// HttpListener 返回 HTTP 流量的监听器
+func (p *PortMux) HttpListener() net.Listener {
+	return NewPortListener(p.httpConn, p.listener.Addr())
+}
+
+// HttpsListener 返回 HTTPS 流量的监听器
+func (p *PortMux) HttpsListener() net.Listener {
+	return NewPortListener(p.httpsConn, p.listener.Addr())
+}
+
+// logErrorf nil 安全的错误日志
+func (p *PortMux) logErrorf(format string, a ...any) {
+	if p.logger != nil {
+		p.logger.Errorf(format, a...)
+	}
+}
+
+// logDebugf nil 安全的调试日志
+func (p *PortMux) logDebugf(format string, a ...any) {
+	if p.logger != nil {
+		p.logger.Debugf(format, a...)
+	}
+}
@@ -0,0 +1,128 @@
+package spmux
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+)
+
+// freePort 找一个空闲端口（有轻微竞态，测试环境可接受）
+func freePort(t *testing.T) int {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	port := l.Addr().(*net.TCPAddr).Port
+	l.Close()
+	return port
+}
+
+func TestPortMuxHttpRouting(t *testing.T) {
+	port := freePort(t)
+
+	pm := NewPortMux(port, nil)
+	if err := pm.Start(); err != nil {
+		t.Fatalf("Start: %v", err)
+	}
+
+	httpDone := make(chan struct{})
+	go func() {
+		defer close(httpDone)
+		http.Serve(pm.HttpListener(), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprint(w, "mux-http-ok")
+		}))
+	}()
+	// 先 Close 解除 Serve 阻塞，再等待协程退出（顺序不能反，否则死锁）
+	defer func() {
+		pm.Close()
+		<-httpDone
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/", port))
+	if err != nil {
+		t.Fatalf("HTTP 请求失败: %v", err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	if string(body) != "mux-http-ok" {
+		t.Errorf("HTTP 响应 = %q", string(body))
+	}
+}
+
+func TestPortMuxTcpRouting(t *testing.T) {
+	port := freePort(t)
+
+	pm := NewPortMux(port, nil)
+	if err := pm.Start(); err != nil {
+		t.Fatalf("Start: %v", err)
+	}
+
+	tcpDone := make(chan struct{})
+	go func() {
+		defer close(tcpDone)
+		for {
+			c, err := pm.TcpListener().Accept()
+			if err != nil {
+				return
+			}
+			go func(cc net.Conn) {
+				defer cc.Close()
+				buf := make([]byte, 1024)
+				for {
+					n, err := cc.Read(buf)
+					if n > 0 {
+						cc.Write(append(buf[:n:n], '\n'))
+					}
+					if err != nil {
+						return
+					}
+				}
+			}(c)
+		}
+	}()
+	// 先 Close 解除 Accept 阻塞，再等待协程退出（顺序不能反，否则死锁）
+	defer func() {
+		pm.Close()
+		<-tcpDone
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// 非 HTTP 数据（超过 8 字节）应路由到 TcpListener
+	raw, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", port))
+	if err != nil {
+		t.Fatalf("TCP 连接失败: %v", err)
+	}
+	fmt.Fprint(raw, "ping-ping")
+	raw.SetReadDeadline(time.Now().Add(3 * time.Second))
+	line, err := bufio.NewReader(raw).ReadString('\n')
+	if err != nil {
+		t.Fatalf("读取 echo: %v", err)
+	}
+	if line != "ping-ping\n" {
+		t.Errorf("echo = %q", line)
+	}
+	raw.Close()
+}
+
+func TestPortListenerAddr(t *testing.T) {
+	port := freePort(t)
+	pm := NewPortMux(port, nil)
+	if err := pm.Start(); err != nil {
+		t.Fatalf("Start: %v", err)
+	}
+	defer pm.Close()
+
+	if pm.HttpListener().Addr().String() == "" {
+		t.Error("HttpListener Addr 不应为空")
+	}
+	if pm.TcpListener().Addr().String() == "" {
+		t.Error("TcpListener Addr 不应为空")
+	}
+}