utils.go

package amqp

import oAmqp "github.com/streadway/amqp"

func (c *Connection) ReadQueue(queueName string, out *chan oAmqp.Delivery, prefetchCount, prefetchSize int, global bool) (*Queue, error) {
	queue := &Queue{
		PrefetchCount: prefetchCount,
		PrefetchSize:  prefetchSize,
		Global:        global,
		connection:    c,
		outChan:       out,
		action:        actionRead,
		queueName:     queueName,
	}

	c.queues = append(c.queues, queue)

	if err := queue.run(); err != nil {
		return nil, err
	}

	return queue, nil
}

func (c *Connection) WriteQueue() (*Queue, error) {
	queue := &Queue{
		PrefetchCount: 1,
		PrefetchSize:  0,
		Global:        false,
		connection:    c,
		action:        actionWrite,
	}

	c.queues = append(c.queues, queue)

	if err := queue.run(); err != nil {
		return nil, err
	}

	return queue, nil
}

func (c *Connection) ReadPriorityQueue(queueName string, callback func(msg oAmqp.Delivery) error, prefetchCount, prefetchSize int, global bool) (*PriorityQueue, error) {
	queue := &PriorityQueue{
		Queue: &Queue{
			PrefetchCount: prefetchCount,
			PrefetchSize:  prefetchSize,
			Global:        global,
			connection:    c,
			action:        actionRead,
			queueName:     queueName,
		},
		callback: callback,
	}

	c.queues = append(c.queues, queue.Queue)

	if err := queue.run(); err != nil {
		return nil, err
	}

	return queue, nil
}

func (c *Connection) WritePriorityQueue() (*PriorityQueue, error) {
	queue := &PriorityQueue{
		Queue: &Queue{
			PrefetchCount: 1,
			PrefetchSize:  0,
			Global:        false,
			connection:    c,
			action:        actionWrite,
		},
	}

	c.queues = append(c.queues, queue.Queue)

	if err := queue.run(); err != nil {
		return nil, err
	}

	return queue, nil
}