summaryrefslogtreecommitdiff
path: root/vendor/google.golang.org/grpc/stream.go
diff options
context:
space:
mode:
authorLibravatar Daenney <daenney@users.noreply.github.com>2023-09-07 13:20:37 +0200
committerLibravatar GitHub <noreply@github.com>2023-09-07 13:20:37 +0200
commit14ef09809942800db57de87fe3963770a56b585b (patch)
tree89e95f21145bc7ad5745f77be1d998faa9c09695 /vendor/google.golang.org/grpc/stream.go
parent[bugfix] fix checks for deref the same status descendants / ascendants (#2181) (diff)
downloadgotosocial-14ef09809942800db57de87fe3963770a56b585b.tar.xz
[feature] Support OTLP HTTP, drop Jaeger (#2184)
* [feature] Add http trace exporter, drop Jaeger Jaeger supports ingesting traces using the OpenTelemetry gRPC or HTTP methods. The Jaeger project has deprecated the old jaeger transport. * Add support for submitting traces over HTTP * Drop support for the old Jaeger protocol * Upgrade the trace libraries to v1.17 Fixes: #2176 Fixes: #2179
Diffstat (limited to 'vendor/google.golang.org/grpc/stream.go')
-rw-r--r--vendor/google.golang.org/grpc/stream.go33
1 files changed, 23 insertions, 10 deletions
diff --git a/vendor/google.golang.org/grpc/stream.go b/vendor/google.golang.org/grpc/stream.go
index d1226a412..de32a7597 100644
--- a/vendor/google.golang.org/grpc/stream.go
+++ b/vendor/google.golang.org/grpc/stream.go
@@ -123,6 +123,9 @@ type ClientStream interface {
// calling RecvMsg on the same stream at the same time, but it is not safe
// to call SendMsg on the same stream in different goroutines. It is also
// not safe to call CloseSend concurrently with SendMsg.
+ //
+ // It is not safe to modify the message after calling SendMsg. Tracing
+ // libraries and stats handlers may use the message lazily.
SendMsg(m interface{}) error
// RecvMsg blocks until it receives a message into m or the stream is
// done. It returns io.EOF when the stream completes successfully. On
@@ -152,6 +155,11 @@ type ClientStream interface {
// If none of the above happen, a goroutine and a context will be leaked, and grpc
// will not call the optionally-configured stats handler with a stats.End message.
func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
+ if err := cc.idlenessMgr.onCallBegin(); err != nil {
+ return nil, err
+ }
+ defer cc.idlenessMgr.onCallEnd()
+
// allow interceptor to see all applicable call options, which means those
// configured as defaults from dial option as well as per-call options
opts = combine(cc.dopts.callOptions, opts)
@@ -469,7 +477,7 @@ func (a *csAttempt) newStream() error {
// It is safe to overwrite the csAttempt's context here, since all state
// maintained in it are local to the attempt. When the attempt has to be
// retried, a new instance of csAttempt will be created.
- if a.pickResult.Metatada != nil {
+ if a.pickResult.Metadata != nil {
// We currently do not have a function it the metadata package which
// merges given metadata with existing metadata in a context. Existing
// function `AppendToOutgoingContext()` takes a variadic argument of key
@@ -479,7 +487,7 @@ func (a *csAttempt) newStream() error {
// in a form passable to AppendToOutgoingContext(), or create a version
// of AppendToOutgoingContext() that accepts a metadata.MD.
md, _ := metadata.FromOutgoingContext(a.ctx)
- md = metadata.Join(md, a.pickResult.Metatada)
+ md = metadata.Join(md, a.pickResult.Metadata)
a.ctx = metadata.NewOutgoingContext(a.ctx, md)
}
@@ -499,7 +507,7 @@ func (a *csAttempt) newStream() error {
return toRPCErr(nse.Err)
}
a.s = s
- a.p = &parser{r: s}
+ a.p = &parser{r: s, recvBufferPool: a.cs.cc.dopts.recvBufferPool}
return nil
}
@@ -1262,17 +1270,22 @@ func newNonRetryClientStream(ctx context.Context, desc *StreamDesc, method strin
return nil, err
}
as.s = s
- as.p = &parser{r: s}
+ as.p = &parser{r: s, recvBufferPool: ac.dopts.recvBufferPool}
ac.incrCallsStarted()
if desc != unaryStreamDesc {
- // Listen on cc and stream contexts to cleanup when the user closes the
- // ClientConn or cancels the stream context. In all other cases, an error
- // should already be injected into the recv buffer by the transport, which
- // the client will eventually receive, and then we will cancel the stream's
- // context in clientStream.finish.
+ // Listen on stream context to cleanup when the stream context is
+ // canceled. Also listen for the addrConn's context in case the
+ // addrConn is closed or reconnects to a different address. In all
+ // other cases, an error should already be injected into the recv
+ // buffer by the transport, which the client will eventually receive,
+ // and then we will cancel the stream's context in
+ // addrConnStream.finish.
go func() {
+ ac.mu.Lock()
+ acCtx := ac.ctx
+ ac.mu.Unlock()
select {
- case <-ac.ctx.Done():
+ case <-acCtx.Done():
as.finish(status.Error(codes.Canceled, "grpc: the SubConn is closing"))
case <-ctx.Done():
as.finish(toRPCErr(ctx.Err()))