upstreambackup

Documentation

Index

• Variables
• type DefaultForwarder
  • func NewDefaultForwarder(name string, outs []string, cfg *DefaultForwarderConfig, l *util.Logger) (*DefaultForwarder, error)
  • func (f *DefaultForwarder) AckMessages() <-chan UpstreamAck
  • func (f *DefaultForwarder) ChangeOut(oldOut, newOut string) error
  • func (f *DefaultForwarder) Forward(inputID uint16, inputMsgID uint32, data []byte) error
  • func (f *DefaultForwarder) GetOldestOutput() (uint32, error)
  • func (f *DefaultForwarder) Run(ctx context.Context) error
• type DefaultForwarderConfig
• type DefaultReceiver
  • func NewDefaultReceiver(addr string, inNames []string, l *util.Logger) *DefaultReceiver
  • func (r *DefaultReceiver) Acks() chan<- UpstreamAck
  • func (r *DefaultReceiver) Messages() <-chan *UpstreamMessage
  • func (r *DefaultReceiver) Run(ctx context.Context) error
• type DownstreamForwarder
  • func NewDownstreamForwarder(downstreamIndex uint16, name string, addr string, iter *LogBufferIterator, ...) *DownstreamForwarder
  • func (f *DownstreamForwarder) Run(ctx context.Context) error
• type ForwardLog
  • func NewForwardLog(forwardLogDir string) (*ForwardLog, error)
  • func (l *ForwardLog) Close() error
  • func (l *ForwardLog) GetOldestOutput() (uint32, error)
  • func (l *ForwardLog) NewIterator() *LogBufferIterator
  • func (l *ForwardLog) Trim(idBorder uint32) (map[uint16]uint32, error)
  • func (l *ForwardLog) Write(inputID uint16, inputMsgID, outputMsgID uint32, data []byte) error
• type LogBufferIterator
  • func NewLogBufferIterator(logBuffer *logBuffer) *LogBufferIterator
  • func (i *LogBufferIterator) Next(ctx context.Context, item *forwardLogItem) error
• type UpstreamAck
  • func (a UpstreamAck) String() string
• type UpstreamMessage
• type UpstreamReceiver
  • func NewUpstreamReceiver(upstreamIndex uint16, name string, tcpConn *connutil.Connection, ...) *UpstreamReceiver
  • func (r *UpstreamReceiver) Ack(ack uint32) error
  • func (r *UpstreamReceiver) Run(ctx context.Context) error

Variables

var DummyUpstreamMessage = &UpstreamMessage{
	dataMessage: &dataMessage{},
}

DummyUpstreamMessage пустое сообщение из upstream, используется для поддержки источников.

var (
	ErrConnectionClosed = errors.New("can not write to closed connection")
)

Список возможных ошибок

var (
	// ErrNoEnoughtData возвращается, когда длина полученного сообщения меньше,
	// чем переданная в заголовке.
	ErrNoEnoughtData = errors.New("too small data")
)

var (
	ErrUnknownOutAddress = errors.New("unknown out address")
)

Возможные ошибки.

var (
	ErrUpstreamUnknown = errors.New("upstreams is unknown")
)

Возможные ошибки работы.

Types

type DefaultForwarder

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

DefaultForwarder предает сообщения дальше по потоку, при этом обеспечивая отказоустойчивость по схеме upstream_backup.

func NewDefaultForwarder

func NewDefaultForwarder(name string, outs []string, cfg *DefaultForwarderConfig, l *util.Logger) (*DefaultForwarder, error)

NewDefaultForwarder создает новый объект DefaultForwarder.

func (*DefaultForwarder) AckMessages

func (f *DefaultForwarder) AckMessages() <-chan UpstreamAck

AckMessages возвращает канал с UpstreamAck для передачи далее по пайплайну.

func (*DefaultForwarder) ChangeOut

func (f *DefaultForwarder) ChangeOut(oldOut, newOut string) error

ChangeOut изменяет выходной поток из oldOut в newOut

func (*DefaultForwarder) Forward

func (f *DefaultForwarder) Forward(inputID uint16, inputMsgID uint32, data []byte) error

Forward отправляет сообщение дальше с гарантиями доставки.

func (*DefaultForwarder) GetOldestOutput

func (f *DefaultForwarder) GetOldestOutput() (uint32, error)

GetOldestOutput возвращает самый старый output_message_id, который хранится в логе.

func (*DefaultForwarder) Run

func (f *DefaultForwarder) Run(ctx context.Context) error

Run запускает Forwarder и блокируется.

type DefaultForwarderConfig

type DefaultForwarderConfig struct {
	ACKPeriod     time.Duration
	ForwardLogDir string
}

DefaultForwarderConfig набор параметров для DefaultForwarder.

type DefaultReceiver

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

DefaultReceiver получает сообщения от вышестоящих узлов.

func NewDefaultReceiver

func NewDefaultReceiver(addr string, inNames []string, l *util.Logger) *DefaultReceiver

NewDefaultReceiver возвращает новый объект DefaultReceiver.

func (*DefaultReceiver) Acks

func (r *DefaultReceiver) Acks() chan<- UpstreamAck

Acks возвращает канал для передачи Ack сообщений вверх по потоку.

func (*DefaultReceiver) Messages

func (r *DefaultReceiver) Messages() <-chan *UpstreamMessage

Messages возвращает канал с сообщениями.

func (*DefaultReceiver) Run

func (r *DefaultReceiver) Run(ctx context.Context) error

Run запускает обработчик.

type DownstreamForwarder

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

DownstreamForwarder клиент для передачи сообщений далее по пайплайну.

func NewDownstreamForwarder

func NewDownstreamForwarder(downstreamIndex uint16, name string, addr string, iter *LogBufferIterator, l *util.Logger) *DownstreamForwarder

NewDownstreamForwarder создает новый объект DownstreamForwarder.

func (*DownstreamForwarder) Run

func (f *DownstreamForwarder) Run(ctx context.Context) error

Run запускает клиент действия и блокируется до окончания работы.

type ForwardLog

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

ForwardLog лог для записи сообщений с целью обеспечения отказоустойчивости.

func NewForwardLog

func NewForwardLog(forwardLogDir string) (*ForwardLog, error)

NewForwardLog создает новый ForwardLog.

func (*ForwardLog) Close

func (l *ForwardLog) Close() error

Close закрывает ForwardLog и очищает занимаемые ресурсы.

func (*ForwardLog) GetOldestOutput

func (l *ForwardLog) GetOldestOutput() (uint32, error)

GetOldestOutput возвращает самый старый output_message_id, который хранится в логе.

func (*ForwardLog) NewIterator

func (l *ForwardLog) NewIterator() *LogBufferIterator

NewIterator возвращает итератор, который позволяет двигаться по ForwardLog с первой записи в прямом направлении.

func (*ForwardLog) Trim

func (l *ForwardLog) Trim(idBorder uint32) (map[uint16]uint32, error)

Trim отрезает от лога все сообщения, у которых output_id <= idBorder. Может обрезать сообщения одновременно с записью, так как никогда не будет обрабатывать одно и то же сообщение из-за того, что отправка происходит после записи в лог, а значит если мы получили подтверждение на это сообщение, то оно уже было отправлено.

func (*ForwardLog) Write

func (l *ForwardLog) Write(inputID uint16, inputMsgID, outputMsgID uint32, data []byte) error

type LogBufferIterator

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

LogBufferIterator итератор для прямого передвижения по logBuffer.

func NewLogBufferIterator

func NewLogBufferIterator(logBuffer *logBuffer) *LogBufferIterator

NewLogBufferIterator содает новый LogBufferIterator, чтение начинается с начала лога.

func (*LogBufferIterator) Next

func (i *LogBufferIterator) Next(ctx context.Context, item *forwardLogItem) error

Next загружает в item следющий элемент. В случае, если итератор находится в конце лога, блокируется в ожидании новых записей.

type UpstreamAck

type UpstreamAck map[uint16]uint32

UpstreamAck отображение upstream_id в содержание ACK сообщения.

func (UpstreamAck) String

func (a UpstreamAck) String() string

type UpstreamMessage

type UpstreamMessage struct {
	InputID uint16
	// contains filtered or unexported fields
}

UpstreamMessage сообщение из вышестоящего узла.

type UpstreamReceiver

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

UpstreamReceiver структура, для получения сообщений от узлов выше по потоку.

func NewUpstreamReceiver

func NewUpstreamReceiver(upstreamIndex uint16, name string, tcpConn *connutil.Connection, l *util.Logger) *UpstreamReceiver

NewUpstreamReceiver создает новый UpstreamReceiver.

func (*UpstreamReceiver) Ack

func (r *UpstreamReceiver) Ack(ack uint32) error

Ack передает ACK сообщение вверх по потку.

func (*UpstreamReceiver) Run

func (r *UpstreamReceiver) Run(ctx context.Context) error

Run запускает UpstreamReceiver и блокируется.