Go 连接器设计模式：通道、回调与实践考量

package connector

type Message struct {
    // 消息内容定义
}

// Connector 接口定义
type Connector interface {
    // Listen 启动监听入站消息。
    // 入站消息将被发送到提供的 msg 通道。
    // 通常在后台 goroutine 中运行。
    Listen(msg chan<- *Message) error

    // Send 将消息发送到外部服务。
    // 此方法应确保非阻塞或可控阻塞。
    Send(msg *Message) error

    // Close 关闭连接器并清理资源。
    Close() error
}

// 示例实现
type MyConnector struct {
    // 内部连接管理字段
}

func NewMyConnector() *MyConnector {
    return &MyConnector{}
}

func (c *MyConnector) Listen(msg chan<- *Message) error {
    // 启动 goroutine 监听外部服务
    go func() {
        defer close(msg) // 监听结束时关闭通道
        for {
            // 模拟从外部服务接收数据
            // parsedMsg := parseExternalData()
            // msg <- parsedMsg
            // if connectionClosed { break }
        }
    }()
    return nil
}

func (c *MyConnector) Send(msg *Message) error {
    // 模拟发送消息到外部服务
    // sendToExternalService(msg)
    return nil
}

func (c *MyConnector) Close() error {
    // 关闭连接
    return nil
}

package connector

type Message struct {
    // 消息内容定义
}

// Connector 接口定义
type Connector interface {
    // ListenAndSend 启动连接器，同时处理入站和出站消息。
    // 入站消息将被发送到 msgIn 通道。
    // 要发送出站消息，将消息放入 msgOut 通道。
    // 此方法通常在后台 goroutine 中运行。
    ListenAndSend(msgIn chan<- *Message, msgOut <-chan *Message) error

    // Close 关闭连接器并清理资源。
    Close() error
}

// 示例实现
type MyBidirectionalConnector struct {
    // 内部连接管理字段
}

func NewMyBidirectionalConnector() *MyBidirectionalConnector {
    return &MyBidirectionalConnector{}
}

func (c *MyBidirectionalConnector) ListenAndSend(msgIn chan<- *Message, msgOut <-chan *Message) error {
    go func() {
        defer close(msgIn) // 入站通道在连接器关闭时关闭
        for {
            select {
            case incoming := <- /* 模拟从外部服务接收数据 */ :
                // parsedMsg := parseExternalData(incoming)
                // msgIn <- parsedMsg
            case outgoing := <-msgOut:
                // 模拟发送消息到外部服务
                // sendToExternalService(outgoing)
            // case <-c.stopChan: // 停止信号
            //     return
            }
        }
    }()
    return nil
}

func (c *MyBidirectionalConnector) Close() error {
    // 关闭连接
    return nil
}

package connector

type Message struct {
    // 消息内容定义
}

// OnReceiveCallback 定义入站消息的回调函数。
// 如果回调返回 false，表示该回调应被注销。
type OnReceiveCallback func(*Message) bool

// Connector 接口定义
type Connector interface {
    // RegisterOnReceive 注册一个回调函数来处理入站消息。
    // 可以注册多个回调。
    RegisterOnReceive(callback OnReceiveCallback)

    // Send 将消息发送到外部服务。
    Send(msg *Message) error

    // Close 关闭连接器并清理资源。
    Close() error
}

// 示例实现
type MyCallbackConnector struct {
    callbacks []OnReceiveCallback
    mu        sync.RWMutex // 保护 callbacks 列表
    // 内部连接管理字段
}

func NewMyCallbackConnector() *MyCallbackConnector {
    return &MyCallbackConnector{}
}

func (c *MyCallbackConnector) RegisterOnReceive(callback OnReceiveCallback) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.callbacks = append(c.callbacks, callback)
}

func (c *MyCallbackConnector) Send(msg *Message) error {
    // 模拟发送消息到外部服务
    return nil
}

func (c *MyCallbackConnector) Close() error {
    // 关闭连接
    return nil
}

// 假设有一个内部 goroutine 负责接收和分发消息
func (c *MyCallbackConnector) runReceiver() {
    for {
        // 模拟接收到消息
        // receivedMsg := receiveFromExternalService()

        c.mu.RLock()
        var activeCallbacks []OnReceiveCallback
        for _, cb := range c.callbacks {
            // if cb(receivedMsg) { // 实际调用回调
            //     activeCallbacks = append(activeCallbacks, cb)
            // }
        }
        c.callbacks = activeCallbacks // 移除返回 false 的回调
        c.mu.RUnlock()
    }
}