lint and fix tests

Signed-off-by: Anton Troshin <[email protected]>

Author: antontroshin

pkg/runtime/processor/manager.go

@@ -16,14 +16,14 @@ package processor
 import (
 	"context"
 	"fmt"
-	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
-	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 
 	"github.com/dapr/durabletask-go/backend"
 
 	"github.com/dapr/components-contrib/bindings"
 	componentsapi "github.com/dapr/dapr/pkg/apis/components/v1alpha1"
+	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	"github.com/dapr/dapr/pkg/runtime/meta"
+	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 )
 
 // manager implements the life cycle events of a component category.

pkg/runtime/processor/subscriber/subscriber.go

@@ -17,14 +17,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	"sync"
 	"sync/atomic"
 
 	"google.golang.org/grpc"
 
 	apierrors "github.com/dapr/dapr/pkg/api/errors"
 	"github.com/dapr/dapr/pkg/api/grpc/manager"
+	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	"github.com/dapr/dapr/pkg/config"
 	runtimev1pb "github.com/dapr/dapr/pkg/proto/runtime/v1"
 	"github.com/dapr/dapr/pkg/resiliency"
@@ -139,7 +139,10 @@ func (s *Subscriber) StartStreamerSubscription(subscription *subapi.Subscription
 		return apierrors.PubSub("").WithMetadata(nil).DeserializeError(errors.New("subscriber is closed"))
 	}
 
-	sub, ok := s.compStore.GetStreamSubscription(subscription)
+	sub, found := s.compStore.GetStreamSubscription(subscription)
+	if !found {
+		return apierrors.PubSub("").WithMetadata(nil).NotFound()
+	}
 
 	pubsub, ok := s.compStore.GetPubSub(sub.PubsubName)
 	if !ok {
@@ -153,8 +156,8 @@ func (s *Subscriber) StartStreamerSubscription(subscription *subapi.Subscription
 
 	key := s.adapterStreamer.StreamerKey(sub.PubsubName, sub.Topic)
 
-	subscriptions, ok := s.streamSubs[sub.PubsubName]
-	if subscriptions == nil {
+	_, exists := s.streamSubs[sub.PubsubName]
+	if !exists {
 		s.streamSubs[sub.PubsubName] = make(map[rtpubsub.ConnectionID]*namedSubscription)
 	}
 
@@ -352,7 +355,6 @@ func (s *Subscriber) reloadPubSubStream(name string, pubsub *rtpubsub.PubsubItem
 		return nil
 	}
 
-	//subs := make([]*namedSubscription, 0, len(s.compStore.ListSubscriptionsStreamByPubSub(name)))
 	subs := make(map[rtpubsub.ConnectionID]*namedSubscription, len(s.compStore.ListSubscriptionsStreamByPubSub(name)))
 	var errs []error
 	for _, sub := range s.compStore.ListSubscriptionsStreamByPubSub(name) {

pkg/runtime/processor/subscriber/subscriber_test.go

@@ -96,23 +96,23 @@ func TestSubscriptionLifecycle(t *testing.T) {
 			Topic:      "topic4",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(1))
 	compStore.AddStreamSubscription(&subapi.Subscription{
 		ObjectMeta: metav1.ObjectMeta{Name: "sub2||"},
 		Spec: subapi.SubscriptionSpec{
 			Pubsubname: "mockPubSub2",
 			Topic:      "topic5",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(2))
 	compStore.AddStreamSubscription(&subapi.Subscription{
 		ObjectMeta: metav1.ObjectMeta{Name: "sub3||"},
 		Spec: subapi.SubscriptionSpec{
 			Pubsubname: "mockPubSub3",
 			Topic:      "topic6",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(3))
 
 	subs := New(Options{
 		CompStore:  compStore,
@@ -356,23 +356,23 @@ func TestReloadPubSub(t *testing.T) {
 			Topic:      "topic4",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(1))
 	compStore.AddStreamSubscription(&subapi.Subscription{
 		ObjectMeta: metav1.ObjectMeta{Name: "sub2||"},
 		Spec: subapi.SubscriptionSpec{
 			Pubsubname: "mockPubSub2",
 			Topic:      "topic5",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(2))
 	compStore.AddStreamSubscription(&subapi.Subscription{
 		ObjectMeta: metav1.ObjectMeta{Name: "sub3||"},
 		Spec: subapi.SubscriptionSpec{
 			Pubsubname: "mockPubSub3",
 			Topic:      "topic6",
 			Routes:     subapi.Routes{Default: "/"},
 		},
-	})
+	}, rtpubsub.ConnectionID(3))
 
 	subs.ReloadPubSub("mockPubSub1")
 	assert.Eventually(t, func() bool {

pkg/runtime/pubsub/streamer/conn.go

@@ -14,19 +14,17 @@ limitations under the License.
 package streamer
 
 import (
-	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 	"sync"
 	"sync/atomic"
 
 	rtv1pb "github.com/dapr/dapr/pkg/proto/runtime/v1"
+	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 )
 
 type conn struct {
-	lock       sync.RWMutex
-	streamLock sync.Mutex
-	stream     rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
-	//publishResponses map[string]chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1
-	//publishResponses2 sync.Map
+	lock              sync.RWMutex
+	streamLock        sync.Mutex
+	stream            rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 	publishResponses3 map[string]map[rtpubsub.ConnectionID]chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1
 	closeCh           chan struct{}
 	closed            atomic.Bool
@@ -38,10 +36,6 @@ func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEv
 	log.Warnf("Lock registerPublishResponse messageId %s ConnectionID%d", id, c.connectionID)
 	c.lock.Lock()
 
-	//ch, ok := c.publishResponses[id]
-	//if !ok {
-	//	c.publishResponses[id] = ch
-	//}
 	if c.publishResponses3[id] == nil {
 		c.publishResponses3[id] = make(map[rtpubsub.ConnectionID]chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1)
 	}
@@ -52,7 +46,6 @@ func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEv
 	return ch, func() {
 		log.Warnf("Lock registerPublishResponse defer messageId %s ConnectionID%d", id, c.connectionID)
 		c.lock.Lock()
-		//delete(c.publishResponses, id)
 
 		delete(c.publishResponses3[id], c.connectionID)
 		if len(c.publishResponses3[id]) == 0 {
@@ -73,7 +66,6 @@ func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEv
 func (c *conn) notifyPublishResponse(resp *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1) {
 	log.Warnf("Lock notifyPublishResponse messageId %s ConnectionID%d", resp.GetId(), c.connectionID)
 	c.lock.RLock()
-	//ch, ok := c.publishResponses[resp.GetId()]
 	ch, ok := c.publishResponses3[resp.GetId()][c.connectionID]
 	c.lock.RUnlock()
 	log.Warnf("Unlock notifyPublishResponse messageId %s ConnectionID%d", resp.GetId(), c.connectionID)

pkg/runtime/pubsub/streamer/streamer.go

@@ -66,8 +66,7 @@ func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 	key := s.StreamerKey(req.GetPubsubName(), req.GetTopic())
 
 	connection := &conn{
-		stream: stream,
-		//publishResponses:  make(map[string]chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1),
+		stream:            stream,
 		publishResponses3: make(map[string]map[rtpubsub.ConnectionID]chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1),
 		closeCh:           make(chan struct{}),
 		connectionID:      connectionID,
@@ -157,7 +156,7 @@ func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage)
 	}
 
 	if connection.closed.Load() {
-		return fmt.Errorf("connection is closed")
+		return errors.New("connection is closed")
 	}
 
 	envelope, span, err := rtpubsub.GRPCEnvelopeFromSubscriptionMessage(ctx, msg, log, s.tracingSpec)

tests/integration/suite/daprd/subscriptions/stream/multi.go

@@ -15,13 +15,14 @@ package stream
 
 import (
 	"context"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
 	"github.com/dapr/dapr/tests/integration/framework"
 	"github.com/dapr/dapr/tests/integration/framework/process/daprd"
@@ -145,8 +146,8 @@ func (m *multi) Run(t *testing.T, ctx context.Context) {
 		stream2: "b",
 		stream3: "c",
 	} {
-		event, err := stream.Recv()
-		require.NoError(t, err)
+		event, recvErr := stream.Recv()
+		require.NoError(t, recvErr)
 		assert.Equal(t, topic, event.GetEventMessage().GetTopic())
 	}
 
@@ -226,7 +227,7 @@ func (m *multi) Run(t *testing.T, ctx context.Context) {
 
 			for receivedMessages < expectedMessages {
 				event, recvErr := stream.Recv()
-				require.NoError(c, recvErr)
+				assert.NoError(c, recvErr)
 				assert.Equal(c, singleTopicMultipleSubscribers, event.GetEventMessage().GetTopic())
 
 				sendErr := stream.Send(&rtv1.SubscribeTopicEventsRequestAlpha1{
@@ -237,7 +238,7 @@ func (m *multi) Run(t *testing.T, ctx context.Context) {
 						},
 					},
 				})
-				require.NoError(c, sendErr)
+				assert.NoError(c, sendErr)
 
 				receivedTotal.Add(1)
 				receivedMessages++
@@ -256,6 +257,6 @@ func (m *multi) Run(t *testing.T, ctx context.Context) {
 
 	assert.Eventually(t, func() bool {
 		wg.Wait()
-		return receivedTotal.Load() == int32(messagesToSend*len(subscribers))
+		return receivedTotal.Load() == int32(messagesToSend*len(subscribers)) //nolint:gosec
 	}, time.Second*10, time.Millisecond*10)
 }