add deadlettertopic support to non-streaming subscriptions (#685)

* add deadlettertopic support to non-streaming subscriptions

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

* fix tests

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

---------

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

Author: yaron2

service/common/type.go

@@ -107,6 +107,8 @@ type Subscription struct {
 	Priority int `json:"priority"`
 	// DisableTopicValidation allows to receive events from publisher topics that differ from the subscribed topic.
 	DisableTopicValidation bool `json:"disableTopicValidation"`
+	// DeadLetterTopic is the name of the deadletter topic.
+	DeadLetterTopic string `json:"deadLetterTopic"`
 }
 
 type SubscriptionResponseStatus string

service/grpc/topic.go

@@ -53,10 +53,11 @@ func (s *Server) ListTopicSubscriptions(ctx context.Context, in *emptypb.Empty)
 	for _, v := range s.topicRegistrar {
 		s := v.Subscription
 		sub := &runtimev1pb.TopicSubscription{
-			PubsubName: s.PubsubName,
-			Topic:      s.Topic,
-			Metadata:   s.Metadata,
-			Routes:     convertRoutes(s.Routes),
+			PubsubName:      s.PubsubName,
+			Topic:           s.Topic,
+			Metadata:        s.Metadata,
+			Routes:          convertRoutes(s.Routes),
+			DeadLetterTopic: s.DeadLetterTopic,
 		}
 		subs = append(subs, sub)
 	}

service/internal/topicregistrar.go

@@ -39,7 +39,7 @@ func (m TopicRegistrar) AddSubscription(sub *common.Subscription, fn common.Topi
 	ts, ok := m[key]
 	if !ok {
 		ts = &TopicRegistration{
-			Subscription:   NewTopicSubscription(sub.PubsubName, sub.Topic),
+			Subscription:   NewTopicSubscription(sub.PubsubName, sub.Topic, sub.DeadLetterTopic),
 			RouteHandlers:  make(map[string]common.TopicEventSubscriber),
 			DefaultHandler: nil,
 		}

service/internal/topicsubscription.go

@@ -18,6 +18,8 @@ type TopicSubscription struct {
 	Routes *TopicRoutes `json:"routes,omitempty"`
 	// Metadata is the subscription metadata.
 	Metadata map[string]string `json:"metadata,omitempty"`
+	// DeadLetterTopic is the name of the deadletter topic.
+	DeadLetterTopic string `json:"deadLetterTopic"`
 }
 
 // TopicRoutes encapsulates the default route and multiple routing rules.
@@ -42,10 +44,11 @@ type TopicRule struct {
 }
 
 // NewTopicSubscription creates a new `TopicSubscription`.
-func NewTopicSubscription(pubsubName, topic string) *TopicSubscription {
+func NewTopicSubscription(pubsubName, topic, deadLetterTopic string) *TopicSubscription {
 	return &TopicSubscription{ //nolint:exhaustivestruct
-		PubsubName: pubsubName,
-		Topic:      topic,
+		PubsubName:      pubsubName,
+		Topic:           topic,
+		DeadLetterTopic: deadLetterTopic,
 	}
 }
 

service/internal/topicsubscription_test.go

@@ -12,7 +12,7 @@ import (
 
 func TestTopicSubscripiton(t *testing.T) {
 	t.Run("duplicate metadata", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.SetMetadata(map[string]string{
 			"test": "test",
 		}))
@@ -22,23 +22,23 @@ func TestTopicSubscripiton(t *testing.T) {
 	})
 
 	t.Run("duplicate route", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.SetDefaultRoute("/test"))
 		assert.Equal(t, "/test", sub.Route)
 		require.EqualError(t, sub.SetDefaultRoute("/test"),
 			"subscription for topic mytopic on pubsub test already has route /test")
 	})
 
 	t.Run("duplicate route after routing rule", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.AddRoutingRule("/other", `event.type == "test"`, 0))
 		require.NoError(t, sub.SetDefaultRoute("/test"))
 		require.EqualError(t, sub.SetDefaultRoute("/test"),
 			"subscription for topic mytopic on pubsub test already has route /test")
 	})
 
 	t.Run("default route after routing rule", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.SetDefaultRoute("/test"))
 		assert.Equal(t, "/test", sub.Route)
 		require.NoError(t, sub.AddRoutingRule("/other", `event.type == "test"`, 0))
@@ -49,14 +49,14 @@ func TestTopicSubscripiton(t *testing.T) {
 	})
 
 	t.Run("duplicate routing rule priority", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.AddRoutingRule("/other", `event.type == "other"`, 1))
 		require.EqualError(t, sub.AddRoutingRule("/test", `event.type == "test"`, 1),
 			"subscription for topic mytopic on pubsub test already has a routing rule with priority 1")
 	})
 
 	t.Run("priority ordering", func(t *testing.T) {
-		sub := internal.NewTopicSubscription("test", "mytopic")
+		sub := internal.NewTopicSubscription("test", "mytopic", "")
 		require.NoError(t, sub.AddRoutingRule("/100", `event.type == "100"`, 100))
 		require.NoError(t, sub.AddRoutingRule("/1", `event.type == "1"`, 1))
 		require.NoError(t, sub.AddRoutingRule("/50", `event.type == "50"`, 50))