transport

Documentation

Overview

Package transport is an interface for synchronous connection based communication

Index

• Variables
• func NewHTTPTransport(opts ...Option) *httpTransport
• type Client
• type Config
  • func NewConfig() *Config
  • func (d *Config) Merge(src *Config) error
• type ConfigStore
  • func NewConfigStore() ConfigStore
  • func (d *ConfigStore) Merge(src *ConfigStore) error
• type DiConfig
  • func ProvideConfig(_ di.DiConfig, flags *DiFlags, config *Config, cliConfig *cli.Config, ...) (DiConfig, error)
  • func ProvideConfigNoFlags(config *Config, configor config.Config) (DiConfig, error)
• type DiFlags
  • func ProvideFlags(config *Config, cliConfig *cli.Config, c cli.Cli) (*DiFlags, error)
• type DialOption
  • func WithStream() DialOption
  • func WithTimeout(d time.Duration) DialOption
• type DialOptions
• type ListenOption
  • func NetListener(customListener net.Listener) ListenOption
• type ListenOptions
• type Listener
• type Message
• type Option
  • func Addrs(addrs ...string) Option
  • func Codec(c codec.Marshaler) Option
  • func Secure(b bool) Option
  • func TLSConfig(t *tls.Config) Option
  • func Timeout(t time.Duration) Option
  • func WithLogger(l logger.Logger) Option
• type Options
• type Socket
• type Transport
  • func NewMemoryTransport(opts ...Option) Transport
  • func Provide(_ DiConfig, config *Config) (Transport, error)
• type TransportConfig

Variables

var DiNoCliSet = wire.NewSet(ProvideConfigNoFlags, Provide)

var DiSet = wire.NewSet(ProvideFlags, ProvideConfig, Provide)

var Plugins = generic.NewContainer(func(opts ...Option) Transport { return nil })

Functions

func NewHTTPTransport

func NewHTTPTransport(opts ...Option) *httpTransport

Types

type Client

type Client interface {
	Socket
}

type Config

type Config struct {
	Transport TransportConfig `json:"broker" yaml:"Transport"`
}

func NewConfig

func NewConfig() *Config

func (*Config) Merge

func (d *Config) Merge(src *Config) error

type ConfigStore

type ConfigStore struct {
	Enabled   bool     `json:"enabled" yaml:"Enabled"`
	Plugin    string   `json:"plugin,omitempty" yaml:"Plugin,omitempty"`
	Addresses []string `json:"addresses,omitempty" yaml:"Addresses,omitempty"`
}

func NewConfigStore

func NewConfigStore() ConfigStore

func (*ConfigStore) Merge

func (d *ConfigStore) Merge(src *ConfigStore) error

type DiConfig

type DiConfig struct{}

DiConfig is marker that DiFlags has been parsed into Config

func ProvideConfig

func ProvideConfig(
	_ di.DiConfig,
	flags *DiFlags,
	config *Config,
	cliConfig *cli.Config,
	configor config.Config,
) (DiConfig, error)

func ProvideConfigNoFlags

func ProvideConfigNoFlags(
	config *Config,
	configor config.Config,
) (DiConfig, error)

type DiFlags

type DiFlags struct {
	Plugin    string
	Addresses string
}

func ProvideFlags

func ProvideFlags(
	config *Config,
	cliConfig *cli.Config,
	c cli.Cli,
) (*DiFlags, error)

type DialOption

type DialOption func(*DialOptions)

func WithStream

func WithStream() DialOption

Indicates whether this is a streaming connection.

func WithTimeout

func WithTimeout(d time.Duration) DialOption

type DialOptions

type DialOptions struct {
	// Tells the transport this is a streaming connection with
	// multiple calls to send/recv and that send may not even be called
	Stream bool
	// Timeout for dialing
	Timeout time.Duration

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type ListenOption

type ListenOption func(*ListenOptions)

func NetListener

func NetListener(customListener net.Listener) ListenOption

NetListener Set net.Listener for httpTransport.

type ListenOptions

type ListenOptions struct {

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type Listener

type Listener interface {
	Addr() string
	Close() error
	Accept(func(Socket)) error
}

type Message

type Message struct {
	Header map[string]string
	Body   []byte
}

Message is a broker message.

type Option

type Option func(*Options)

func Addrs

func Addrs(addrs ...string) Option

Addrs to use for transport.

func Codec

func Codec(c codec.Marshaler) Option

Codec sets the codec used for encoding where the transport does not support message headers.

func Secure

func Secure(b bool) Option

Use secure communication. If TLSConfig is not specified we use InsecureSkipVerify and generate a self signed cert.

func TLSConfig

func TLSConfig(t *tls.Config) Option

TLSConfig to be used for the transport.

func Timeout

func Timeout(t time.Duration) Option

Timeout sets the timeout for Send/Recv execution.

func WithLogger

func WithLogger(l logger.Logger) Option

WithLogger sets the underline logger.

type Options

type Options struct {
	// Addrs is the list of intermediary addresses to connect to
	Addrs []string
	// Codec is the codec interface to use where headers are not supported
	// by the transport and the entire payload must be encoded
	Codec codec.Marshaler
	// Secure tells the transport to secure the connection.
	// In the case TLSConfig is not specified best effort self-signed
	// certs should be used
	Secure bool
	// TLSConfig to secure the connection. The assumption is that this
	// is mTLS keypair
	TLSConfig *tls.Config
	// Timeout sets the timeout for Send/Recv
	Timeout time.Duration
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
	// Logger is the underline logger
	Logger logger.Logger
}

type Socket

type Socket interface {
	Recv(*Message) error
	Send(*Message) error
	Close() error
	Local() string
	Remote() string
}

type Transport

type Transport interface {
	Init(...Option) error
	Options() Options
	Dial(addr string, opts ...DialOption) (Client, error)
	Listen(addr string, opts ...ListenOption) (Listener, error)
	String() string
}

Transport is an interface which is used for communication between services. It uses connection based socket send/recv semantics and has various implementations; http, grpc, quic.

var (
	DefaultTransport Transport = NewHTTPTransport()

	DefaultDialTimeout = time.Second * 5
)

func NewMemoryTransport

func NewMemoryTransport(opts ...Option) Transport

func Provide

func Provide(

	_ DiConfig,

	config *Config,
) (Transport, error)

type TransportConfig

type TransportConfig struct {
	Enabled   bool     `json:"enabled" yaml:"Enabled"`
	Plugin    string   `json:"plugin,omitempty" yaml:"Plugin,omitempty"`
	Addresses []string `json:"addresses,omitempty" yaml:"Addresses,omitempty"`
}