session

Documentation

Overview

session implements amqp connection and session management though the concept of a Container, inspired by the Conatiner interface exposed by the qpid proton amqp libraries. This abstraction allows vanflow components to be written without repeating connection management tasks.

Index

• type Container
  • func NewContainer(address string, config ContainerConfig) Container
  • func NewMockContainer(router *MockRouter) Container
• type ContainerConfig
• type ContainerFactory
  • func NewContainerFactory(address string, cfg ContainerConfig) ContainerFactory
  • func NewMockContainerFactory() ContainerFactory
• type MockRouter
  • func NewMockRouter() *MockRouter
• type Receiver
• type ReceiverOptions
• type RetryableError
• type SASLType
• type Sender
• type SenderOptions

Types

type Container

type Container interface {
	// Start the container
	Start(context.Context)
	// NewReceiver adds a new reciver link using the container's session
	NewReceiver(address string, opts ReceiverOptions) Receiver
	// NewSender adds a new sender link using the container's session
	NewSender(address string, opts SenderOptions) Sender
	// OnSessionError
	OnSessionError(func(err error))
}

func NewContainer

func NewContainer(address string, config ContainerConfig) Container

NewContainer creats an amqp container that will attempt to create a single connection + session pair using the supplied amqp connection options for use with the container's Senders and Receivers. Will recreate the connection and session when a link encounters an error using the specified backoff strategy.

func NewMockContainer

func NewMockContainer(router *MockRouter) Container

NewMockContainer creates a mock container using the mock router

type ContainerConfig

type ContainerConfig struct {
	ContainerID  string
	MaxFrameSize uint32
	TLSConfig    *tls.Config
	SASLType     SASLType
	// BackOff strategy to use when reestablishing a connection defaults to an
	// exponential backoff capped at 30 second intervals with no set retry
	// limit.
	BackOff backoff.BackOff
}

type ContainerFactory

type ContainerFactory interface {
	Create() Container
}

ContainerFactory creates Containers with a given configuration

func NewContainerFactory

func NewContainerFactory(address string, cfg ContainerConfig) ContainerFactory

func NewMockContainerFactory

func NewMockContainerFactory() ContainerFactory

type MockRouter

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

func NewMockRouter

func NewMockRouter() *MockRouter

NewMockRouter returns a mock router that can be used by containers to send and receive multicast messages. Behaves like a very rough approximation of the skupper router for routing vanflow messages.

type Receiver

type Receiver interface {
	Next(context.Context) (*amqp.Message, error)
	Accept(context.Context, *amqp.Message) error
	Close(context.Context) error
}

type ReceiverOptions

type ReceiverOptions struct {
	Credit int
}

type RetryableError

type RetryableError interface {
	Retry() time.Duration
}

type SASLType

type SASLType string

const (
	SASLTypeExternal SASLType = "EXTERNAL"
)

type Sender

type Sender interface {
	Send(context.Context, *amqp.Message) error
	Close(context.Context) error
}

type SenderOptions

type SenderOptions struct {
}