feat: bulksubscribe http

examples/pubsub/pub/pub.go

@@ -23,8 +23,9 @@ import (
 
 var (
 	// set the environment as instructions.
-	pubsubName = os.Getenv("DAPR_PUBSUB_NAME")
-	topicName  = "neworder"
+	pubsubName    = os.Getenv("DAPR_PUBSUB_NAME")
+	topicName     = "neworder"
+	bulkTopicName = "newbulkorder"
 )
 
 func main() {
@@ -44,7 +45,7 @@ func main() {
 	}
 
 	// Publish multiple events
-	if res := client.PublishEvents(ctx, pubsubName, topicName, publishEventsData); res.Error != nil {
+	if res := client.PublishEvents(ctx, pubsubName, bulkTopicName, publishEventsData); res.Error != nil {
 		panic(err)
 	}
 

examples/pubsub/sub/sub.go

@@ -35,6 +35,12 @@ var defaultSubscription = &common.Subscription{
 	Route:      "/orders",
 }
 
+var bulkSubscription = &common.Subscription{
+	PubsubName: "bulkmessages",
+	Topic:      "newbulkorder",
+	Route:      "/bulkorders",
+}
+
 var importantSubscription = &common.Subscription{
 	PubsubName: "messages",
 	Topic:      "neworder",
@@ -46,11 +52,11 @@ var importantSubscription = &common.Subscription{
 func main() {
 	s := daprd.NewService(":8080")
 
-	if err := s.AddTopicEventHandler(defaultSubscription, eventHandler); err != nil {
+	if err := s.AddBulkTopicEventHandler(defaultSubscription, eventHandler, 10, 100); err != nil {
 		log.Fatalf("error adding topic subscription: %v", err)
 	}
 
-	if err := s.AddTopicEventHandler(importantSubscription, importantEventHandler); err != nil {
+	if err := s.AddTopicEventHandler(defaultSubscription, eventHandler); err != nil {
 		log.Fatalf("error adding topic subscription: %v", err)
 	}
 

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) {