feat: bulksubscribe http

examples/pubsub/sub/sub.go

@@ -46,12 +46,16 @@ var importantSubscription = &common.Subscription{
 func main() {
 	s := daprd.NewService(":8080")
 
-	if err := s.AddTopicEventHandler(defaultSubscription, eventHandler); err != nil {
-		log.Fatalf("error adding topic subscription: %v", err)
-	}
+	bulkSubscribe := true
 
-	if err := s.AddTopicEventHandler(importantSubscription, importantEventHandler); err != nil {
-		log.Fatalf("error adding topic subscription: %v", err)
+	if bulkSubscribe {
+		if err := s.AddBulkTopicEventHandler(defaultSubscription, eventHandler, 10, 100); err != nil {
+			log.Fatalf("error adding topic subscription: %v", err)
+		}
+	} else {
+		if err := s.AddTopicEventHandler(defaultSubscription, eventHandler); err != nil {
+			log.Fatalf("error adding topic subscription: %v", err)
+		}
 	}
 
 	if err := s.Start(); err != nil && err != http.ErrServerClosed {

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 @@ func (s *Server) AddTopicEventHandler(sub *common.Subscription, fn common.TopicE
 	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 convertRoutes(routes *internal.TopicRoutes) *runtimev1pb.TopicRoutes {
 	}
 }
 
+func convertBulkSubscribe(bulkSubscribe *internal.BulkSubscribeOptions) *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) {