topologyService.go

package topology

import (
	"context"

	"code.fbi.h-da.de/danet/gosdn/controller/event"
	eventInterfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/event"
	query "code.fbi.h-da.de/danet/gosdn/controller/store"
	"code.fbi.h-da.de/danet/gosdn/controller/topology/links"
	"code.fbi.h-da.de/danet/gosdn/controller/topology/nodes"
	"code.fbi.h-da.de/danet/gosdn/controller/topology/ports"
	log "github.com/sirupsen/logrus"
)

const (
	// LinkEventTopic is the used topic for link related entity changes.
	LinkEventTopic = "link"
)

// Service defines an interface for a Service.
type Service interface {
	AddLink(links.Link) error
	UpdateLink(links.Link) error
	DeleteLink(links.Link) error
	Get(query.Query) (links.Link, error)
	GetAll() ([]links.Link, error)
}

// TopoService manages the links between nodes and their related ports.
type TopoService struct {
	store        Store
	nodeService  nodes.Service
	portService  ports.Service
	eventService eventInterfaces.Service
}

// NewTopologyService creates a new TopologyService.
func NewTopologyService(
	store Store,
	nodeService nodes.Service,
	portService ports.Service,
	eventService eventInterfaces.Service,
) Service {
	return &TopoService{
		store:        store,
		nodeService:  nodeService,
		portService:  portService,
		eventService: eventService,
	}
}

// AddLink adds a new link to the topology.
func (t *TopoService) AddLink(link links.Link) error {
	ctx := context.Background()

	err := t.store.Add(ctx, link)
	if err != nil {
		return err
	}

	pubEvent := event.NewAddEvent(link.ID)
	if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
		go func() {
			t.eventService.Reconnect()

			retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
			if retryErr != nil {
				log.Error(retryErr)
			}
		}()
	}

	return nil
}

// UpdateLink updates an existing link.
func (t *TopoService) UpdateLink(link links.Link) error {
	ctx := context.Background()
	err := t.store.Update(ctx, link)
	if err != nil {
		return err
	}

	pubEvent := event.NewUpdateEvent(link.ID)
	if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
		go func() {
			t.eventService.Reconnect()

			retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
			if retryErr != nil {
				log.Error(retryErr)
			}
		}()
	}

	return nil
}

// DeleteLink deletes a link.
func (t *TopoService) DeleteLink(link links.Link) error {
	ctx := context.Background()
	// TODO: Delete should also check if a node or port is used somewhere else and
	// if not, delete the node and its ports
	err := t.store.Delete(ctx, link)
	if err != nil {
		return err
	}

	pubEvent := event.NewDeleteEvent(link.ID)
	if err := t.eventService.PublishEvent(LinkEventTopic, pubEvent); err != nil {
		go func() {
			t.eventService.Reconnect()

			retryErr := t.eventService.RetryPublish(LinkEventTopic, pubEvent)
			if retryErr != nil {
				log.Error(retryErr)
			}
		}()
	}

	return nil
}

// GetAll returns the current topology.
func (t *TopoService) GetAll() ([]links.Link, error) {
	ctx := context.Background()
	topo, err := t.store.GetAll(ctx)
	if err != nil {
		return topo, err
	}
	return topo, nil
}

// Get returns the current topology.
func (t *TopoService) Get(query query.Query) (links.Link, error) {
	ctx := context.Background()
	link, err := t.store.Get(ctx, query)
	if err != nil {
		return link, err
	}

	return link, nil
}