Go 语言连接器设计：模式选择与最佳实践

type Message struct {
    // 消息内容定义
    ID   string
    Data []byte
}

// Listen 监听入站消息。
// 入站消息将被发送到提供的通道。
func Listen(msgIn chan *Message) {
    // 内部逻辑：连接外部服务，接收数据，解析为Message，然后发送到msgIn
    go func() {
        for {
            // 模拟从外部服务接收消息
            // receivedMsg := readFromExternalService()
            // parsedMsg := parseToMessage(receivedMsg)
            // msgIn <- parsedMsg
            // 假设每秒接收一条消息
            msgIn <- &Message{ID: "inbound-" + "123", Data: []byte("hello from external")}
            // time.Sleep(time.Second)
        }
    }()
}

// Send 将消息发送到外部服务。
func Send(msgOut *Message) error {
    // 内部逻辑：将msgOut序列化并发送到外部服务
    // serializedMsg := serializeMessage(msgOut)
    // err := writeToExternalService(serializedMsg)
    // return err
    // 模拟发送成功
    // fmt.Printf("Sending message: %v\n", msgOut.ID)
    return nil
}

// ListenAndSend 监听入站消息并发送出站消息。
// 入站消息被发送到 msgIn 通道。
// 要发送消息，将消息放入 msgOut 通道。
func ListenAndSend(msgIn chan *Message, msgOut chan *Message) {
    // 内部逻辑：
    // 1. 启动 Goroutine 接收外部服务数据，解析后发送到 msgIn
    // 2. 启动 Goroutine 监听 msgOut，将消息序列化后发送到外部服务
    go func() {
        for {
            // 模拟从外部服务接收消息
            msgIn <- &Message{ID: "inbound-" + "456", Data: []byte("another external message")}
            // time.Sleep(time.Second)
        }
    }()

    go func() {
        for msg := range msgOut {
            // 模拟发送消息到外部服务
            // fmt.Printf("Sending message via channel: %v\n", msg.ID)
            // writeToExternalService(serializeMessage(msg))
        }
    }()
}

// Message 定义与之前相同
// type Message struct { ... }

// OnReceiveCallback 定义入站消息的回调函数签名。
// 如果回调函数返回 false，表示希望注销此回调。
type OnReceiveCallback func(*Message) bool

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

    // UnregisterOnReceive 移除一个已注册的回调函数。
    UnregisterOnReceive(callback OnReceiveCallback) // 可能需要某种ID或函数指针比较

    // Send 将消息发送到外部服务。
    // 此方法应确保是非阻塞的，内部通过 Goroutine 处理实际发送。
    Send(msg *Message) error

    // Start 启动连接器，开始监听和发送。
    Start() error

    // Stop 停止连接器并清理资源。
    Stop() error
}

// 实际的连接器实现 (简化版)
type MyConnector struct {
    // 内部管理回调函数列表
    callbacks []OnReceiveCallback
    // 内部用于发送的通道
    sendQueue chan *Message
    // 其他连接管理字段
}

func NewMyConnector() *MyConnector {
    return &MyConnector{
        callbacks: make([]OnReceiveCallback, 0),
        sendQueue: make(chan *Message, 100), // 缓冲通道防止阻塞
    }
}

func (c *MyConnector) RegisterOnReceive(callback OnReceiveCallback) {
    c.callbacks = append(c.callbacks, callback)
}

func (c *MyConnector) UnregisterOnReceive(callback OnReceiveCallback) {
    // 实际实现需要更复杂的逻辑来移除特定的回调函数，例如通过唯一ID或反射
    // 这里简化处理
}

func (c *MyConnector) Send(msg *Message) error {
    select {
    case c.sendQueue <- msg:
        return nil
    default:
        // 如果通道已满，则立即返回错误，避免阻塞调用者
        return fmt.Errorf("send queue is full, message %s dropped", msg.ID)
    }
}

func (c *MyConnector) Start() error {
    // 启动 Goroutine 监听外部服务并触发回调
    go func() {
        for {
            // 模拟从外部服务接收消息
            incomingMsg := &Message{ID: "inbound-" + "789", Data: []byte("message via callback")}
            for _, cb := range c.callbacks {
                if !cb(incomingMsg) {
                    // 如果回调返回 false，表示希望注销。
                    // 实际中需要更复杂的机制来安全地移除回调。
                }
            }
            // time.Sleep(time.Second)
        }
    }()

    // 启动 Goroutine 监听 sendQueue 并发送消息
    go func() {
        for msg := range c.sendQueue {
            // 模拟发送到外部服务
            // writeToExternalService(serializeMessage(msg))
            // fmt.Printf("Sending message via method: %v\n", msg.ID)
        }
    }()
    return nil
}

func (c *MyConnector) Stop() error {
    // 关闭通道，停止 Goroutine，清理资源
    close(c.sendQueue)
    return nil
}