add unit tests + fix some errors

Signed-off-by: NikitaSkrynnik <[email protected]>

pkg/networkservice/common/monitor/client_filter.go

@@ -38,7 +38,7 @@ func newClientFilter(client networkservice.MonitorConnection_MonitorConnectionsC
 func (c *clientFilter) Recv() (*networkservice.ConnectionEvent, error) {
 	for {
 		if c == nil || c.MonitorConnection_MonitorConnectionsClient == nil {
-			return nil, nil
+			return nil, errors.New("MonitorConnections cilent is nil")
 		}
 		eventIn, err := c.MonitorConnection_MonitorConnectionsClient.Recv()
 		if err != nil {

pkg/networkservice/common/monitor/eventloop.go

@@ -18,6 +18,7 @@ package monitor
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	"google.golang.org/grpc"
@@ -42,8 +43,6 @@ func newEventLoop(ctx context.Context, ec EventConsumer, cc grpc.ClientConnInter
 
 	// Create new eventLoopCtx and store its eventLoopCancel
 	eventLoopCtx, eventLoopCancel := context.WithCancel(ctx)
-
-	// Create selector to only ask for events related to our Connection
 	cev := &eventLoop{
 		eventLoopCtx:  eventLoopCtx,
 		conn:          conn,
@@ -58,6 +57,7 @@ func newEventLoop(ctx context.Context, ec EventConsumer, cc grpc.ClientConnInter
 }
 
 func (cev *eventLoop) eventLoop() {
+	// Create selector to only ask for events related to our Connection
 	selector := &networkservice.MonitorScopeSelector{
 		PathSegments: []*networkservice.PathSegment{
 			{
@@ -67,7 +67,9 @@ func (cev *eventLoop) eventLoop() {
 		},
 	}
 
+	fmt.Println("Trying to connect to monitor")
 	client, err := networkservice.NewMonitorConnectionClient(cev.cc).MonitorConnections(cev.eventLoopCtx, selector)
+	fmt.Println("connected (maybe)")
 	if err != nil {
 		log.FromContext(cev.eventLoopCtx).Infof("failed to get a MonitorConnections client: %s", err.Error())
 		cev.cancel()

pkg/networkservice/common/monitor/server_test.go

@@ -34,8 +34,10 @@ import (
 	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/cls"
 	kernelmech "github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/kernel"
 
+	"github.com/networkservicemesh/sdk/pkg/networkservice/common/clientconn"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/monitor"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/chain"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/checks/checkcontext"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
@@ -209,6 +211,38 @@ func TestMonitorServer_RequestConnEqualsToMonitorConn(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestMonitorServer_Connection(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	cc, err := grpc.Dial("1.1.1.1:5000", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	require.NotNil(t, cc)
+
+	// Create monitorServer, monitorClient, and server.
+	var monitorServer networkservice.MonitorConnectionServer
+	server := chain.NewNetworkServiceServer(
+		metadata.NewServer(),
+		checkcontext.NewServer(t, func(t *testing.T, ctx context.Context) {
+			clientconn.Store(ctx, cc)
+		}),
+		monitor.NewServer(ctx, &monitorServer),
+	)
+
+	request := &networkservice.NetworkServiceRequest{
+		Connection: &networkservice.Connection{Id: "id"},
+	}
+
+	conn, err := server.Request(ctx, request)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	monitorClient := adapters.NewMonitorServerToClient(monitorServer)
+	client, err := monitorClient.MonitorConnections(ctx, &networkservice.MonitorScopeSelector{})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+}
+
 type metricsServer struct{}
 
 func (m *metricsServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {