queue.go

package amqp

import (
	"github.com/streadway/amqp"
	"log"
	"time"
)

const (
	actionRead  = "read queue"
	actionWrite = "write queue"
)

type Queue struct {
	PrefetchCount int
	PrefetchSize  int
	Global        bool
	outChan       *chan amqp.Delivery
	action        string
	queueName     string

	channel    *amqp.Channel
	connection *Connection
}

func (c *Queue) openChannel() (*amqp.Channel, error) {
	if c.channel != nil {
		c.channel.Close()
	}

	channel, err := c.connection.amqpConnection.Channel()
	if err != nil {
		return nil, err
	}
	if err := channel.Qos(c.PrefetchCount, c.PrefetchSize, c.Global); err != nil {
		defer channel.Close()
		return nil, err
	}

	return channel, nil
}

func (q *Queue) run() error {
	channel, err := q.openChannel()
	if err != nil {
		return err
	}
	q.channel = channel

	if q.action == actionRead {
		out, err := q.channel.Consume(q.queueName, "", false, false, false, false, nil)
		if err != nil {
			return err
		}

		// Прослойка для отслеживания падения queue
		go func(q *Queue, out <-chan amqp.Delivery) {
			for {
				select {
				case msg, ok := <-out:
					if !ok {
						if q.connection.amqpConnection.IsClosed() {
							return
						}
						go func(q *Queue) {
							for {
								if err := q.run(); err != nil {
									time.Sleep(500 * time.Millisecond)
									log.Printf("rerun queue error: %v\n", err)
									continue
								}
								break
							}
						}(q)
						return
					}
					*q.outChan <- msg
				}
			}
		}(q, out)
	}

	return nil
}

func (q Queue) stop() error {
	if q.channel != nil {
		q.channel.Close()
		q.channel = nil
	}
	return nil
}

func (q *Queue) Send(exchange string, route string, params amqp.Publishing) error {
	var err error
	err = q.channel.Publish(
		exchange,
		route,
		false,
		false,
		params,
	)
	if err != nil {
		return err
	}
	return err
}