func RetryStreamClientInterceptor(optFuncs ...CallOption) grpc.StreamClientInterceptor {
	intOpts := reuseOrNewWithCallOptions(defaultRetryOptions, optFuncs)
	return func(parentCtx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		grpcOpts, retryOpts := filterCallOptions(opts)
		callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
		// short circuit for simplicity, and avoiding allocations.
		if callOpts.max == 0 {
			return streamer(parentCtx, desc, cc, method, grpcOpts...)
		}
		if desc.ClientStreams {
			return nil, status.Errorf(codes.Unimplemented, "grpc_retry: cannot retry on ClientStreams, set grpc_retry.Disable()")
		}

		var lastErr error
		for attempt := uint(0); attempt < callOpts.max; attempt++ {
			if err := waitRetryBackoff(parentCtx, attempt, callOpts); err != nil {
				return nil, err
			}
			callCtx, cancel := perCallContext(parentCtx, callOpts, 0)

			var newStreamer grpc.ClientStream
			func() {
				defer cancel()
				newStreamer, lastErr = streamer(callCtx, desc, cc, method, grpcOpts...)
			}()
			if lastErr == nil {
				retryingStreamer := &serverStreamingRetryingStream{
					ClientStream: newStreamer,
					callOpts:     callOpts,
					parentCtx:    parentCtx,
					streamerCall: func(ctx context.Context) (grpc.ClientStream, error) {
						return streamer(ctx, desc, cc, method, grpcOpts...)
					},
				}
				return retryingStreamer, nil
			}

			if isContextError(lastErr) {
				if parentCtx.Err() != nil {
					// its the parent context deadline or cancellation.
					return nil, lastErr
				} else if callOpts.perCallTimeout != 0 {
					// We have set a perCallTimeout in the retry middleware, which would result in a context error if
					// the deadline was exceeded, in which case try again.
					continue
				}
			}
			if !isRetriable(lastErr, callOpts) {
				return nil, lastErr
			}
		}
		return nil, lastErr
	}
}
// RetryStreamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls.
//
// The default configuration of the interceptor is to not retry *at all*. This behavior can be
// changed through options (e.g. WithMax) on creation of the interceptor or on call (through grpc.CallOptions).
//
// Retry logic is available *only for ServerStreams*, i.e. 1:n streams, as the internal logic needs
// to buffer the messages sent by the client. If retry is enabled on any other streams (ClientStreams,
// BidiStreams), the retry interceptor will fail the call.
RetryStreamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls.

The default configuration of the interceptor is to not retry *at all*. This behavior can be
changed through options  on creation of the interceptor or on call (through grpc.CallOptions).

Retry logic is available *only for ServerStreams*, i.e. 1:n streams, as the internal logic needs
to buffer the messages sent by the client. If retry is enabled on any other streams (ClientStreams,
BidiStreams), the retry interceptor will fail the call.
// RetryStreamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls . // // The default configuration of the interceptor is to not retry * at all * . This behavior can be // changed through options ( e . g . WithMax ) on creation of the interceptor or on call ( through grpc . CallOptions ) . // // Retry logic is available * only for ServerStreams * i . e . 1 : n streams as the internal logic needs // to buffer the messages sent by the client . If retry is enabled on any other streams ( ClientStreams // BidiStreams ) the retry interceptor will fail the call .
func RetryStreamClientInterceptor(optFuncs ...CallOption) grpc.StreamClientInterceptor {
	intOpts := reuseOrNewWithCallOptions(defaultRetryOptions, optFuncs)
	return func(parentCtx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		grpcOpts, retryOpts := filterCallOptions(opts)
		callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
		if callOpts.max == 0 {
			return streamer(parentCtx, desc, cc, method, grpcOpts...)
		}
		if desc.ClientStreams {
			return nil, status.Errorf(codes.Unimplemented, "grpc_retry: cannot retry on ClientStreams, set grpc_retry.Disable()")
		}
		var lastErr error
		for attempt := uint(0); attempt < callOpts.max; attempt++ {
			if err := waitRetryBackoff(parentCtx, attempt, callOpts); err != nil {
				return nil, err
			}
			callCtx, cancel := perCallContext(parentCtx, callOpts, 0)
			var newStreamer grpc.ClientStream
			func() {
				defer cancel()
				newStreamer, lastErr = streamer(callCtx, desc, cc, method, grpcOpts...)
			}()
			if lastErr == nil {
				retryingStreamer := &serverStreamingRetryingStream{
					ClientStream: newStreamer,
					callOpts:     callOpts,
					parentCtx:    parentCtx,
					streamerCall: func(ctx context.Context) (grpc.ClientStream, error) {
						return streamer(ctx, desc, cc, method, grpcOpts...)
					},
				}
				return retryingStreamer, nil
			}
			if isContextError(lastErr) {
				if parentCtx.Err() != nil {
					return nil, lastErr
				} else if callOpts.perCallTimeout != 0 {
					continue
				}
			}
			if !isRetriable(lastErr, callOpts) {
				return nil, lastErr
			}
		}
		return nil, lastErr
	}
}
func RetryStreamClientInterceptor ( optFuncs ... CallOption ) grpc . StreamClientInterceptor { intOpts := reuseOrNewWithCallOptions ( defaultRetryOptions , optFuncs ) return func ( parentCtx context . Context , desc * grpc . StreamDesc , cc * grpc . ClientConn , method string , streamer grpc . Streamer , opts ... grpc . CallOption ) ( grpc . ClientStream , error ) { grpcOpts , retryOpts := filterCallOptions ( opts ) callOpts := reuseOrNewWithCallOptions ( intOpts , retryOpts ) if callOpts . max == 0 { return streamer ( parentCtx , desc , cc , method , grpcOpts ... ) } if desc . ClientStreams { return nil , status . Errorf ( codes . Unimplemented , " " ) } var lastErr error for attempt := uint ( 0 ) ; attempt < callOpts . max ; attempt ++ { if err := waitRetryBackoff ( parentCtx , attempt , callOpts ) ; err != nil { return nil , err } callCtx , cancel := perCallContext ( parentCtx , callOpts , 0 ) var newStreamer grpc . ClientStream func ( ) { defer cancel ( ) newStreamer , lastErr = streamer ( callCtx , desc , cc , method , grpcOpts ... ) } ( ) if lastErr == nil { retryingStreamer := & serverStreamingRetryingStream { ClientStream : newStreamer , callOpts : callOpts , parentCtx : parentCtx , streamerCall : func ( ctx context . Context ) ( grpc . ClientStream , error ) { return streamer ( ctx , desc , cc , method , grpcOpts ... ) } , } return retryingStreamer , nil } if isContextError ( lastErr ) { if parentCtx . Err ( ) != nil { return nil , lastErr } else if callOpts . perCallTimeout != 0 { continue } } if ! isRetriable ( lastErr , callOpts ) { return nil , lastErr } } return nil , lastErr } }
RetryStreamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls.
RetryStreamClientInterceptor returns a new retrying stream client interceptor for server side streaming calls .
// short circuit for simplicity, and avoiding allocations. // its the parent context deadline or cancellation. // We have set a perCallTimeout in the retry middleware, which would result in a context error if // the deadline was exceeded, in which case try again.