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

Fix double close in stream client #2693

Merged
merged 10 commits into from
Feb 15, 2024
Merged

Fix double close in stream client #2693

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
21bac54
[fix] etcd config source prefix issue (#2389)
xpunch Dec 17, 2021
5a2f37c
http transport data race issue (#2436)
xpunch Feb 23, 2022
c20c971
Merge remote-tracking branch 'upstream/v3'
Ak-Army Jul 14, 2022
62f73ac
Merge remote-tracking branch 'upstream/master'
Ak-Army Jul 15, 2022
735aee8
[feature] Ability to close connection while receiving.
Ak-Army Jul 15, 2022
7bb5b6b
Merge branch 'asim:master' into master
Ak-Army Jul 22, 2022
bcb7c20
Merge branch 'go-micro:master' into master
Ak-Army Sep 14, 2023
7332184
Merge branch 'go-micro:master' into master
Ak-Army Jan 15, 2024
4f60aec
Merge branch 'go-micro:master' into master
Ak-Army Feb 15, 2024
ddc2279
[fix] Do not close the transport client twice in stream connection , …
Ak-Army Feb 15, 2024
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
26 changes: 16 additions & 10 deletions client/rpc_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,13 @@ func (r *rpcClient) newCodec(contentType string) (codec.NewCodec, error) {
return nil, fmt.Errorf("unsupported Content-Type: %s", contentType)
}

func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request, resp interface{}, opts CallOptions) error {
func (r *rpcClient) call(
ctx context.Context,
node *registry.Node,
req Request,
resp interface{},
opts CallOptions,
) error {
address := node.Address
logger := r.Options().Logger

Expand Down Expand Up @@ -292,12 +298,6 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
r.codec = codec
}

releaseFunc := func(_ error) {
if err = c.Close(); err != nil {
logger.Log(log.ErrorLevel, err)
}
}

stream := &rpcStream{
id: id,
context: ctx,
Expand All @@ -308,7 +308,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
closed: make(chan bool),
// signal the end of stream,
sendEOS: true,
release: releaseFunc,
release: func(_ error) {},
}

// wait for error response
Expand Down Expand Up @@ -490,7 +490,10 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
return merrors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
}

return merrors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
return merrors.InternalServerError("go.micro.client",
"error getting next %s node: %s",
service,
err.Error())
}

// make the call
Expand Down Expand Up @@ -586,7 +589,10 @@ func (r *rpcClient) Stream(ctx context.Context, request Request, opts ...CallOpt
return nil, merrors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
}

return nil, merrors.InternalServerError("go.micro.client", "error getting next %s node: %s", service, err.Error())
return nil, merrors.InternalServerError("go.micro.client",
"error getting next %s node: %s",
service,
err.Error())
}

stream, err := r.stream(ctx, node, request, callOpts)
Expand Down
Loading