package util

import (
	"errors"
	amqp "github.com/rabbitmq/amqp091-go"
	"sync"
	"time"
)

var ErrAmqpChannelClosed = errors.New(`channel is closed`)

type AmqpClient struct {
	channelPool sync.Pool
	Connection  *amqp.Connection
	url         string
	lock        sync.RWMutex
}

// NewAmqpClient 新的连接池
func NewAmqpClient(url string) (*AmqpClient, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, err
	}
	ac := &AmqpClient{
		url:        url,
		Connection: conn,
	}
	ac.channelPool = sync.Pool{New: func() any {
		var err error
		var ch *amqp.Channel
		if ac.Connection.IsClosed() {
			ac.lock.Lock()
			ac.Connection, err = amqp.Dial(ac.url)
			ac.lock.Unlock()
			if err != nil {
				return err
			}
		}
		ac.lock.RLock()
		defer ac.lock.RUnlock()
		if err = Retry(5, 5*time.Microsecond, func() error {
			if ch, err = ac.Connection.Channel(); err != nil {
				return err
			}
			return nil
		}); err != nil {
			return err
		}
		return ch
	}}
	go func(c *AmqpClient) {
		ch, _ := c.GetChannel()
		time.Sleep(3 * time.Second)
		c.PutChannel(ch)
	}(ac)
	return ac, nil
}

// Close 关闭连接
func (ac *AmqpClient) Close() error {
	if ac.Connection.IsClosed() {
		return nil
	}
	return ac.Connection.Close()
}

// GetChannel 获取一个连接
func (ac *AmqpClient) GetChannel() (*amqp.Channel, error) {
	var ch *amqp.Channel
	var ok bool
	var err error
	if err = Retry(5, 5*time.Microsecond, func() error {
		ich := ac.channelPool.Get()
		if ch, ok = ich.(*amqp.Channel); ok {
			if ch.IsClosed() {
				return ErrAmqpChannelClosed
			}
		} else if err, ok = ich.(error); ok {
			return err
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return ch, nil
}

// PutChannel 归还一个连接
func (ac *AmqpClient) PutChannel(ch *amqp.Channel) {
	if ch == nil || ch.IsClosed() {
		return
	}
	ac.channelPool.Put(ch)
}