package message_queue

import (
	"channel/ring_buffer"
	"context"
	"log"
	"sync"
)

type Message struct {
	content string
}

// MessageQueue channel实现消息队列 -- 生产者-消费者模型、发布订阅模式
type MessageQueue struct {
	chanList          map[string][]chan Message   // 切片的数量就是消费者的数量
	buffers           map[string]ring_buffer.Ring // 使用ring Buffer存储缓存
	bufferSize        int                         // 缓冲容量
	mutex             sync.RWMutex
	ringBufferCreator ring_buffer.RingBufferCreator
}

// Send 生产者发布消息
func (m *MessageQueue) Send(ctx context.Context, topic string, msg Message) error {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	if _, ok := m.chanList[topic]; !ok {
		// 不存在消费者，初始化
		m.chanList[topic] = make([]chan Message, 0)
	}
	if _, ok := m.buffers[topic]; !ok {
		// 不存在缓冲区，初始化
		m.buffers[topic] = m.ringBufferCreator(m.bufferSize)
	}

	// 消费者存在，就将消息发送给消费者
	for _, ch := range m.chanList[topic] {
		select {
		case ch <- msg:
		default: // 消费者阻塞则丢弃消息或记录日志
		}
	}

	if len(m.chanList[topic]) == 0 {
		return m.buffers[topic].Write(ctx, msg)
	}

	return nil
}

func (m *MessageQueue) distributeBufferMsg(topic string, ringBuffer ring_buffer.Ring) {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	for !ringBuffer.IsEmpty() {
		for _, ch := range m.chanList[topic] {
			msg, err := ringBuffer.Read()
			if err != nil {
				log.Println("messageQueue: ", err)
				return
			}

			select {
			case ch <- msg.(Message):
			default:
				//丢弃
			}
		}
	}

}

// Subscribe 消费者订阅消息 cap指缓存消息的容量；当消费者没有来得及取消息时，最多能缓存住cap条消息
// 每个消费者订阅时创建一个子channel
func (m *MessageQueue) Subscribe(topic string, cap int) <-chan Message {
	msgChan := make(chan Message, cap)
	m.mutex.Lock()
	m.chanList[topic] = append(m.chanList[topic], msgChan)

	// 如果缓冲区有数据需要把缓冲区的数据取出
	if ringBuffer, ok := m.buffers[topic]; ok && !ringBuffer.IsEmpty() {
		go m.distributeBufferMsg(topic, ringBuffer)
	}

	defer m.mutex.Unlock()
	return msgChan
}

func (m *MessageQueue) Close() {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	for _, topicList := range m.chanList {
		for _, ch := range topicList {
			close(ch)
		}
	}
}

type Opt func(*MessageQueue)

func WithRingBuffer(r ring_buffer.RingBufferCreator) Opt {
	return func(m *MessageQueue) {
		m.ringBufferCreator = r
	}
}

func NewMessageQueue(bufferSize int) *MessageQueue {
	return &MessageQueue{
		bufferSize:        bufferSize,
		chanList:          make(map[string][]chan Message),
		buffers:           make(map[string]ring_buffer.Ring),
		ringBufferCreator: ring_buffer.NewRingBuffer,
	}
}