diff --git a/controller/api/initialise_test.go b/controller/api/initialise_test.go
index a6bd57aa0c63e059badbd98336c6bea0960798a8..8c0dfabb393604ea5d2ed3f167710dd5f45a38b6 100644
--- a/controller/api/initialise_test.go
+++ b/controller/api/initialise_test.go
@@ -157,9 +157,17 @@ func bootstrapUnitTest() {
portService := ports.NewPortService(portStore, eventService)
topoloyStore := store.NewGenericStore[links.Link]()
- topologyService := topology.NewTopologyService(topoloyStore, nodeService, portService)
-
- northbound := nbi.NewNBI(pndStore, userService, roleService, *jwtManager, topologyService, nodeService, portService)
+ topologyService := topology.NewTopologyService(topoloyStore, nodeService, portService, eventService)
+
+ northbound := nbi.NewNBI(
+ pndStore,
+ userService,
+ roleService,
+ *jwtManager,
+ topologyService,
+ nodeService,
+ portService,
+ )
cpb.RegisterCoreServiceServer(s, northbound.Core)
ppb.RegisterPndServiceServer(s, northbound.Pnd)
diff --git a/controller/controller.go b/controller/controller.go
index 044f9d1b45a8e4dd03989c2561591aa7bbc07144..387bae4769b833b126c8a386fbf9eb067c0cf1ce 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -88,6 +88,7 @@ func initialize() error {
topology.NewDatabaseTopologyStore(),
nodeService,
portService,
+ eventService,
),
nodeService: nodeService,
portService: portService,
diff --git a/controller/topology/topologyService.go b/controller/topology/topologyService.go
index 67b40a006a9d2791cf61c17be5c862146da9d34e..78ab003d73085cd34558159d16e5c02a81200f70 100644
--- a/controller/topology/topologyService.go
+++ b/controller/topology/topologyService.go
@@ -1,12 +1,19 @@
package topology
import (
+ "code.fbi.h-da.de/danet/gosdn/controller/event"
+ eventInterfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/event"
"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"
"code.fbi.h-da.de/danet/gosdn/controller/topology/store"
)
+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
@@ -18,9 +25,10 @@ type Service interface {
// service is a service for ports
type service struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
// NewTopologyService creates a new TopologyService
@@ -28,11 +36,13 @@ func NewTopologyService(
store Store,
nodeService nodes.Service,
portService ports.Service,
+ eventService eventInterfaces.Service,
) Service {
return &service{
- store: store,
- nodeService: nodeService,
- portService: portService,
+ store: store,
+ nodeService: nodeService,
+ portService: portService,
+ eventService: eventService,
}
}
@@ -63,6 +73,8 @@ func (t *service) AddLink(link links.Link) error {
return err
}
+ t.eventService.PublishEvent(LinkEventTopic, event.NewAddEvent(link.ID))
+
return nil
}
@@ -73,6 +85,8 @@ func (t *service) UpdateLink(link links.Link) error {
return err
}
+ t.eventService.PublishEvent(LinkEventTopic, event.NewUpdateEvent(link.ID))
+
return nil
}
@@ -83,6 +97,8 @@ func (t *service) DeleteLink(link links.Link) error {
return err
}
+ t.eventService.PublishEvent(LinkEventTopic, event.NewDeleteEvent(link.ID))
+
return nil
}
diff --git a/controller/topology/topologyService_test.go b/controller/topology/topologyService_test.go
index b0f6e379b37ae69de662c178eae35681eacdf8d3..71955c6d74fbd5bdf048734ff285d728f0ea10e9 100644
--- a/controller/topology/topologyService_test.go
+++ b/controller/topology/topologyService_test.go
@@ -5,6 +5,7 @@ import (
"testing"
eventservice "code.fbi.h-da.de/danet/gosdn/controller/eventService"
+ eventInterfaces "code.fbi.h-da.de/danet/gosdn/controller/interfaces/event"
"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"
@@ -97,9 +98,10 @@ func getTestStoreWithPorts(t *testing.T, portsToAdd []ports.Port) ports.Store {
func TestNewTopologyService(t *testing.T) {
type args struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
tests := []struct {
name string
@@ -109,20 +111,27 @@ func TestNewTopologyService(t *testing.T) {
{
name: "should create a new topology service",
args: args{
- store: getTestStoreWithLinks(t, []links.Link{}),
- nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
- portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ store: getTestStoreWithLinks(t, []links.Link{}),
+ nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
+ portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ eventService: eventservice.NewMockEventService(),
},
want: NewTopologyService(
getTestStoreWithLinks(t, []links.Link{}),
nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ eventservice.NewMockEventService(),
),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- if got := NewTopologyService(tt.args.store, tt.args.nodeService, tt.args.portService); !reflect.DeepEqual(got, tt.want) {
+ if got := NewTopologyService(
+ tt.args.store,
+ tt.args.nodeService,
+ tt.args.portService,
+ tt.args.eventService,
+ ); !reflect.DeepEqual(got, tt.want) {
t.Errorf("NewNodeService() = %v, want %v", got, tt.want)
}
})
@@ -131,9 +140,10 @@ func TestNewTopologyService(t *testing.T) {
func TestTopologyService_AddLink(t *testing.T) {
type fields struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
type args struct {
link links.Link
@@ -163,6 +173,7 @@ func TestTopologyService_AddLink(t *testing.T) {
),
eventservice.NewMockEventService(),
),
+ eventService: eventservice.NewMockEventService(),
},
args: args{
link: getTestLink(),
@@ -174,9 +185,10 @@ func TestTopologyService_AddLink(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
p := &service{
- store: tt.fields.store,
- nodeService: tt.fields.nodeService,
- portService: tt.fields.portService,
+ store: tt.fields.store,
+ nodeService: tt.fields.nodeService,
+ portService: tt.fields.portService,
+ eventService: tt.fields.eventService,
}
err := p.AddLink(tt.args.link)
if (err != nil) != tt.wantErr {
@@ -198,9 +210,10 @@ func TestTopologyService_AddLink(t *testing.T) {
func TestTopologyService_Update(t *testing.T) {
type fields struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
type args struct {
link links.Link
@@ -215,9 +228,10 @@ func TestTopologyService_Update(t *testing.T) {
{
name: "should update an existing link",
fields: fields{
- store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
- nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
- portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
+ nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
+ portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ eventService: eventservice.NewMockEventService(),
},
args: args{
link: getTestLink(),
@@ -229,9 +243,10 @@ func TestTopologyService_Update(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
p := &service{
- store: tt.fields.store,
- nodeService: tt.fields.nodeService,
- portService: tt.fields.portService,
+ store: tt.fields.store,
+ nodeService: tt.fields.nodeService,
+ portService: tt.fields.portService,
+ eventService: tt.fields.eventService,
}
err := p.UpdateLink(tt.args.link)
if (err != nil) != tt.wantErr {
@@ -253,9 +268,10 @@ func TestTopologyService_Update(t *testing.T) {
func TestNodeService_Delete(t *testing.T) {
type fields struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
type args struct {
link links.Link
@@ -270,9 +286,10 @@ func TestNodeService_Delete(t *testing.T) {
{
name: "should delete an existing link",
fields: fields{
- store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
- nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
- portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
+ nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
+ portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ eventService: eventservice.NewMockEventService(),
},
args: args{
link: getTestLink(),
@@ -295,9 +312,10 @@ func TestNodeService_Delete(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
p := &service{
- store: tt.fields.store,
- nodeService: tt.fields.nodeService,
- portService: tt.fields.portService,
+ store: tt.fields.store,
+ nodeService: tt.fields.nodeService,
+ portService: tt.fields.portService,
+ eventService: tt.fields.eventService,
}
if err := p.DeleteLink(tt.args.link); (err != nil) != tt.wantErr {
t.Errorf("service.Delete() error = %v, wantErr %v", err, tt.wantErr)
@@ -308,9 +326,10 @@ func TestNodeService_Delete(t *testing.T) {
func TestTopologyService_Get(t *testing.T) {
type fields struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
type args struct {
query store.Query
@@ -356,9 +375,10 @@ func TestTopologyService_Get(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
p := &service{
- store: tt.fields.store,
- nodeService: tt.fields.nodeService,
- portService: tt.fields.portService,
+ store: tt.fields.store,
+ nodeService: tt.fields.nodeService,
+ portService: tt.fields.portService,
+ eventService: tt.fields.eventService,
}
got, err := p.Get(tt.args.query)
if (err != nil) != tt.wantErr {
@@ -374,9 +394,10 @@ func TestTopologyService_Get(t *testing.T) {
func TestTopologyService_GetAll(t *testing.T) {
type fields struct {
- store Store
- nodeService nodes.Service
- portService ports.Service
+ store Store
+ nodeService nodes.Service
+ portService ports.Service
+ eventService eventInterfaces.Service
}
tests := []struct {
name string
@@ -387,9 +408,10 @@ func TestTopologyService_GetAll(t *testing.T) {
{
name: "should get all stored links",
fields: fields{
- store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
- nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
- portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ store: getTestStoreWithLinks(t, []links.Link{getTestLink()}),
+ nodeService: nodes.NewNodeService(getTestStoreWithNodes(t, []nodes.Node{}), eventservice.NewMockEventService()),
+ portService: ports.NewPortService(getTestStoreWithPorts(t, []ports.Port{}), eventservice.NewMockEventService()),
+ eventService: eventservice.NewMockEventService(),
},
want: []links.Link{getTestLink()},
wantErr: false,
@@ -398,9 +420,10 @@ func TestTopologyService_GetAll(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
p := &service{
- store: tt.fields.store,
- nodeService: tt.fields.nodeService,
- portService: tt.fields.portService,
+ store: tt.fields.store,
+ nodeService: tt.fields.nodeService,
+ portService: tt.fields.portService,
+ eventService: tt.fields.eventService,
}
got, err := p.GetAll()
if (err != nil) != tt.wantErr {