Rework pkg/registry to use clock.Clock (#784)

* Rework registry/querycache to clock

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

* Rework registry/expire to clock

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

* Rework registry/connect to clock

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

* Rework registry/refresh to clock

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

* Remove time.Sleep from clocked registry tests

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

* Remove clockMock.IsTimerSet

Signed-off-by: Vladimir Popov <vladimir.popov@xored.com>

pkg/registry/common/connect/ns_server.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020 Doc.ai and/or its affiliates.
+// Copyright (c) 2020-2021 Doc.ai and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/networkservicemesh/sdk/pkg/tools/clienturlctx"
+	"github.com/networkservicemesh/sdk/pkg/tools/clock"
 
 	"github.com/networkservicemesh/sdk/pkg/registry/common/clienturl"
 	"github.com/networkservicemesh/sdk/pkg/tools/extend"
@@ -33,7 +34,7 @@ import (
 )
 
 type nsCacheEntry struct {
-	expirationTimer *time.Timer
+	expirationTimer clock.Timer
 	client          registry.NetworkServiceRegistryClient
 }
 
@@ -43,6 +44,7 @@ type connectNSServer struct {
 	cache             nsClientMap
 	connectExpiration time.Duration
 	ctx               context.Context
+	clock             clock.Clock
 }
 
 // NewNetworkServiceRegistryServer creates new connect NetworkServiceEndpointRegistryServer with specific chain context, registry client factory and options
@@ -52,6 +54,7 @@ func NewNetworkServiceRegistryServer(ctx context.Context, clientFactory func(ctx
 		ctx:               ctx,
 		clientFactory:     clientFactory,
 		connectExpiration: defaultConnectExpiration,
+		clock:             clock.FromContext(ctx),
 	}
 	for _, o := range options {
 		o.apply(r)
@@ -87,7 +90,7 @@ func (c *connectNSServer) connect(ctx context.Context) registry.NetworkServiceRe
 	ctx = extend.WithValuesFromContext(c.ctx, ctx)
 	client := clienturl.NewNetworkServiceRegistryClient(ctx, c.clientFactory, c.dialOptions...)
 	cached, _ := c.cache.LoadOrStore(key, &nsCacheEntry{
-		expirationTimer: time.AfterFunc(c.connectExpiration, func() {
+		expirationTimer: c.clock.AfterFunc(c.connectExpiration, func() {
 			c.cache.Delete(key)
 		}),
 		client: client,

pkg/registry/common/connect/nse_server.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020 Doc.ai and/or its affiliates.
+// Copyright (c) 2020-2021 Doc.ai and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/networkservicemesh/sdk/pkg/tools/clienturlctx"
+	"github.com/networkservicemesh/sdk/pkg/tools/clock"
 
 	"github.com/networkservicemesh/sdk/pkg/registry/common/clienturl"
 	"github.com/networkservicemesh/sdk/pkg/tools/extend"
@@ -33,7 +34,7 @@ import (
 )
 
 type nseCacheEntry struct {
-	expirationTimer *time.Timer
+	expirationTimer clock.Timer
 	client          registry.NetworkServiceEndpointRegistryClient
 }
 
@@ -43,6 +44,7 @@ type connectNSEServer struct {
 	cache             nseClientMap
 	connectExpiration time.Duration
 	ctx               context.Context
+	clock             clock.Clock
 }
 
 // NewNetworkServiceEndpointRegistryServer creates new connect NetworkServiceEndpointEndpointRegistryServer with specific chain context, registry client factory and options
@@ -55,6 +57,7 @@ func NewNetworkServiceEndpointRegistryServer(ctx context.Context,
 		ctx:               ctx,
 		clientFactory:     clientFactory,
 		connectExpiration: defaultConnectExpiration,
+		clock:             clock.FromContext(ctx),
 	}
 	for _, o := range options {
 		o.apply(r)
@@ -88,7 +91,7 @@ func (c *connectNSEServer) connect(ctx context.Context) registry.NetworkServiceE
 	ctx = extend.WithValuesFromContext(c.ctx, ctx)
 	client := clienturl.NewNetworkServiceEndpointRegistryClient(ctx, c.clientFactory, c.dialOptions...)
 	cached, _ := c.cache.LoadOrStore(key, &nseCacheEntry{
-		expirationTimer: time.AfterFunc(c.connectExpiration, func() {
+		expirationTimer: c.clock.AfterFunc(c.connectExpiration, func() {
 			c.cache.Delete(key)
 		}),
 		client: client,

pkg/registry/common/expire/ns_server.go

@@ -20,9 +20,9 @@ import (
 	"context"
 	"errors"
 	"sync"
-	"time"
 
 	"github.com/networkservicemesh/sdk/pkg/registry/core/next"
+	"github.com/networkservicemesh/sdk/pkg/tools/clock"
 	"github.com/networkservicemesh/sdk/pkg/tools/extend"
 
 	"github.com/golang/protobuf/ptypes/empty"
@@ -39,12 +39,14 @@ type expireNSServer struct {
 }
 
 type nsState struct {
-	Timers  map[string]*time.Timer
+	Timers  map[string]clock.Timer
 	Context context.Context
 	sync.Mutex
 }
 
 func (n *expireNSServer) checkUpdates(eventCh <-chan *registry.NetworkServiceEndpoint) {
+	clockTime := clock.FromContext(n.chainCtx)
+
 	for event := range eventCh {
 		nse := event
 		if nse.ExpirationTime == nil {
@@ -59,10 +61,10 @@ func (n *expireNSServer) checkUpdates(eventCh <-chan *registry.NetworkServiceEnd
 			}
 			state.Lock()
 			timer, ok := state.Timers[nse.Name]
-			expirationDuration := time.Until(nse.ExpirationTime.AsTime().Local())
+			expirationDuration := clockTime.Until(nse.ExpirationTime.AsTime().Local())
 			if !ok {
 				if expirationDuration > 0 {
-					state.Timers[nse.Name] = time.AfterFunc(expirationDuration, func() {
+					state.Timers[nse.Name] = clockTime.AfterFunc(expirationDuration, func() {
 						state.Lock()
 						ctx := state.Context
 						delete(state.Timers, nse.Name)
@@ -119,7 +121,7 @@ func (n *expireNSServer) Register(ctx context.Context, request *registry.Network
 	valuesCtx := extend.WithValuesFromContext(n.chainCtx, ctx)
 
 	v, _ := n.nsStates.LoadOrStore(request.Name, &nsState{
-		Timers: make(map[string]*time.Timer),
+		Timers: make(map[string]clock.Timer),
 	})
 
 	v.Lock()