streaming

Documentation

Overview

Package streaming interface

Index

• Constants
• func CallWithTimeout(timeout time.Duration, cancel context.CancelFunc, f func() (err error)) error
• func FinishStream(s Stream, err error)
• func NewCtxWithStream(ctx context.Context, stream Stream) context.Context
• func UnaryCompatibleMiddleware(mode serviceinfo.StreamingMode, allow bool) bool
• type Args
• type Result
• type Stream
  • func GetStream(ctx context.Context) Stream
• type WithDoFinish

Constants

const KitexUnusedProtection = 0

KitexUnusedProtection may be anonymously referenced in another package to avoid build error

Functions

func CallWithTimeout

func CallWithTimeout(timeout time.Duration, cancel context.CancelFunc, f func() (err error)) error

CallWithTimeout executes a function with timeout. If timeout is 0, the function will be executed without timeout; panic is not recovered in this case; If time runs out, it will return a kerrors.ErrRPCTimeout; If your function panics, it will also return a kerrors.ErrRPCTimeout with the panic details; Other kinds of errors are always returned by your function.

NOTE: the `cancel` function is necessary to cancel the underlying transport, otherwise the recv/send call will block until the transport is closed, which may cause surge of goroutines.

func FinishStream

func FinishStream(s Stream, err error)

FinishStream records the end of stream you can call it manually when all business logic is done, and you don't want to call Recv/Send for the io.EOF (which triggers the DoFinish automatically). Note: if you're to wrap the original stream in a Client middleware, you should also implement WithDoFinish in your Stream implementation.

func NewCtxWithStream

func NewCtxWithStream(ctx context.Context, stream Stream) context.Context

func UnaryCompatibleMiddleware

func UnaryCompatibleMiddleware(mode serviceinfo.StreamingMode, allow bool) bool

UnaryCompatibleMiddleware returns whether to use compatible middleware for unary.

Types

type Args

type Args struct {
	Stream Stream
}

Args endpoint request

type Result

type Result struct {
	Stream Stream
}

Result endpoint response

type Stream

type Stream interface {
	// SetHeader sets the header metadata. It may be called multiple times.
	// When call multiple times, all the provided metadata will be merged.
	// All the metadata will be sent out when one of the following happens:
	//  - ServerStream.SendHeader() is called;
	//  - The first response is sent out;
	//  - An RPC status is sent out (error or success).
	SetHeader(metadata.MD) error
	// SendHeader sends the header metadata.
	// The provided md and headers set by SetHeader() will be sent.
	// It fails if called multiple times.
	SendHeader(metadata.MD) error
	// SetTrailer sets the trailer metadata which will be sent with the RPC status.
	// When called more than once, all the provided metadata will be merged.
	SetTrailer(metadata.MD)
	// Header is used for client side stream to receive header from server.
	Header() (metadata.MD, error)
	// Trailer is used for client side stream to receive trailer from server.
	Trailer() metadata.MD
	// Context the stream context.Context
	Context() context.Context
	// RecvMsg recvive message from peer
	// will block until an error or a message received
	// not concurrent-safety
	RecvMsg(m interface{}) error
	// SendMsg send message to peer
	// will block until an error or enough buffer to send
	// not concurrent-safety
	SendMsg(m interface{}) error
	// not concurrent-safety with SendMsg
	io.Closer
}

Stream both client and server stream

func GetStream

func GetStream(ctx context.Context) Stream

type WithDoFinish

type WithDoFinish interface {
	DoFinish(error)
}

WithDoFinish should be implemented when: (1) you want to wrap a stream in client middleware, and (2) you want to manually call streaming.FinishStream(stream, error) to record the end of stream Note: the DoFinish should be reentrant, better with a sync.Once.