Don't restore eventFactory in case the connection has already been cl…

…osed/unregistered

Signed-off-by: Artem Glazychev <artem.glazychev@xored.com>

pkg/networkservice/common/begin/client.go

@@ -59,7 +59,7 @@ func (b *beginClient) Request(ctx context.Context, request *networkservice.Netwo
 	<-eventFactoryClient.executor.AsyncExec(func() {
 		// If the eventFactory has changed, usually because the connection has been Closed and re-established
 		// go back to the beginning and try again.
-		currentEventFactoryClient, _ := b.LoadOrStore(request.GetConnection().GetId(), eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(request.GetConnection().GetId())
 		if currentEventFactoryClient != eventFactoryClient {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryClient != eventFactoryClient")
 			conn, err = b.Request(ctx, request, opts...)
@@ -103,7 +103,7 @@ func (b *beginClient) Close(ctx context.Context, conn *networkservice.Connection
 		}
 
 		// If this isn't the connection we started with, do nothing
-		currentEventFactoryClient, _ := b.LoadOrStore(conn.GetId(), eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(conn.GetId())
 		if currentEventFactoryClient != eventFactoryClient {
 			return
 		}

pkg/networkservice/common/begin/server.go

@@ -55,7 +55,7 @@ func (b *beginServer) Request(ctx context.Context, request *networkservice.Netwo
 		),
 	)
 	<-eventFactoryServer.executor.AsyncExec(func() {
-		currentEventFactoryServer, _ := b.LoadOrStore(request.GetConnection().GetId(), eventFactoryServer)
+		currentEventFactoryServer, _ := b.Load(request.GetConnection().GetId())
 		if currentEventFactoryServer != eventFactoryServer {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryServer != eventFactoryServer")
 			conn, err = b.Request(ctx, request)
@@ -93,7 +93,7 @@ func (b *beginServer) Close(ctx context.Context, conn *networkservice.Connection
 		if eventFactoryServer.state != established || eventFactoryServer.request == nil {
 			return
 		}
-		currentServerClient, _ := b.LoadOrStore(conn.GetId(), eventFactoryServer)
+		currentServerClient, _ := b.Load(conn.GetId())
 		if currentServerClient != eventFactoryServer {
 			return
 		}

pkg/registry/common/begin/ns_client.go

@@ -55,7 +55,7 @@ func (b *beginNSClient) Register(ctx context.Context, in *registry.NetworkServic
 	<-eventFactoryClient.executor.AsyncExec(func() {
 		// If the eventFactory has changed, usually because the connection has been Closed and re-established
 		// go back to the beginning and try again.
-		currentEventFactoryClient, _ := b.LoadOrStore(id, eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(id)
 		if currentEventFactoryClient != eventFactoryClient {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryClient != eventFactoryClient")
 			resp, err = b.Register(ctx, in, opts...)
@@ -101,7 +101,7 @@ func (b *beginNSClient) Unregister(ctx context.Context, in *registry.NetworkServ
 		}
 
 		// If this isn't the connection we started with, do nothing
-		currentEventFactoryClient, _ := b.LoadOrStore(id, eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(id)
 		if currentEventFactoryClient != eventFactoryClient {
 			return
 		}

pkg/registry/common/begin/ns_server.go

@@ -54,7 +54,7 @@ func (b *beginNSServer) Register(ctx context.Context, in *registry.NetworkServic
 	var err error
 
 	<-eventFactoryServer.executor.AsyncExec(func() {
-		currentEventFactoryServer, _ := b.LoadOrStore(id, eventFactoryServer)
+		currentEventFactoryServer, _ := b.Load(id)
 		if currentEventFactoryServer != eventFactoryServer {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryServer != eventFactoryServer")
 			resp, err = b.Register(ctx, in)
@@ -96,7 +96,7 @@ func (b *beginNSServer) Unregister(ctx context.Context, in *registry.NetworkServ
 		if eventFactoryServer.state != established || eventFactoryServer.registration == nil {
 			return
 		}
-		currentServerClient, _ := b.LoadOrStore(id, eventFactoryServer)
+		currentServerClient, _ := b.Load(id)
 		if currentServerClient != eventFactoryServer {
 			return
 		}

pkg/registry/common/begin/nse_client.go

@@ -55,7 +55,7 @@ func (b *beginNSEClient) Register(ctx context.Context, in *registry.NetworkServi
 	<-eventFactoryClient.executor.AsyncExec(func() {
 		// If the eventFactory has changed, usually because the connection has been Closed and re-established
 		// go back to the beginning and try again.
-		currentEventFactoryClient, _ := b.LoadOrStore(id, eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(id)
 		if currentEventFactoryClient != eventFactoryClient {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryClient != eventFactoryClient")
 			resp, err = b.Register(ctx, in, opts...)
@@ -101,7 +101,7 @@ func (b *beginNSEClient) Unregister(ctx context.Context, in *registry.NetworkSer
 		}
 
 		// If this isn't the connection we started with, do nothing
-		currentEventFactoryClient, _ := b.LoadOrStore(id, eventFactoryClient)
+		currentEventFactoryClient, _ := b.Load(id)
 		if currentEventFactoryClient != eventFactoryClient {
 			return
 		}

pkg/registry/common/begin/nse_server.go

@@ -54,7 +54,7 @@ func (b *beginNSEServer) Register(ctx context.Context, in *registry.NetworkServi
 	var err error
 
 	<-eventFactoryServer.executor.AsyncExec(func() {
-		currentEventFactoryServer, _ := b.LoadOrStore(id, eventFactoryServer)
+		currentEventFactoryServer, _ := b.Load(id)
 		if currentEventFactoryServer != eventFactoryServer {
 			log.FromContext(ctx).Debug("recalling begin.Request because currentEventFactoryServer != eventFactoryServer")
 			resp, err = b.Register(ctx, in)
@@ -96,7 +96,7 @@ func (b *beginNSEServer) Unregister(ctx context.Context, in *registry.NetworkSer
 		if eventFactoryServer.state != established || eventFactoryServer.registration == nil {
 			return
 		}
-		currentServerClient, _ := b.LoadOrStore(id, eventFactoryServer)
+		currentServerClient, _ := b.Load(id)
 		if currentServerClient != eventFactoryServer {
 			return
 		}