Service.go

package eventservice

import (
	"context"
	"encoding/json"

	"code.fbi.h-da.de/danet/gosdn/controller/config"
	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
	"code.fbi.h-da.de/danet/gosdn/controller/event"

	interfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/event"

	amqp "github.com/rabbitmq/amqp091-go"
	log "github.com/sirupsen/logrus"
)

// EventService is used to setup a connection to a broker and publish events to topics.
type EventService struct {
	connection *amqp.Connection
	channel    *amqp.Channel
}

// NewEventService creates a new connection to the broker and opens a channel for later usage.
func NewEventService() (interfaces.Service, error) {
	// TODO: This is an fugly hack to mitigate that some tests are trying to connect to an actual broker. (staester)
	if config.AMQPPrefix == "" {
		return NewMockEventService(), nil
	}

	conn, err := amqp.Dial(amqpURIBuilder(config.AMQPPrefix, config.AMQPUser, config.AMQPPassword, config.AMQPHost, config.AMQPPort))
	if err != nil {
		return nil, &customerrs.AMQPInitFailError{Action: "failed to connect to RabbitMQ", Err: err}
	}

	ch, err := conn.Channel()
	if err != nil {
		return nil, &customerrs.AMQPInitFailError{Action: "failed to open a channel", Err: err}
	}

	return &EventService{
		connection: conn,
		channel:    ch,
	}, nil
}

// PublishEvent declares a queue and publishes events.
func (e *EventService) PublishEvent(topic string, event event.Event) error {
	q, err := e.channel.QueueDeclare(
		topic, // name
		false, // durable
		false, // delete when unused
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		return &customerrs.AMQPInitFailError{Action: "failed declaring queue", Err: err}
	}

	eventBody, err := json.Marshal(event)
	if err != nil {
		return &customerrs.CouldNotMarshallError{Identifier: topic + " " + event.EntityID.String(), Type: event.Type, Err: err}
	}

	err = e.channel.PublishWithContext(
		context.TODO(),
		"",     // exchange
		q.Name, // routing key
		false,  // mandatory
		false,  // immediate
		amqp.Publishing{
			ContentType: "application/json",
			Body:        eventBody,
		})
	if err != nil {
		return &customerrs.AMQPMessageFailError{Action: "failed to publish message", Err: err}
	}

	return nil
}

// CloseConnection closes an exisiting connection.
func (e *EventService) CloseConnection() {
	if err := e.connection.Close(); err != nil {
		log.Error(err)
	}
}