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
• func NewTape(ls core.LocalStorage, pulse core.PulseNumber) (*storageTape, error)
• func NewTapeFromReader(ctx context.Context, ls core.LocalStorage, r io.Reader) (*storageTape, error)
• type MessageBus
  • func NewMessageBus(config configuration.Configuration) (*MessageBus, error)
  • func (mb *MessageBus) Acquire(ctx context.Context)
  • func (mb *MessageBus) CreateParcel(ctx context.Context, msg core.Message, token core.DelegationToken) (core.Parcel, error)
  • func (mb *MessageBus) MustRegister(p core.MessageType, handler core.MessageHandler)
  • func (mb *MessageBus) NewPlayer(ctx context.Context, r io.Reader) (core.MessageBus, error)
  • func (mb *MessageBus) NewRecorder(ctx context.Context) (core.MessageBus, error)
  • func (mb *MessageBus) Register(p core.MessageType, handler core.MessageHandler) error
  • func (mb *MessageBus) Release(ctx context.Context)
  • 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, options *core.MessageSendOptions) (core.Reply, error)
  • func (mb *MessageBus) Start(ctx context.Context) error
  • func (mb *MessageBus) Stop(ctx context.Context) error
  • func (mb *MessageBus) WriteTape(ctx context.Context, writer io.Writer) error

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

func NewTape

func NewTape(ls core.LocalStorage, pulse core.PulseNumber) (*storageTape, error)

NewTape creates new storageTape with random id.

func NewTapeFromReader

func NewTapeFromReader(ctx context.Context, ls core.LocalStorage, r io.Reader) (*storageTape, error)

NewTapeFromReader creates and fills a new storageTape from a stream.

This is a very long operation, as it saves replies in storage until the stream is exhausted.

Types

type MessageBus

type MessageBus struct {
	Service                    core.Network                    `inject:""`
	JetCoordinator             core.JetCoordinator             `inject:""`
	LocalStorage               core.LocalStorage               `inject:""`
	PulseManager               core.PulseManager               `inject:""`
	ActiveNodes                core.NodeNetwork                `inject:""`
	PlatformCryptographyScheme core.PlatformCryptographyScheme `inject:""`
	CryptographyService        core.CryptographyService        `inject:""`
	DelegationTokenFactory     core.DelegationTokenFactory     `inject:""`
	ParcelFactory              message.ParcelFactory           `inject:""`
	// 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) Acquire

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

func (*MessageBus) CreateParcel

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

CreateParcel creates signed message from provided message.

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, r 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) (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) 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) Release

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

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, 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) WriteTape

func (mb *MessageBus) WriteTape(ctx context.Context, writer io.Writer) error

WriteTape for MessageBus is not available.