feat: bulksubscribe http #478

sadath-12 · 2023-12-02T07:42:15Z

Fixes #423 for http bulksubscribe

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

examples/pubsub/sub/sub.go

service/http/topic.go

codecov · 2023-12-04T05:22:25Z

Codecov Report

Attention: Patch coverage is 75.36232% with 34 lines in your changes are missing coverage. Please review.

Project coverage is 58.63%. Comparing base (27248ba) to head (300e5ac).
Report is 11 commits behind head on main.

Files	Patch %	Lines
service/http/topic.go	66.66%	21 Missing and 7 partials ⚠️
service/internal/topicregistrar.go	86.66%	2 Missing and 2 partials ⚠️
service/internal/topicsubscription.go	75.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #478      +/-   ##
==========================================
+ Coverage   58.04%   58.63%   +0.58%     
==========================================
  Files          55       55              
  Lines        3568     3701     +133     
==========================================
+ Hits         2071     2170      +99     
- Misses       1375     1399      +24     
- Partials      122      132      +10

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

daixiang0 · 2023-12-06T02:43:35Z

Codes look good, please add unit tests.

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 · 2023-12-06T07:22:22Z

Added it 🙂

mikeee

Although the limited tests are passing and likewise with the validator, I'd like to see a validated bulk subscription.

examples/pubsub/sub/sub.go

service/grpc/topic.go

service/internal/topicsubscription.go

service/http/topic_test.go

service/internal/topicsubscription.go

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 · 2023-12-07T19:26:35Z

Thank you for the review @mikeee . Added them

examples/pubsub/sub/sub.go

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

daixiang0 · 2023-12-18T07:39:19Z

Please fix conflicts.

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 · 2023-12-18T18:12:19Z

Done @daixiang0

mikeee

There are regressions with the tests moving back to assert rather than require when evaluating errors

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee

go-sdk/examples/pubsub/README.md

Needs to be updated inline with your modifications to the pubsub example

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

daixiang0 · 2023-12-28T02:19:50Z

Please make CI happy.

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 · 2023-12-28T07:56:46Z

Yup @daixiang0 it seems to be chilling now . 😅 Thanks

examples/pubsub/README.md

examples/pubsub/pub/pub.go

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee

A few quick-win tests can be implemented on the code to improve coverage on top of the identified which I've marked

examples/pubsub/README.md

service/grpc/topic.go

service/grpc/topic_test.go

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?

I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

mikeee · 2024-01-05T18:34:05Z

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 {


As my previous review - the validation of arguments passed to these parameters should be implemented as per the implementation spec. I do think that this is something we need to address both sdk-side and in the runtime explicitly as part of best practice.

what default values you suggest if nil values are given?

Error-out if either are X <= 0 this would prevent a negative value, a nil value is not possible for the int type.

mikeee · 2024-01-05T18:45:02Z

service/grpc/topic_test.go

 func TestTopicErrors(t *testing.T) {
 	server := getTestServer()
 	err := server.AddTopicEventHandler(nil, nil)
-	require.Errorf(t, err, "expected error on nil sub")
+	require.Error(t, err, "expected error on nil sub with AddTopicEventHandler")
+
+	err = server.AddBulkTopicEventHandler(nil, nil, 0, 0)
+	require.Error(t, err, "expected error on nil sub with AddBulkTopicEventHandler")

 	sub := &common.Subscription{}
 	err = server.AddTopicEventHandler(sub, nil)
-	require.Errorf(t, err, "expected error on invalid sub")
+	require.Error(t, err, "expected error on invalid sub with AddTopicEventHandler")
+	err = server.AddBulkTopicEventHandler(sub, nil, 0, 0)
+	require.Error(t, err, "expected error on invalid sub with AddBulkTopicEventHandler")

 	sub.PubsubName = "messages"
 	err = server.AddTopicEventHandler(sub, nil)
-	require.Errorf(t, err, "expected error on sub without topic")
+	require.Error(t, err, "expected error on sub without topic with AddTopicEventHandler")
+	sub.PubsubName = "messages"
+	err = server.AddBulkTopicEventHandler(sub, nil, 0, 0)
+	require.Error(t, err, "expected error on sub without topic with AddBulkTopicEventHandler")

 	sub.Topic = "test"
 	err = server.AddTopicEventHandler(sub, nil)
-	require.Errorf(t, err, "expected error on sub without handler")
+	require.Error(t, err, "expected error on sub without handler")
+	err = server.AddBulkTopicEventHandler(sub, nil, 0, 0)
+	require.Error(t, err, "expected error on sub without handler")


Break this down into idiomatic sub-tests, you can share the setup steps here but it doesn't read well when running tests.

mikeee · 2024-01-05T18:54:46Z

examples/pubsub/README.md

@@ -44,10 +48,9 @@ expected_stdout_lines:
 background: true
 sleep: 15
 -->
+#### Note: pub/pub.go contains both PublishEvents (used for publish of messages) and PublishEvent (used for bulkPublish of messages) 


mikeee · 2024-01-05T18:54:54Z

examples/pubsub/README.md

+#### Note: sub/sub.go contains both AddTopicEventHandler (used for subscribe of messages) and AddBulkTopicEventHandler (used for bulksubscribe of messages)
+


go-sdk/examples/pubsub/README.md

Line 3 in de720d9

This folder contains two Go files that use the Go SDK to invoke the Dapr Pub/Sub API.

Remove this and modify the above to include something along the lines of:

This folder contains a publisher and subscriber that demonstrates both standard pubsub and bulk-pubsub

mikeee · 2024-01-05T18:59:05Z

examples/pubsub/sub/sub.go

@@ -35,12 +35,10 @@ var defaultSubscription = &common.Subscription{
 	Route:      "/orders",
 }

-var importantSubscription = &common.Subscription{


I'd like to see the importantSubscription re-implemented as a way to validate the route being correct for a message on the same topic alongside match/priority

mikeee · 2024-01-05T20:23:31Z

service/http/topic.go

+			if err != nil {
+				http.Error(w, err.Error(), PubSubHandlerDropStatusCode)
+				return
+			}


There is no unhandled error at this point, could you clarify that if a single event is dropped it will be replayed/retried at a later date?

mikeee · 2024-01-05T20:26:11Z

service/http/topic.go

+func writeBulkStatus(w http.ResponseWriter, s BulkSubscribeResponse) {
+	if err := json.NewEncoder(w).Encode(s); err != nil {
+		http.Error(w, err.Error(), PubSubHandlerRetryStatusCode)
+	}
+}


It's fine to create a new function as it is manifestly different or is idiomatic, do you think this is better or would it be better to pass the slice to your function as an argument and wrap it within the function?

mikeee · 2024-01-05T20:30:19Z

service/http/topic_test.go

+	if assert.Len(t, subs, 2, "unexpected subscription count") {
+		assert.Equal(t, "messages", subs[0].PubsubName)
+		assert.Equal(t, "errors", subs[0].Topic)
+
+		assert.Equal(t, "messages", subs[1].PubsubName)
+		assert.Equal(t, "test", subs[1].Topic)
+		assert.Equal(t, "", subs[1].Route)
+		assert.Equal(t, "/", subs[1].Routes.Default)
+		if assert.Len(t, subs[1].Routes.Rules, 1, "unexpected rules count") {
+			assert.Equal(t, `event.type == "other"`, subs[1].Routes.Rules[0].Match)
+			assert.Equal(t, "/other", subs[1].Routes.Rules[0].Path)
+		}
+	}


This is quite confusing to read - would it be more idiomatic to create a known map and compare this to the subscriptions?

mikeee · 2024-01-05T20:32:28Z

service/http/topic_test.go

+func TestAddingInvalidBulkEventHandlers(t *testing.T) {
+	s := newServer("", nil)
+	err := s.AddBulkTopicEventHandler(nil, testTopicFunc, 10, 1000)
+	require.Error(t, err, "expected error adding no sub event handler")
+
+	sub := &common.Subscription{Metadata: map[string]string{}}
+	err = s.AddBulkTopicEventHandler(sub, testTopicFunc, 10, 1000)
+	require.Error(t, err, "expected error adding empty sub event handler")
+
+	sub.Topic = "test"
+	err = s.AddBulkTopicEventHandler(sub, testTopicFunc, 10, 1000)
+	require.Error(t, err, "expected error adding sub without component event handler")
+
+	sub.PubsubName = "messages"
+	err = s.AddBulkTopicEventHandler(sub, testTopicFunc, 10, 1000)
+	require.Error(t, err, "expected error adding sub without route event handler")
 }


Please split this into subtests

mikeee · 2024-01-05T20:33:53Z

service/internal/topicsubscription_test.go

Provide negative tests

sadath-12 · 2024-01-06T04:58:45Z

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?

I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

can we keep the migration in separate pr ?

mikeee · 2024-01-06T12:49:27Z

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?
I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

can we keep the migration in separate pr ?

I have to answer that with another question - has the bulk pub/sub been validated elsewhere? All I'm seeing is fallbacks.

sadath-12 · 2024-01-06T16:27:09Z

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?
I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

can we keep the migration in separate pr ?

I have to answer that with another question - has the bulk pub/sub been validated elsewhere? All I'm seeing is fallbacks.

sure no issues lets test it for others as well . Before I jump wanted to confirm what you mean by fallback here ? because when I see the demo of bulksubscribe in the dapr docs and how js-sdk has implemented bulksubscribe . I have made sure similar behaviour is achieved here . would you like to explain what do you expect clearly ? Happy to implement whatever works best for the project 😊 If possible maybe we could drive the talk on the discord as well since some discussion is required on this

mikeee · 2024-01-06T17:48:30Z

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?
I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

can we keep the migration in separate pr ?

I have to answer that with another question - has the bulk pub/sub been validated elsewhere? All I'm seeing is fallbacks.

sure no issues lets test it for others as well . Before I jump wanted to confirm what you mean by fallback here ? because when I see the demo of bulksubscribe in the dapr docs and how js-sdk has implemented bulksubscribe . I have made sure similar behaviour is achieved here . would you like to explain what do you expect clearly ? Happy to implement whatever works best for the project 😊 If possible maybe we could drive the talk on the discord as well since some discussion is required on this

The issue is that since the broker used does not implement bulk pub/sub methods, we are effectively dropping down to single pub/sub. Looking at the validation run it is highlighted in the logs, likewise in the js-sdk I note this is an issue too as it uses rabbitmq but this is not validated so you'd run into the same result if run in debug mode.

sadath-12 · 2024-01-06T18:57:07Z

I've missed this over the past few reviews but Redis is not suitable for testing bulk pub/sub. Would it be wise to migrate to another broker (for example kafka or I believe ASB) so that we can validate this example?
I'd like to see a validated run that doesn't fallback to singular pub/sub before I review again?

can we keep the migration in separate pr ?

I have to answer that with another question - has the bulk pub/sub been validated elsewhere? All I'm seeing is fallbacks.

sure no issues lets test it for others as well . Before I jump wanted to confirm what you mean by fallback here ? because when I see the demo of bulksubscribe in the dapr docs and how js-sdk has implemented bulksubscribe . I have made sure similar behaviour is achieved here . would you like to explain what do you expect clearly ? Happy to implement whatever works best for the project 😊 If possible maybe we could drive the talk on the discord as well since some discussion is required on this

The issue is that since the broker used does not implement bulk pub/sub methods, we are effectively dropping down to single pub/sub. Looking at the validation run it is highlighted in the logs, likewise in the js-sdk I note this is an issue too as it uses rabbitmq but this is not validated so you'd run into the same result if run in debug mode.

Ya the dapr itself sends back the response one by one . so with all those brokers we would get the same result right ? we run the callback for each entry and send response to dapr

mikeee · 2024-02-20T15:41:50Z

examples/pubsub/pub/pub.go

@@ -44,7 +44,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)


This doesn't seem entirely right, you're panicking here with an unrelated (potentially) nil error

sadath-12 · 2024-02-21T08:55:24Z

@mikeee what you want to say about this approach are we going up with this pr after those things resolved ?

mikeee · 2024-02-21T09:09:46Z

@mikeee what you want to say about this approach are we going up with this pr after those things resolved ?

I think we should definitely try to get this merged even as a fallback validation, as long as there is validation locally just to make sure it's working then that should be fine 👍

Eileen-Yu · 2024-03-22T20:49:46Z

wondering if there's any progress on this PR?

I'm happy to help finish this one. Would you please resolve those conflicts if you'd like me to keep your commits? @sadath-12

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 · 2024-05-15T04:06:04Z

@Eileen-Yu

feat: bulksubscribe http

3f5bcd8

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 requested a review from a team as a code owner December 2, 2023 07:42

sadath-12 added 6 commits December 2, 2023 13:51

minor changes

1c45f4e

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

data retrive fix

2193f47

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

bulkresponse

5d4e498

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

reformat

f19a505

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

bulksub success

fdd425a

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

feat: bulksubscribe support

cb26d03

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 mentioned this pull request Dec 3, 2023

Bulk Pubsub implementation in SDK dapr/python-sdk#464

Open

daixiang0 reviewed Dec 4, 2023

View reviewed changes

examples/pubsub/sub/sub.go Outdated Show resolved Hide resolved

daixiang0 reviewed Dec 4, 2023

View reviewed changes

service/http/topic.go Outdated Show resolved Hide resolved

daixiang0 reviewed Dec 4, 2023

View reviewed changes

service/http/topic.go Show resolved Hide resolved

sadath-12 added 2 commits December 4, 2023 17:29

lint fix

597c099

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

lint fix

10d2c67

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

bulksubscribe unit tests

e218130

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee requested changes Dec 7, 2023

View reviewed changes

sadath-12 added 2 commits December 7, 2023 23:46

enhancements

5131387

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

test: added unit tests

81b72ef

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee reviewed Dec 7, 2023

View reviewed changes

examples/pubsub/sub/sub.go Outdated Show resolved Hide resolved

minor advances

203d31c

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 added 2 commits December 18, 2023 23:38

merge main

068ef1c

minor adv

36b78f5

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee requested changes Dec 19, 2023

View reviewed changes

require.Error

38bc154

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee requested changes Dec 23, 2023

View reviewed changes

updated readme.md

0b41a28

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 added 6 commits December 28, 2023 10:17

fix-lint

d05856f

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

lintfix

92e7eef

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

fixlint

f315c9e

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

correct sub.go

0f2d9bb

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

examples correction

f4a3b65

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

markdown

3577081

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee reviewed Dec 28, 2023

View reviewed changes

examples/pubsub/README.md Show resolved Hide resolved

mikeee reviewed Dec 28, 2023

View reviewed changes

examples/pubsub/pub/pub.go Outdated Show resolved Hide resolved

remove unused

5e1f64d

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

mikeee reviewed Dec 29, 2023

View reviewed changes

examples/pubsub/README.md Show resolved Hide resolved

service/grpc/topic.go Show resolved Hide resolved

service/grpc/topic_test.go Outdated Show resolved Hide resolved

remove alias

de720d9

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

daixiang0 requested a review from mikeee January 5, 2024 06:16

mikeee requested changes Jan 5, 2024

View reviewed changes

mikeee requested changes Feb 20, 2024

View reviewed changes

Merge branch 'main' into bulksubscribe1

05f3ad7

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

sadath-12 requested a review from a team as a code owner May 14, 2024 13:26

lint fix

300e5ac

Signed-off-by: sadath-12 <sadathsadu2002@gmail.com>

		#### Note: sub/sub.go contains both AddTopicEventHandler (used for subscribe of messages) and AddBulkTopicEventHandler (used for bulksubscribe of messages)

feat: bulksubscribe http #478

Are you sure you want to change the base?

feat: bulksubscribe http #478

Conversation

sadath-12 commented Dec 2, 2023

codecov bot commented Dec 4, 2023 • edited

Codecov Report

daixiang0 commented Dec 6, 2023

sadath-12 commented Dec 6, 2023

mikeee left a comment

Choose a reason for hiding this comment

sadath-12 commented Dec 7, 2023

daixiang0 commented Dec 18, 2023

sadath-12 commented Dec 18, 2023

mikeee left a comment

Choose a reason for hiding this comment

mikeee left a comment

Choose a reason for hiding this comment

daixiang0 commented Dec 28, 2023

sadath-12 commented Dec 28, 2023

mikeee left a comment

Choose a reason for hiding this comment

mikeee left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sadath-12 commented Jan 6, 2024

mikeee commented Jan 6, 2024

sadath-12 commented Jan 6, 2024 • edited

mikeee commented Jan 6, 2024

sadath-12 commented Jan 6, 2024 • edited

Choose a reason for hiding this comment

sadath-12 commented Feb 21, 2024

mikeee commented Feb 21, 2024

Eileen-Yu commented Mar 22, 2024 • edited

sadath-12 commented May 15, 2024

codecov bot commented Dec 4, 2023 •

edited

sadath-12 commented Jan 6, 2024 •

edited

sadath-12 commented Jan 6, 2024 •

edited

Eileen-Yu commented Mar 22, 2024 •

edited