feat: bulksubscribe http

examples/pubsub/sub/sub.go

@@ -46,13 +46,14 @@ var importantSubscription = &common.Subscription{
 func main() {
 	s := daprd.NewService(":8080")
 
+	// for single event subscribing
 	if err := s.AddTopicEventHandler(defaultSubscription, eventHandler); err != nil {
 		log.Fatalf("error adding topic subscription: %v", err)
 	}
 
-	if err := s.AddTopicEventHandler(importantSubscription, importantEventHandler); err != nil {
-		log.Fatalf("error adding topic subscription: %v", err)
-	}
+	// if err := s.AddBulkTopicEventHandler(defaultSubscription, eventHandler,10,100); err != nil {
+	// 	log.Fatalf("error adding topic subscription: %v", err)
+	// }
 
 	if err := s.Start(); err != nil && err != http.ErrServerClosed {
 		log.Fatalf("error listenning: %v", err)
@@ -64,6 +65,13 @@ func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err er
 	return false, nil
 }
 
+func bulkeventHandler(ctx context.Context, e []common.TopicEvent) (retry bool, err error) {
+	for _, event := range e {
+		log.Printf("event - PubsubName: %s, Topic: %s, ID: %s, Data: %s", event.PubsubName, event.Topic, event.ID, event.Data)
+	}
+	return false, nil
+}
+
 func importantEventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
 	log.Printf("important event - PubsubName: %s, Topic: %s, ID: %s, Data: %s", e.PubsubName, e.Topic, e.ID, e.Data)
 	return false, nil

service/common/service.go

@@ -35,6 +35,8 @@ type Service interface {
 	// AddTopicEventHandler appends provided event handler with its topic and optional metadata to the service.
 	// Note, retries are only considered when there is an error. Lack of error is considered as a success
 	AddTopicEventHandler(sub *Subscription, fn TopicEventHandler) error
+	// AddBulkTopicEventHandler appends provided event handler with its topic along with configuring maxMessagesCount, maxAwaitDurationMs for bulk handling  and optional metadata to the service.
+	AddBulkTopicEventHandler(sub *Subscription, fn TopicEventHandler, maxMessagesCount, maxAwaitDurationMs int32) error
 	// AddBindingInvocationHandler appends provided binding invocation handler with its name to the service.
 	AddBindingInvocationHandler(name string, fn BindingInvocationHandler) error
 	// RegisterActorImplFactory Register a new actor to actor runtime of go sdk

service/grpc/topic.go

@@ -37,16 +37,25 @@
 	return s.topicRegistrar.AddSubscription(sub, fn)
 }
 
+func (s *Server) AddBulkTopicEventHandler(sub *common.Subscription, fn common.TopicEventHandler, maxMessagesCount, maxAwaitDurationMs int32) error {
+	if sub == nil {
+		return errors.New("subscription required")
+	}
+
+	return s.topicRegistrar.AddBulkSubscription(sub, fn, maxMessagesCount, maxAwaitDurationMs)
+}
+
 // ListTopicSubscriptions is called by Dapr to get the list of topics in a pubsub component the app wants to subscribe to.
 func (s *Server) ListTopicSubscriptions(ctx context.Context, in *empty.Empty) (*runtimev1pb.ListTopicSubscriptionsResponse, error) {
 	subs := make([]*runtimev1pb.TopicSubscription, 0)
 	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),
+			BulkSubscribe: convertBulkSubscribe(s.BulkSubscribe),
 		}
 		subs = append(subs, sub)
 	}
@@ -73,6 +82,17 @@
 	}
 }
 
+func convertBulkSubscribe(bulkSubscribe *internal.BulkSubscribe) *runtimev1pb.BulkSubscribeConfig {
+	if bulkSubscribe == nil {
+		return nil
+	}
+	return &runtimev1pb.BulkSubscribeConfig{
+		Enabled:            bulkSubscribe.Enabled,
+		MaxMessagesCount:   bulkSubscribe.MaxMessagesCount,
+		MaxAwaitDurationMs: bulkSubscribe.MaxAwaitDurationMs,
+	}
+}
+
 // OnTopicEvent fired whenever a message has been published to a topic that has been subscribed.
 // Dapr sends published messages in a CloudEvents v1.0 envelope.
 func (s *Server) OnTopicEvent(ctx context.Context, in *runtimev1pb.TopicEventRequest) (*runtimev1pb.TopicEventResponse, error) {

service/http/topic.go

@@ -42,9 +42,10 @@
 // topicEventJSON is identical to `common.TopicEvent`
 // except for it treats `data` as a json.RawMessage so it can
 // be used as bytes or interface{}.
+// Merge itemMap into topicEventJSON
 type topicEventJSON struct {
 	// ID identifies the event.
-	ID string `json:"id"`
+	ID string `json:"id"` // y
 	// The version of the CloudEvents specification.
 	SpecVersion string `json:"specversion"`
 	// The type of event related to the originating occurrence.
@@ -113,6 +114,29 @@
 	return data, rawData
 }
 
+type AppResponseStatus string
+
+const (
+	// Success means the message is received and processed correctly.
+	Success AppResponseStatus = "SUCCESS"
+	// Retry means the message is received but could not be processed and must be retried.
+	Retry AppResponseStatus = "RETRY"
+	// Drop means the message is received but should not be processed.
+	Drop AppResponseStatus = "DROP"
+)
+
+type BulkSubscribeResponseEntry struct {
+	// The id of the bulk subscribe entry
+	EntryId string `json:"entryId"`
+
+	// The response status of the bulk subscribe entry
+	Status AppResponseStatus `json:"status"`
+}
+
+type BulkSubscribeResponse struct {
+	Statuses []BulkSubscribeResponseEntry `json:"statuses"`
+}
+
 func (s *Server) registerBaseHandler() {
 	// register subscribe handler
 	f := func(w http.ResponseWriter, r *http.Request) {
@@ -301,9 +325,128 @@
 	return nil
 }
 
+func (s *Server) AddBulkTopicEventHandler(sub *common.Subscription, fn common.TopicEventHandler, maxMessagesCount, maxAwaitDurationMs int32) error {
+	if sub == nil {
+		return errors.New("subscription required")
+	}
+	// Route is only required for HTTP but should be specified for the
+	// app protocol to be interchangeable.
+	if sub.Route == "" {
+		return errors.New("handler route name")
+	}
+	if err := s.topicRegistrar.AddBulkSubscription(sub, fn, maxMessagesCount, maxAwaitDurationMs); err != nil {
+		return err
+	}
+
+	s.mux.Handle(sub.Route, optionsHandler(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			// check for post with no data
+			var (
+				body []byte
+				err  error
+			)
+			if r.Body != nil {
+				body, err = io.ReadAll(r.Body)
+				if err != nil {
+					http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+					return
+				}
+			}
+			if len(body) == 0 {
+				http.Error(w, "nil content", PubSubHandlerDropStatusCode)
+				return
+			}
+
+			// deserialize the event
+			var ins internal.BulkSubscribeEnvelope
+			if err = json.Unmarshal(body, &ins); err != nil {
+				http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+				return
+			}
+
+			statuses := make([]BulkSubscribeResponseEntry, 0, len(ins.Entries))
+
+			for _, entry := range ins.Entries {
+				itemJSON, err := json.Marshal(entry.Event)
+				if err != nil {
+					http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+					return
+				}
+				var in topicEventJSON
+
+				if err := json.Unmarshal(itemJSON, &in); err != nil {
+					http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+					return
+				}
+				if in.PubsubName == "" {
+					in.Topic = sub.PubsubName
+				}
+				if in.Topic == "" {
+					in.Topic = sub.Topic
+				}
+				data, rawData := in.getData()
+
+				te := common.TopicEvent{
+					ID:              in.ID,
+					SpecVersion:     in.SpecVersion,
+					Type:            in.Type,
+					Source:          in.Source,
+					DataContentType: in.DataContentType,
+					Data:            data,
+					RawData:         rawData,
+					DataBase64:      in.DataBase64,
+					Subject:         in.Subject,
+					PubsubName:      in.PubsubName,
+					Topic:           in.Topic,
+				}
+
+				retry, err := fn(r.Context(), &te)
+				if err == nil {
+					statuses = append(statuses, BulkSubscribeResponseEntry{
+						EntryId: entry.EntryId,
+						Status:  Success,
+					},
+					)
+				} else if retry {
+					statuses = append(statuses, BulkSubscribeResponseEntry{
+						EntryId: entry.EntryId,
+						Status:  Retry,
+					},
+					)
+				} else {
+					statuses = append(statuses, BulkSubscribeResponseEntry{
+						EntryId: entry.EntryId,
+						Status:  Drop,
+					},
+					)
+				}
+			}
+
+			resp := BulkSubscribeResponse{
+				Statuses: statuses,
+			}
+			if err != nil {
+				http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+				return
+			}
+			w.Header().Add("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+
+			writeBulkStatus(w, resp)
+		})))
+
+	return nil
+}
+
 func writeStatus(w http.ResponseWriter, s string) {
 	status := &common.SubscriptionResponse{Status: s}
 	if err := json.NewEncoder(w).Encode(status); err != nil {
 		http.Error(w, err.Error(), PubSubHandlerRetryStatusCode)
 	}
 }
+
+func writeBulkStatus(w http.ResponseWriter, s BulkSubscribeResponse) {
+	if err := json.NewEncoder(w).Encode(s); err != nil {
+		http.Error(w, err.Error(), PubSubHandlerRetryStatusCode)
+	}
+}

service/internal/topicregistrar.go

@@ -14,9 +14,9 @@
 
 // TopicRegistration encapsulates the subscription and handlers.
 type TopicRegistration struct {
-	Subscription   *TopicSubscription
-	DefaultHandler common.TopicEventHandler
-	RouteHandlers  map[string]common.TopicEventHandler
+	Subscription       *TopicSubscription
+	DefaultHandler     common.TopicEventHandler
+	RouteHandlers      map[string]common.TopicEventHandler
 }
 
 func (m TopicRegistrar) AddSubscription(sub *common.Subscription, fn common.TopicEventHandler) error {
@@ -62,3 +62,51 @@
 
 	return nil
 }
+
+func (m TopicRegistrar) AddBulkSubscription(sub *common.Subscription, fn common.TopicEventHandler, maxMessagesCount, maxAwaitDurationMs int32) error {
+	if sub.Topic == "" {
+		return errors.New("topic name required")
+	}
+	if sub.PubsubName == "" {
+		return errors.New("pub/sub name required")
+	}
+	if fn == nil {
+		return fmt.Errorf("topic handler required")
+	}
+
+	var key string
+	if !sub.DisableTopicValidation {
+		key = sub.PubsubName + "-" + sub.Topic
+	} else {
+		key = sub.PubsubName
+	}
+
+	ts, ok := m[key]
+	if !ok {
+		ts = &TopicRegistration{
+			Subscription:   NewTopicSubscription(sub.PubsubName, sub.Topic),
+			RouteHandlers:  make(map[string]common.TopicEventHandler),
+			DefaultHandler: nil,
+		}
+		ts.Subscription.SetMetadata(sub.Metadata)
+		m[key] = ts
+	}
+
+	ts.Subscription.SetBulkSubscribe(maxMessagesCount, maxAwaitDurationMs)
+
+	if sub.Match != "" {
+		if err := ts.Subscription.AddRoutingRule(sub.Route, sub.Match, sub.Priority); err != nil {
+			return err
+		}
+	} else {
+		if err := ts.Subscription.SetDefaultRoute(sub.Route); err != nil {
+			return err
+		}
+
+		ts.DefaultHandler = fn
+	}
+
+	ts.RouteHandlers[sub.Route] = fn
+
+	return nil
+}

service/internal/topicsubscription.go

@@ -18,6 +18,31 @@
 	Routes *TopicRoutes `json:"routes,omitempty"`
 	// Metadata is the subscription metadata.
 	Metadata map[string]string `json:"metadata,omitempty"`
+	// bulksubsribe
+	BulkSubscribe *BulkSubscribe `json:"bulkSubscribe,omitempty"`
+}
+
+type BulkSubscribeMessageItem struct {
+	EntryId     string            `json:"entryId"` //nolint:stylecheck
+	Event       interface{}       `json:"event"`
+	Metadata    map[string]string `json:"metadata"`
+	ContentType string            `json:"contentType,omitempty"`
+}
+
+type BulkSubscribeEnvelope struct {
+	ID        string
+	Entries   []BulkSubscribeMessageItem
+	Metadata  map[string]string
+	Topic     string
+	Pubsub    string
+	EventType string
+}
+
+
+type BulkSubscribe struct {
+	Enabled            bool  `json:"enabled"`
+	MaxMessagesCount   int32 `json:"maxMessagesCount,omitempty"`
+	MaxAwaitDurationMs int32 `json:"maxAwaitDurationMs,omitempty"`
 }
 
 // TopicRoutes encapsulates the default route and multiple routing rules.
@@ -60,6 +85,19 @@
 	return nil
 }
 
+func (s *TopicSubscription) SetBulkSubscribe(maxMessagesCount,maxAwaitDurationMs int32) error {
+	if s.BulkSubscribe != nil {
+		return fmt.Errorf("subscription for topic %s on pubsub %s already has bulkSubscribe set", s.Topic, s.PubsubName)
+	}
+	s.BulkSubscribe = &BulkSubscribe{
+		Enabled:            true,
+		MaxMessagesCount:   maxMessagesCount,
+		MaxAwaitDurationMs: maxAwaitDurationMs,
+	}
+
+	return nil
+}
+
 // SetDefaultRoute sets the default route if not already set.
 // An error is returned if it is already set.
 func (s *TopicSubscription) SetDefaultRoute(path string) error {