clawgo/pkg/bus/bus.go

package bus

import (
	"clawgo/pkg/logger"
	"context"
	"sync"
	"time"
)

type MessageBus struct {
	inbound  chan InboundMessage
	outbound chan OutboundMessage
	handlers map[string]MessageHandler
	mu       sync.RWMutex
}

const queueWriteTimeout = 2 * time.Second

func NewMessageBus() *MessageBus {
	return &MessageBus{
		inbound:  make(chan InboundMessage, 100),
		outbound: make(chan OutboundMessage, 100),
		handlers: make(map[string]MessageHandler),
	}
}

func (mb *MessageBus) PublishInbound(msg InboundMessage) {
	select {
	case mb.inbound <- msg:
	case <-time.After(queueWriteTimeout):
		logger.ErrorCF("bus", "PublishInbound timeout (queue full)", map[string]interface{}{
			"channel":     msg.Channel,
			"chat_id":     msg.ChatID,
			"session_key": msg.SessionKey,
		})
	}
}

func (mb *MessageBus) ConsumeInbound(ctx context.Context) (InboundMessage, bool) {
	select {
	case msg := <-mb.inbound:
		return msg, true
	case <-ctx.Done():
		return InboundMessage{}, false
	}
}

func (mb *MessageBus) PublishOutbound(msg OutboundMessage) {
	select {
	case mb.outbound <- msg:
	case <-time.After(queueWriteTimeout):
		logger.ErrorCF("bus", "PublishOutbound timeout (queue full)", map[string]interface{}{
			"channel": msg.Channel,
			"chat_id": msg.ChatID,
		})
	}
}

func (mb *MessageBus) SubscribeOutbound(ctx context.Context) (OutboundMessage, bool) {
	select {
	case msg := <-mb.outbound:
		return msg, true
	case <-ctx.Done():
		return OutboundMessage{}, false
	}
}

func (mb *MessageBus) RegisterHandler(channel string, handler MessageHandler) {
	mb.mu.Lock()
	defer mb.mu.Unlock()
	mb.handlers[channel] = handler
}

func (mb *MessageBus) GetHandler(channel string) (MessageHandler, bool) {
	mb.mu.RLock()
	defer mb.mu.RUnlock()
	handler, ok := mb.handlers[channel]
	return handler, ok
}

func (mb *MessageBus) Close() {
	close(mb.inbound)
	close(mb.outbound)
}