broker.go

package brokerdriver

import (
	"os"
	"query-service/internal/adapters/brokeradapter"
	"query-service/pkg/errorhandler"

	"strconv"
	"time"

	"github.com/thijsheijden/alice"
)

// Driver models an Alice RabbitMQ broker
type Driver struct {
	broker  alice.Broker
	gateway brokeradapter.GatewayInterface
}

// CreateAliceBroker creates an Alice broker
func CreateAliceBroker(gateway brokeradapter.GatewayInterface) *Driver {

	// Create connection config using environment variables
	rabbitUser := os.Getenv("RABBIT_USER")
	rabbitPassword := os.Getenv("RABBIT_PASSWORD")
	rabbitHost := os.Getenv("RABBIT_HOST")
	rabbitPort, err := strconv.Atoi(os.Getenv("RABBIT_PORT"))
	errorhandler.FailWithError(err, "invalid rabbitmq port given")

	config := alice.CreateConfig(rabbitUser, rabbitPassword, rabbitHost, rabbitPort, true, time.Minute*1, alice.DefaultErrorHandler)

	return &Driver{
		broker:  alice.CreateBroker(config),
		gateway: gateway,
	}
}

// CreateConsumer creates an AliceConsumer on a certain exchange and queue
func (d *Driver) CreateConsumer() Consumer {
	exchangeID := "query-requests"
	routingKey := "aql-user-request"

	// Declare the exchange we want to bind to
	exchange, err := alice.CreateDefaultExchange(exchangeID, alice.Direct)
	if err != nil {
		errorhandler.FailWithError(err, "failed to create exchange")
	}

	// Declare the queue we will consume from
	queue := alice.CreateQueue(exchange, "aql-query-queue", true, false, true, false, nil)

	// Create the consumer
	c, err := d.broker.CreateConsumer(queue, routingKey, alice.DefaultConsumerErrorHandler)
	if err != nil {
		errorhandler.FailWithError(err, "failed to create consumer")
	}

	consumer := &AliceConsumer{
		broker:   d,
		consumer: c,
	}

	return consumer
}

// CreateProducer creates an AliceProducer on a certain exchange
func (d *Driver) CreateProducer() Producer {
	exchange, err := alice.CreateDefaultExchange("ui-direct-exchange", alice.Direct)
	if err != nil {
		errorhandler.FailWithError(err, "failed to create exchange for producer")
	}

	p, err := d.broker.CreateProducer(exchange, alice.DefaultProducerErrorHandler)
	if err != nil {
		errorhandler.FailWithError(err, "failed to created producer")
	}

	producer := &AliceProducer{
		broker:   d,
		producer: p,
	}

	return producer
}