transport

Documentation

Overview

Package transport implements the low level concerns of sending and receiving bytes.

Index

• func CanonicalizeHeaderKey(k string) string
• func IsBadRequestError(err error) bool
• func IsTimeoutError(err error) bool
• func IsUnexpectedError(err error) bool
• type Channel
  • func IdentityChannel(caller, service string, out Outbound) Channel
• type Deps
  • func (d Deps) Tracer() opentracing.Tracer
  • func (d Deps) WithTracer(t opentracing.Tracer) Deps
• type Encoding
• type Filter
• type FilterFunc
  • func (f FilterFunc) Call(ctx context.Context, request *Request, out Outbound) (*Response, error)
• type Handler
  • func ApplyInterceptor(h Handler, i Interceptor) Handler
• type Headers
  • func HeadersFromMap(m map[string]string) Headers
  • func NewHeaders() Headers
  • func NewHeadersWithCapacity(capacity int) Headers
  • func (h Headers) Del(k string)
  • func (h Headers) Get(k string) (string, bool)
  • func (h Headers) Items() map[string]string
  • func (h Headers) Len() int
  • func (h Headers) With(k, v string) Headers
• type Inbound
• type Interceptor
• type InterceptorFunc
  • func (f InterceptorFunc) Handle(ctx context.Context, opts Options, req *Request, resw ResponseWriter, ...) error
• type MapRegistry
  • func NewMapRegistry(defaultService string) MapRegistry
  • func (m MapRegistry) GetHandler(service, procedure string) (Handler, error)
  • func (m MapRegistry) Register(service, procedure string, handler Handler)
• type Options
  • func (o Options) Get(k interface{}) (interface{}, bool)
  • func (o Options) Merge(others ...Options) Options
  • func (o Options) With(key, val interface{}) Options
• type Outbound
  • func ApplyFilter(o Outbound, f Filter) Outbound
• type Outbounds
• type Registry
• type Request
• type Response
• type ResponseWriter

Functions

func CanonicalizeHeaderKey

func CanonicalizeHeaderKey(k string) string

CanonicalizeHeaderKey canonicalizes the given header key for storage into the Headers map.

func IsBadRequestError

func IsBadRequestError(err error) bool

IsBadRequestError returns true if the request could not be processed because it was invalid.

func IsTimeoutError

func IsTimeoutError(err error) bool

IsTimeoutError return true if the given error is a TimeoutError.

func IsUnexpectedError

func IsUnexpectedError(err error) bool

IsUnexpectedError returns true if the server failed to process the request because of an unhandled error.

Types

type Channel

type Channel interface {
	// Name of the service making the request.
	Caller() string

	// Name of the service to which the request is being made.
	Service() string

	// Returns an outbound to send the request through.
	//
	// MAY be called multiple times for a request. MAY return different outbounds
	// for each call. The returned outbound MUST have already been started.
	GetOutbound() Outbound
}

Channel scopes outbounds to a single caller-service pair.

func IdentityChannel

func IdentityChannel(caller, service string, out Outbound) Channel

IdentityChannel constructs a simple Channel for the given caller-service pair which always returns the given Outbound.

type Deps

type Deps struct {
	// contains filtered or unexported fields
}

Deps is a struct shared by all inbounds and outbounds in the context of a dispatcher. The dispatcher starts every transport with these dependencies. A zero Deps struct is suitable for testing and provides noop implementations of all dependencies.

var NoDeps Deps

NoDeps is a singleton zero Deps instance.

func (Deps) Tracer

func (d Deps) Tracer() opentracing.Tracer

Tracer provides the opentracing Tracer instance needed by transports.

func (Deps) WithTracer

func (d Deps) WithTracer(t opentracing.Tracer) Deps

WithTracer returns a variant of these dependencies with a given opentracing Tracer.

type Encoding

type Encoding string

Encoding represents an encoding format for requests.

type Filter

type Filter interface {
	Call(ctx context.Context, request *Request, out Outbound) (*Response, error)
}

Filter defines transport-level middleware for Outbounds.

Filters MAY

- change the context - change the request - change the returned response - handle the returned error - call the given outbound zero or more times

Filters MUST

- always return a non-nil Response or error. - be thread-safe

Filters are re-used across requests and MAY be called multiple times on the same request.

var NopFilter Filter = nopFilter{}

NopFilter is a filter that does not do anything special. It simply calls the underlying Outbound.

type FilterFunc

type FilterFunc func(context.Context, *Request, Outbound) (*Response, error)

FilterFunc adapts a function into a Filter.

func (FilterFunc) Call

func (f FilterFunc) Call(ctx context.Context, request *Request, out Outbound) (*Response, error)

Call for FilterFunc.

type Handler

type Handler interface {
	// Handle the given request, writing the response to the given
	// ResponseWriter.
	//
	// An error may be returned in case of failures. BadRequestError must be
	// returned for invalid requests. All other failures are treated as
	// UnexpectedErrors.
	Handle(
		ctx context.Context,
		opts Options,
		req *Request,
		resw ResponseWriter,
	) error
}

Handler handles a single transport-level request.

func ApplyInterceptor

func ApplyInterceptor(h Handler, i Interceptor) Handler

ApplyInterceptor applies the given Interceptor to the given Handler.

type Headers

type Headers internal.Headers

Headers is the transport-level representation of application headers.

Keys in the map MUST be canonicalized with CanonicalizeHeaderKey.

You probably want to look at yarpc.Headers instead.

func HeadersFromMap

func HeadersFromMap(m map[string]string) Headers

HeadersFromMap builds a new Headers object from the given map of header key-value pairs.

func NewHeaders

func NewHeaders() Headers

NewHeaders builds a new Headers object.

func NewHeadersWithCapacity

func NewHeadersWithCapacity(capacity int) Headers

NewHeadersWithCapacity builds a new Headers object with the given capacity.

func (Headers) Del

func (h Headers) Del(k string)

Del deletes the header with the given name from the Headers map.

This is a no-op if the key does not exist.

func (Headers) Get

func (h Headers) Get(k string) (string, bool)

Get retrieves the value associated with the given header name.

func (Headers) Items

func (h Headers) Items() map[string]string

Items returns the underlying map for this Headers map.

Keys in the map are normalized using CanonicalizeHeaderKey.

The returned map MUST NOT be mutated.

func (Headers) Len

func (h Headers) Len() int

Len returns the number of headers defined on this object.

func (Headers) With

func (h Headers) With(k, v string) Headers

With returns a Headers object with the given key-value pair added to it. The returned object MAY not point to the same Headers underlying data store as the original Headers so the returned Headers MUST always be used instead of the original object.

headers = headers.With("foo", "bar").With("baz", "qux")

type Inbound

type Inbound interface {
	// Starts accepting new requests and dispatches them to the given Handler.
	//
	// The function MUST return immediately, although it SHOULD block until
	// the inbound is ready to start accepting new requests.
	//
	// Implementations can assume that this function is called at most once.
	Start(handler Handler, deps Deps) error

	// Stops the inbound. No new requests will be processed.
	//
	// This MAY block while the server drains ongoing requests.
	Stop() error
}

Inbound is a transport that knows how to receive requests for procedure calls.

type Interceptor

type Interceptor interface {
	Handle(
		ctx context.Context,
		opts Options,
		req *Request,
		resw ResponseWriter,
		h Handler,
	) error
}

Interceptor defines a transport-level middleware for Inbounds.

Interceptors MAY

- change the context - change the request - call the ResponseWriter - modify the response body by wrapping the ResponseWriter - handle the returned error - call the given handler zero or more times

Interceptors MUST be thread-safe.

Interceptors are re-used across requests and MAY be called multiple times for the same request.

var NopInterceptor Interceptor = nopInterceptor{}

NopInterceptor is a interceptor that does not do anything special. It simply calls the underlying Handler.

type InterceptorFunc

type InterceptorFunc func(context.Context, Options, *Request, ResponseWriter, Handler) error

InterceptorFunc adapts a function into an Interceptor.

func (InterceptorFunc) Handle

func (f InterceptorFunc) Handle(ctx context.Context, opts Options, req *Request, resw ResponseWriter, h Handler) error

Handle for InterceptorFunc

type MapRegistry

type MapRegistry struct {
	// contains filtered or unexported fields
}

MapRegistry is a Registry that maintains a map of the registered procedures.

func NewMapRegistry

func NewMapRegistry(defaultService string) MapRegistry

NewMapRegistry builds a new MapRegistry that uses the given name as the default service name.

func (MapRegistry) GetHandler

func (m MapRegistry) GetHandler(service, procedure string) (Handler, error)

GetHandler retrieves the Handler for the given Procedure or returns an error.

func (MapRegistry) Register

func (m MapRegistry) Register(service, procedure string, handler Handler)

Register registers the procedure with the MapRegistry.

type Options

type Options struct {
	// contains filtered or unexported fields
}

Options act as an extension point for transports to configure behavior of other parts of the system.

A component that that wishes to be customizable based on transport.Options should declare a private type and key values off that.

package foo

type bar struct{}

func OptionFoo(v string) (o transport.Options) {
	return o.With(bar{}, v)
}

A transport that wishes to change behavior simply needs to provide an Options object, merging OptionFoo into it.

func (myOutbound) Options() (opts transport.Options) {
	return opts.Merge(foo.OptionFoo("hello"), bar.OptionBar(false))
}

Now the implementation of foo can use Options.Get to act differently based on the outbound's options.

func (Options) Get

func (o Options) Get(k interface{}) (interface{}, bool)

Get returns the value associated with the given key.

func (Options) Merge

func (o Options) Merge(others ...Options) Options

Merge returns a copy of an Options object with items from all the given Options merged into it.

Values in the rightmost Options object take precedence in case of conflicts.

func (Options) With

func (o Options) With(key, val interface{}) Options

With returns a copy of this Options object with the given key-value pair added to it.

The key should be a custom type to avoid conflicts with options of other components.

opts = opts.With(foo{}, bar)
opts = opts.With(baz{}, qux)

type Outbound

type Outbound interface {
	// Sets up the outbound to start making calls.
	//
	// This MUST block until the outbound is ready to start sending requests.
	//
	// Implementations can assume that this function is called at most once.
	Start(deps Deps) error

	// Stops the outbound, cleaning up any resources held by the Outbound.
	Stop() error

	// Options for all requests made through this Outbound.
	Options() Options

	// Call sends the given request through this transport and returns its
	// response.
	//
	// This MUST NOT be called before Start() has been called successfully. This
	// MAY panic if called without calling Start(). This MUST be safe to call
	// concurrently.
	Call(ctx context.Context, request *Request) (*Response, error)
}

Outbound is a transport that knows how to send requests for procedure calls.

func ApplyFilter

func ApplyFilter(o Outbound, f Filter) Outbound

ApplyFilter applies the given Filter to the given Outbound.

type Outbounds

type Outbounds map[string]Outbound

Outbounds is a map of service name to Outbound for that service.

type Registry

type Registry interface {
	// Registers a procedure with this registry under the given service name.
	//
	// service may be empty to indicate that the default service name should
	// be used.
	Register(service, procedure string, handler Handler)

	// Gets the handler for the given service, procedure tuple. An
	// UnrecognizedProcedureError will be returned if the handler does not
	// exist.
	//
	// service may be empty to indicate that the default service name should
	// be used.
	GetHandler(service, procedure string) (Handler, error)
}

Registry maintains and provides access to a collection of procedures and their handlers.

type Request

type Request struct {
	// Name of the service making the request.
	Caller string

	// Name of the service to which the request is being made.
	// The service refers to the canonical traffic group for the service.
	Service string

	// Name of the encoding used for the request body.
	Encoding Encoding

	// Name of the procedure being called.
	Procedure string

	// Headers for the request.
	Headers Headers

	// ShardKey is an opaque string that is meaningful to the destined service
	// for how to relay a request within a cluster to the shard that owns the
	// key.
	ShardKey string

	// RoutingKey refers to a traffic group for the destined service, and when
	// present may override the service name for purposes of routing.
	RoutingKey string

	// RoutingDelegate refers to the traffic group for a service that proxies
	// for the destined service for routing purposes. The routing delegate may
	// override the routing key and service.
	RoutingDelegate string

	// Request payload.
	Body io.Reader
}

Request is the low level request representation.

type Response

type Response struct {
	Headers Headers
	Body    io.ReadCloser
}

Response is the low level response representation.

type ResponseWriter

type ResponseWriter interface {
	io.Writer

	// AddHeaders adds the given headers to the response. If called, this MUST
	// be called before any invocation of Write().
	//
	// This MUST NOT panic if Headers is nil.
	AddHeaders(Headers)

	// SetApplicationError specifies that this response contains an
	// application error. If called, this MUST be called before any invocation
	// of Write().
	SetApplicationError()
}

ResponseWriter allows Handlers to write responses in a streaming fashion.