Initial rework pkg/registry to use begin pattern

.golangci.yml

@@ -35,7 +35,7 @@ linters-settings:
   dupl:
     threshold: 150
   funlen:
-    Lines: 100
+    Lines: 110
     Statements: 50
   goconst:
     min-len: 2
@@ -219,10 +219,6 @@ issues:
       linters:
         - interfacer
       text: "can be `fmt.Stringer`"
-    - path: pkg/networkservice/chains/nsmgr/peertracker/server.go
-      linters:
-        - interfacer
-      text: "can be `fmt.Stringer`"
     - path: pkg/networkservice/core/trace/client.go
       linters:
         - dupl

pkg/networkservice/chains/nsmgr/heal_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 Doc.ai and/or its affiliates.
+// Copyright (c) 2020-2022 Doc.ai and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -21,12 +21,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 
 	"github.com/networkservicemesh/api/pkg/api/registry"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/count"
+	"github.com/networkservicemesh/sdk/pkg/registry/chains/client"
 	"github.com/networkservicemesh/sdk/pkg/tools/sandbox"
 )
 
@@ -71,7 +73,7 @@ func testNSMGRHealEndpoint(t *testing.T, nodeNum int) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	nseReg := defaultRegistryEndpoint(nsReg.Name)
@@ -148,7 +150,7 @@ func testNSMGRHealForwarder(t *testing.T, nodeNum int) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	counter := new(count.Server)
@@ -230,7 +232,7 @@ func testNSMGRHealNSMgr(t *testing.T, nodeNum int, restored bool) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	nseReg := defaultRegistryEndpoint(nsReg.Name)
@@ -300,7 +302,7 @@ func TestNSMGR_HealRegistry(t *testing.T) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	nseReg := defaultRegistryEndpoint(nsReg.Name)
@@ -366,14 +368,14 @@ func testNSMGRCloseHeal(t *testing.T, withNSEExpiration bool) {
 		SetRegistryProxySupplier(nil)
 
 	if withNSEExpiration {
-		builder = builder.SetRegistryExpiryDuration(sandbox.RegistryExpiryDuration)
+		builder = builder.SetRegistryExpiryDuration(time.Second / 2)
 	}
 
 	domain := builder.Build()
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	nseCtx, nseCtxCancel := context.WithCancel(ctx)
@@ -407,15 +409,31 @@ func testNSMGRCloseHeal(t *testing.T, withNSEExpiration bool) {
 
 	if withNSEExpiration {
 		// 3.1 Wait for the endpoint expiration
-		time.Sleep(sandbox.RegistryExpiryDuration)
+		time.Sleep(time.Second)
+		c := client.NewNetworkServiceEndpointRegistryClient(ctx, domain.Nodes[0].NSMgr.URL, client.WithDialOptions(sandbox.DialOptions(sandbox.WithTokenGenerator(sandbox.GenerateTestToken))...))
+
+		stream, err := c.Find(ctx, &registry.NetworkServiceEndpointQuery{
+			NetworkServiceEndpoint: &registry.NetworkServiceEndpoint{
+				Name: "final-endpoint",
+			},
+		})
+
+		require.NoError(t, err)
+
+		require.Len(t, registry.ReadNetworkServiceEndpointList(stream), 0)
 	}
 
 	// 4. Close connection
 	_, _ = nsc.Close(nscCtx, conn.Clone())
 
 	nscCtxCancel()
 
+	for _, fwd := range domain.Nodes[0].Forwarders {
+		fwd.Cancel()
+	}
+
 	require.Eventually(t, func() bool {
+		logrus.Error(goleak.Find())
 		return goleak.Find(ignoreCurrent) == nil
 	}, timeout, tick)
 

pkg/networkservice/chains/nsmgr/server.go

@@ -43,18 +43,21 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/metrics"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/adapters"
 	"github.com/networkservicemesh/sdk/pkg/registry"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/begin"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/clientconn"
 	registryclientinfo "github.com/networkservicemesh/sdk/pkg/registry/common/clientinfo"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/clienturl"
 	registryconnect "github.com/networkservicemesh/sdk/pkg/registry/common/connect"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/dial"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/expire"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/localbypass"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/memory"
 	registryrecvfd "github.com/networkservicemesh/sdk/pkg/registry/common/recvfd"
 	registrysendfd "github.com/networkservicemesh/sdk/pkg/registry/common/sendfd"
+	"github.com/networkservicemesh/sdk/pkg/registry/switchcase"
 
-	registryserialize "github.com/networkservicemesh/sdk/pkg/registry/common/serialize"
 	registryadapter "github.com/networkservicemesh/sdk/pkg/registry/core/adapters"
-	registrychain "github.com/networkservicemesh/sdk/pkg/registry/core/chain"
+	"github.com/networkservicemesh/sdk/pkg/registry/core/chain"
 	"github.com/networkservicemesh/sdk/pkg/tools/grpcutils"
 	"github.com/networkservicemesh/sdk/pkg/tools/token"
 )
@@ -160,38 +163,62 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 	var nsRegistry = memory.NewNetworkServiceRegistryServer()
 	if opts.regURL != nil {
 		// Use remote registry
-		nsRegistry = registrychain.NewNetworkServiceRegistryServer(
-			clienturl.NewNetworkServiceRegistryServer(opts.regURL),
-			registryconnect.NewNetworkServiceRegistryServer(ctx, registryconnect.WithDialOptions(opts.regDialOptions...)),
+		nsRegistry = registryconnect.NewNetworkServiceRegistryServer(
+			chain.NewNetworkServiceRegistryClient(
+				clienturl.NewNetworkServiceRegistryClient(opts.regURL),
+				begin.NewNetworkServiceRegistryClient(),
+				clientconn.NewNetworkServiceRegistryClient(),
+				dial.NewNetworkServiceRegistryClient(ctx,
+					dial.WithDialOptions(opts.dialOptions...),
+				),
+				registryconnect.NewNetworkServiceRegistryClient(),
+			),
 		)
 	}
 
-	nsRegistry = registrychain.NewNetworkServiceRegistryServer(
-		registryserialize.NewNetworkServiceRegistryServer(),
+	nsRegistry = chain.NewNetworkServiceRegistryServer(
 		nsRegistry,
 	)
 
-	var nseInMemoryRegistry = memory.NewNetworkServiceEndpointRegistryServer()
-
-	var nseRegistry = registrychain.NewNetworkServiceEndpointRegistryServer(
+	var nseRegistry = chain.NewNetworkServiceEndpointRegistryServer(
 		registryclientinfo.NewNetworkServiceEndpointRegistryServer(),
-		registryserialize.NewNetworkServiceEndpointRegistryServer(),
+		begin.NewNetworkServiceEndpointRegistryServer(),
 		expire.NewNetworkServiceEndpointRegistryServer(ctx, time.Minute),
 		registryrecvfd.NewNetworkServiceEndpointRegistryServer(), // Allow to receive a passed files
-		registrysendfd.NewNetworkServiceEndpointRegistryServer(),
-		nseInMemoryRegistry,
+		switchcase.NewNetworkServiceEndpointRegistryServer(
+			switchcase.NSEServerCase{
+				Condition: func(c context.Context, nse *registryapi.NetworkServiceEndpoint) bool {
+					return opts.regURL != nil
+				},
+				Action: registrysendfd.NewNetworkServiceEndpointRegistryServer(),
+			},
+		),
 		localbypass.NewNetworkServiceEndpointRegistryServer(opts.url),
+		switchcase.NewNetworkServiceEndpointRegistryServer(
+			switchcase.NSEServerCase{
+				Condition: func(c context.Context, nse *registryapi.NetworkServiceEndpoint) bool {
+					return opts.regURL == nil
+				},
+				Action: memory.NewNetworkServiceEndpointRegistryServer(),
+			},
+			switchcase.NSEServerCase{
+				Condition: func(c context.Context, nse *registryapi.NetworkServiceEndpoint) bool {
+					return opts.regURL != nil
+				},
+				Action: registryconnect.NewNetworkServiceEndpointRegistryServer(
+					chain.NewNetworkServiceEndpointRegistryClient(
+						begin.NewNetworkServiceEndpointRegistryClient(),
+						clienturl.NewNetworkServiceEndpointRegistryClient(opts.regURL),
+						clientconn.NewNetworkServiceEndpointRegistryClient(),
+						dial.NewNetworkServiceEndpointRegistryClient(ctx,
+							dial.WithDialOptions(opts.dialOptions...),
+						),
+						registryconnect.NewNetworkServiceEndpointRegistryClient(),
+					),
+				),
+			}),
 	)
 
-	if opts.regURL != nil {
-		// Add remote registry
-		nseRegistry = registrychain.NewNetworkServiceEndpointRegistryServer(
-			nseRegistry,
-			clienturl.NewNetworkServiceEndpointRegistryServer(opts.regURL),
-			registryconnect.NewNetworkServiceEndpointRegistryServer(ctx, registryconnect.WithDialOptions(opts.regDialOptions...)),
-		)
-	}
-
 	// Construct Endpoint
 	rv.Endpoint = endpoint.NewServer(ctx, tokenGenerator,
 		endpoint.WithName(opts.name),
@@ -200,8 +227,9 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 			adapters.NewClientToServer(clientinfo.NewClient()),
 			discoverforwarder.NewServer(
 				registryadapter.NetworkServiceServerToClient(nsRegistry),
-				registryadapter.NetworkServiceEndpointServerToClient(nseInMemoryRegistry),
+				registryadapter.NetworkServiceEndpointServerToClient(nseRegistry),
 				discoverforwarder.WithForwarderServiceName(opts.forwarderServiceName),
+				discoverforwarder.WithNSMgrURL(opts.url),
 			),
 			excludedprefixes.NewServer(ctx),
 			recvfd.NewServer(), // Receive any files passed

pkg/networkservice/chains/nsmgr/single_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 Doc.ai and/or its affiliates.
+// Copyright (c) 2020-2022 Doc.ai and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -50,7 +50,7 @@ func Test_DNSUsecase(t *testing.T) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService(t.Name()))
 	require.NoError(t, err)
 
 	nseReg := defaultRegistryEndpoint(nsReg.Name)
@@ -126,7 +126,7 @@ func Test_ShouldParseNetworkServiceLabelsTemplate(t *testing.T) {
 
 	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
 
-	nsReg := defaultRegistryService()
+	nsReg := defaultRegistryService(t.Name())
 	nsReg.Matches = []*registry.Match{
 		{
 			Routes: []*registry.Destination{
@@ -176,7 +176,7 @@ func Test_UsecasePoint2MultiPoint(t *testing.T) {
 		SetNodeSetup(func(ctx context.Context, node *sandbox.Node, _ int) {
 			node.NewNSMgr(ctx, "nsmgr", nil, sandbox.GenerateTestToken, nsmgr.NewServer)
 		}).
-		SetRegistryExpiryDuration(sandbox.RegistryExpiryDuration).
+		SetRegistryExpiryDuration(time.Second).
 		Build()
 
 	domain.Nodes[0].NewForwarder(ctx, &registry.NetworkServiceEndpoint{
@@ -297,7 +297,7 @@ func Test_RemoteUsecase_Point2MultiPoint(t *testing.T) {
 		SetNodeSetup(func(ctx context.Context, node *sandbox.Node, _ int) {
 			node.NewNSMgr(ctx, "nsmgr", nil, sandbox.GenerateTestToken, nsmgr.NewServer)
 		}).
-		SetRegistryExpiryDuration(sandbox.RegistryExpiryDuration).
+		SetRegistryExpiryDuration(time.Second).
 		Build()
 
 	for i := 0; i < nodeCount; i++ {