messagebus

Documentation

Overview

Package messagebus routes messages (contract calls & responses) and delivers'em to relevant modules.

Usage:

TODO:

Index

• Variables
• func GetMessageHash(scheme core.PlatformCryptographyScheme, msg core.Parcel) []byte
• func NewParcelFactory() message.ParcelFactory
• type MessageBus
  • func NewMessageBus(config configuration.Configuration) (*MessageBus, error)
  • func (mb *MessageBus) CreateParcel(ctx context.Context, msg core.Message, token core.DelegationToken, ...) (core.Parcel, error)
  • func (mb *MessageBus) Lock(ctx context.Context)
  • func (mb *MessageBus) MustRegister(p core.MessageType, handler core.MessageHandler)
  • func (mb *MessageBus) NewPlayer(ctx context.Context, reader io.Reader) (core.MessageBus, error)
  • func (mb *MessageBus) NewRecorder(ctx context.Context, currentPulse core.Pulse) (core.MessageBus, error)
  • func (mb *MessageBus) OnPulse(context.Context, core.Pulse) error
  • func (mb *MessageBus) Register(p core.MessageType, handler core.MessageHandler) error
  • func (mb *MessageBus) Send(ctx context.Context, msg core.Message, ops *core.MessageSendOptions) (core.Reply, error)
  • func (mb *MessageBus) SendParcel(ctx context.Context, parcel core.Parcel, currentPulse core.Pulse, ...) (core.Reply, error)
  • func (mb *MessageBus) Start(ctx context.Context) error
  • func (mb *MessageBus) Stop(ctx context.Context) error
  • func (mb *MessageBus) Unlock(ctx context.Context)
• type TapeItem

Variables

var (
	// ErrNoReply is returned from player when there is no stored reply for provided message.
	ErrNoReply = errors.New("no such reply")
)

Functions

func GetMessageHash

func GetMessageHash(scheme core.PlatformCryptographyScheme, msg core.Parcel) []byte

GetMessageHash calculates message hash.

func NewParcelFactory

func NewParcelFactory() message.ParcelFactory

NewParcelFactory returns new instance of parcelFactory

Types

type MessageBus

type MessageBus struct {
	Network                    core.Network                    `inject:""`
	JetCoordinator             core.JetCoordinator             `inject:""`
	LocalStorage               core.LocalStorage               `inject:""`
	NodeNetwork                core.NodeNetwork                `inject:""`
	PlatformCryptographyScheme core.PlatformCryptographyScheme `inject:""`
	CryptographyService        core.CryptographyService        `inject:""`
	DelegationTokenFactory     core.DelegationTokenFactory     `inject:""`
	ParcelFactory              message.ParcelFactory           `inject:""`
	PulseStorage               core.PulseStorage               `inject:""`

	NextPulseMessagePoolChan    chan interface{}
	NextPulseMessagePoolCounter uint32
	NextPulseMessagePoolLock    sync.RWMutex
	// contains filtered or unexported fields
}

MessageBus is component that routes application logic requests, e.g. glue between network and logic runner

func NewMessageBus

func NewMessageBus(config configuration.Configuration) (*MessageBus, error)

NewMessageBus creates plain MessageBus instance. It can be used to create Player and Recorder instances that wrap it, providing additional functionality.

func (*MessageBus) CreateParcel

func (mb *MessageBus) CreateParcel(ctx context.Context, msg core.Message, token core.DelegationToken, currentPulse core.Pulse) (core.Parcel, error)

CreateParcel creates signed message from provided message.

func (*MessageBus) Lock

func (mb *MessageBus) Lock(ctx context.Context)

func (*MessageBus) MustRegister

func (mb *MessageBus) MustRegister(p core.MessageType, handler core.MessageHandler)

MustRegister is a Register wrapper that panics if an error was returned.

func (*MessageBus) NewPlayer

func (mb *MessageBus) NewPlayer(ctx context.Context, reader io.Reader) (core.MessageBus, error)

NewPlayer creates a new player from stream. This is a very long operation, as it saves replies in storage until the stream is exhausted.

Player can be created from MessageBus and passed as MessageBus instance.

func (*MessageBus) NewRecorder

func (mb *MessageBus) NewRecorder(ctx context.Context, currentPulse core.Pulse) (core.MessageBus, error)

NewRecorder creates a new recorder with unique tape that can be used to store message replies.

Recorder can be created from MessageBus and passed as MessageBus instance.

func (*MessageBus) OnPulse

func (mb *MessageBus) OnPulse(context.Context, core.Pulse) error

func (*MessageBus) Register

func (mb *MessageBus) Register(p core.MessageType, handler core.MessageHandler) error

Register sets a function as a handler for particular message type, only one handler per type is allowed

func (*MessageBus) Send

func (mb *MessageBus) Send(ctx context.Context, msg core.Message, ops *core.MessageSendOptions) (core.Reply, error)

Send an `Message` and get a `Value` or error from remote host.

func (*MessageBus) SendParcel

func (mb *MessageBus) SendParcel(
	ctx context.Context,
	parcel core.Parcel,
	currentPulse core.Pulse,
	options *core.MessageSendOptions,
) (core.Reply, error)

SendParcel sends provided message via network.

func (*MessageBus) Start

func (mb *MessageBus) Start(ctx context.Context) error

Start initializes message bus.

func (*MessageBus) Stop

func (mb *MessageBus) Stop(ctx context.Context) error

Stop releases resources and stops the bus

func (*MessageBus) Unlock

func (mb *MessageBus) Unlock(ctx context.Context)

type TapeItem

type TapeItem struct {
	Reply core.Reply
	Error error
}

TapeItem stores reply/error pair for tape.