Add status watcher for bayeux client #10

kush-elastic · 2022-04-19T13:57:28Z

Add client to watch over status of subscriptions. To check how many connections bayeux instance has.
Make sure to close channel as soon as context is canceled to ensure it won't hold channel open for forever.

elasticmachine · 2022-04-19T14:00:13Z

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

Start Time: 2022-04-21T10:38:20.680+0000
Duration: 2 min 23 sec

🤖 GitHub comments

To re-run your PR in the CI, just comment with:

/test : Re-trigger the build.

mtojek

Please rephrase the issue description to present why we need this change/what kind of issue it fixes, and link it with the other issue.

mtojek · 2022-04-20T08:26:39Z

bayeux.go

+}
+
+func (st *Status) disconnect(out chan MaybeMsg) {
+	close(out)


That's an anti-pattern in general, you shouldn't close channels you don't own/create.

Sure, I will update it.

mtojek · 2022-04-20T08:26:59Z

bayeux.go

-	id    clientIDAndCookies
+	creds  Credentials
+	id     clientIDAndCookies
+	status Status


Does the Status need to be exposed?

We don't really need it exposed.
Status was already exposed, I am just reusing it to get connection information from bayeux object.

Ok, let's hide it then if it isn't necessary.

mtojek · 2022-04-20T08:28:02Z

bayeux.go

 		defer wg.Done()
-		defer close(out)
+		defer status.disconnect(out)


You have 3 deferred functions now, maybe you can replace it with a block defer func() { ... }?

I will update it to:
defer func() { close(out) wg.Done() status.disconnect() }()

mtojek · 2022-04-20T08:28:48Z

bayeux.go

@@ -266,7 +295,12 @@ func (b *Bayeux) connect(ctx context.Context, out chan MaybeMsg) chan MaybeMsg {
 						return
 					}
 					for _, e := range x {
-						out <- MaybeMsg{Msg: e}


Could you please add any comments regarding this change?

Sure, Thanks

mtojek · 2022-04-20T08:30:53Z

bayeux.go

@@ -298,6 +332,7 @@ func GetSalesforceCredentials(ap AuthenticationParameters) (creds *Credentials,

 func (b *Bayeux) Channel(ctx context.Context, out chan MaybeMsg, r string, creds Credentials, channel string) chan MaybeMsg {
 	b.creds = creds
+	b.status = status


Could you please explain what are you trying to achieve? In L132 the status is global (which is not good in general), but here you're setting it as part of the Bayeux?

I have added one parameter connectCount in Status struct. I need to ensure if connections are closed as soon as context is canceled from filebeat side(for test case to ensure if channel is closing or not). To do that I need some mechanism in bayeux. but i don't have any way to access status object so for reference I used bayeux object.

Are you sure that this code doesn't need any concurrency safety mechanism? For example in MQTT input, I used a WaitGroup.

What do you think of this for concurrency safety mechanism for in flight messages. as we can only close channel from bayeux side. we will wait till all the inbound messages are sent via channel.

var waitMsgs sync.WaitGroup wg.Add(1) go func() { defer func() { waitMsgs.Wait() close(out) status.disconnect() wg.Done() }() for { for _, e := range x { waitMsgs.Add(1) go func() { defer waitMsgs.Done() // close channel as soon as context is canceled to ensure it won't hold channel open for forever out <- MaybeMsg{Msg: e} }() } } }() return out

It's hard for me to tell, as the entire library code is tangled a bit. I can't confirm that this change doesn't introduce any other issues (deadlocks). What I don't like about the suggested approach is for-in-for loop.

WDYT about refactoring the library code or writing it from scratch? We may try to get rid of all globals and focus on concurrency safety.

Oh, sorry about that, I removed other things there to make it simple to understand but actual snip should look like this:

func (b *Bayeux) connect(ctx context.Context, out chan MaybeMsg) chan MaybeMsg { var waitMsgs sync.WaitGroup wg.Add(1) go func() { defer func() { waitMsgs.Wait() close(out) status.disconnect() wg.Done() }() for { select { case <-ctx.Done(): return default: postBody := fmt.Sprintf(`{"channel": "/meta/connect", "connectionType": "long-polling", "clientId": "%s"} `, b.id.clientID) resp, err := b.call(ctx, postBody, b.creds.bayeuxUrl()) if err != nil { if errors.Is(err, context.Canceled) { return } out <- MaybeMsg{Err: fmt.Errorf("cannot connect to bayeux: %s, trying again", err)} } else { if os.Getenv("DEBUG") != "" { var b []byte if resp.Body != nil { b, _ = ioutil.ReadAll(resp.Body) } // Restore the io.ReadCloser to its original state resp.Body = ioutil.NopCloser(bytes.NewBuffer(b)) // Use the content s := string(b) logger.Printf("Response Body: %s", s) } var x []TriggerEvent decoder := json.NewDecoder(resp.Body) if err := decoder.Decode(&x); err != nil && err == io.EOF { out <- MaybeMsg{Err: err} return } for _, e := range x { waitMsgs.Add(1) go func() { defer waitMsgs.Done() // close channel as soon as context is canceled to ensure it won't hold channel open for forever out <- MaybeMsg{Msg: e} }() } } } } }() return out }

In theory, it looks good to me, let's get it into the PR code.

mtojek · 2022-04-20T10:01:04Z

bayeux.go

+}
+
+type Client struct {
+	BayOb Bayeux


What does Ob stand for?

let me update it to something meaningful.

mtojek · 2022-04-21T08:25:21Z

bayeux.go

@@ -298,6 +332,7 @@ func GetSalesforceCredentials(ap AuthenticationParameters) (creds *Credentials,

 func (b *Bayeux) Channel(ctx context.Context, out chan MaybeMsg, r string, creds Credentials, channel string) chan MaybeMsg {
 	b.creds = creds
+	b.status = status


In theory, it looks good to me, let's get it into the PR code.

mtojek · 2022-04-21T08:28:15Z

bayeux.go

@@ -298,6 +335,7 @@ func GetSalesforceCredentials(ap AuthenticationParameters) (creds *Credentials,

 func (b *Bayeux) Channel(ctx context.Context, out chan MaybeMsg, r string, creds Credentials, channel string) chan MaybeMsg {
 	b.creds = creds
+	b.status = status


Back to the original question:

Why you can't access the global "status"? If you need to expose the global state, just expose a static function.

add status watcher for bayeux client

234f58a

kush-elastic added the enhancement New feature or request label Apr 19, 2022

kush-elastic requested review from cmacknz and mtojek April 19, 2022 13:57

kush-elastic self-assigned this Apr 19, 2022

mtojek suggested changes Apr 20, 2022

View reviewed changes

kush-elastic changed the title ~~add status watcher for bayeux client~~ Add status watcher for bayeux client Apr 20, 2022

kush-elastic requested a review from mtojek April 20, 2022 09:08

Resolve review comments

a3c3e97

mtojek reviewed Apr 20, 2022

View reviewed changes

update client struct

da6ebc6

mtojek reviewed Apr 21, 2022

View reviewed changes

kush-elastic added 2 commits April 21, 2022 15:13

resolve comments

4e2690c

Add wait group for in flight messages

8099bf9

kush-elastic requested a review from mtojek April 21, 2022 10:38

mtojek approved these changes Apr 21, 2022

View reviewed changes

kush-elastic merged commit acdef69 into elastic:master Apr 21, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add status watcher for bayeux client #10

Add status watcher for bayeux client #10

kush-elastic commented Apr 19, 2022 •

edited

Loading

elasticmachine commented Apr 19, 2022 •

edited

Loading

Build stats

mtojek left a comment

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022 •

edited

Loading

mtojek Apr 20, 2022

kush-elastic Apr 21, 2022 •

edited by mtojek

Loading

mtojek Apr 21, 2022

mtojek Apr 20, 2022

kush-elastic Apr 20, 2022

kush-elastic Apr 20, 2022

mtojek Apr 21, 2022

mtojek Apr 21, 2022

kush-elastic Apr 21, 2022

Add status watcher for bayeux client #10

Add status watcher for bayeux client #10

Conversation

kush-elastic commented Apr 19, 2022 • edited Loading

elasticmachine commented Apr 19, 2022 • edited Loading

💚 Build Succeeded

Build stats

🤖 GitHub comments

mtojek 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

Choose a reason for hiding this comment

kush-elastic Apr 20, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kush-elastic Apr 21, 2022 • edited by mtojek Loading

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

kush-elastic commented Apr 19, 2022 •

edited

Loading

elasticmachine commented Apr 19, 2022 •

edited

Loading

kush-elastic Apr 20, 2022 •

edited

Loading

kush-elastic Apr 21, 2022 •

edited by mtojek

Loading