Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Feature/implement autopaho rpc #98

Merged
merged 13 commits into from
Aug 26, 2022
Merged

Feature/implement autopaho rpc #98

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
2db2542
Fix router unit test for shared subscriptions
bkneis Aug 9, 2022
e4fda94
Add build commands to makefile and ignore bin
bkneis Aug 9, 2022
e818d6a
Fix paho RPC extension using context.Background
bkneis Aug 9, 2022
6eedf91
Implement RPC extension for autopaho
bkneis Aug 9, 2022
8780381
Add function to use the underlying client in connection manager in a …
bkneis Aug 9, 2022
e66bbf6
Remove .env from rpc example
bkneis Aug 15, 2022
1b134ad
Remove UseClient
bkneis Aug 15, 2022
10f9f48
Update RPC example with passing router through client
bkneis Aug 15, 2022
e095ad9
Implement PR feedback
bkneis Aug 16, 2022
5549e39
Update reference to response topic in autopaho rpc extension
bkneis Aug 16, 2022
3d51531
Rename to ResponseTopicFmt
bkneis Aug 16, 2022
ace041a
Remove redudant need to pass router
bkneis Aug 19, 2022
c650e72
Prevent blocking on recieving a response by using a select with context
bkneis Aug 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions autopaho/extensions/rpc/rpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func (h *Handler) getCorrelIDChan(cID string) chan *paho.Publish {
return rChan
}

func (h *Handler) Request(ctx context.Context, pb *paho.Publish) (*paho.Publish, error) {
func (h *Handler) Request(ctx context.Context, pb *paho.Publish) (resp *paho.Publish, err error) {
cID := fmt.Sprintf("%d", time.Now().UnixNano())
rChan := make(chan *paho.Publish)

Expand All @@ -79,13 +79,17 @@ func (h *Handler) Request(ctx context.Context, pb *paho.Publish) (*paho.Publish,
pb.Properties.ResponseTopic = h.responseTopic
pb.Retain = false

_, err := h.cm.Publish(ctx, pb)
_, err = h.cm.Publish(ctx, pb)
if err != nil {
return nil, err
}

resp := <-rChan
return resp, nil
select {
case <-ctx.Done():
return nil, fmt.Errorf("context ended")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it would be better to return ctx.Error() here (its fairly common to check for context errors).

case resp = <-rChan:
return resp, nil
}
}

func (h *Handler) responseHandler(pb *paho.Publish) {
Expand Down