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,6 +42,7 @@
 // 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"`
@@ -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"` //nolint:stylecheck
+
+	// 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, entryErr := json.Marshal(entry.Event)
+				if entryErr != nil {
+					http.Error(w, entryErr.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, funcErr := fn(r.Context(), &te)
+				if funcErr == 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)
+	}
+}