messagequeue

Documentation

Index

• type Allocator
• type Builder
  • func NewBuilder(ctx context.Context, topic Topic) *Builder
  • func (b *Builder) AddBlockData(requestID graphsync.RequestID, blockData graphsync.BlockData)
  • func (b *Builder) BlockData() map[graphsync.RequestID][]graphsync.BlockData
  • func (b *Builder) Context() context.Context
  • func (b *Builder) ResponseStreams() map[graphsync.RequestID]io.Closer
  • func (b *Builder) ScrubResponses(requestIDs []graphsync.RequestID) uint64
  • func (b *Builder) SetResponseStream(requestID graphsync.RequestID, stream io.Closer)
  • func (b *Builder) SetSubscriber(requestID graphsync.RequestID, subscriber notifications.Subscriber)
  • func (b *Builder) Subscribers() map[graphsync.RequestID]notifications.Subscriber
• type Event
• type EventName
• type MessageNetwork
• type MessageQueue
  • func New(ctx context.Context, p peer.ID, network MessageNetwork, allocator Allocator, ...) *MessageQueue
  • func (mq *MessageQueue) AllocateAndBuildMessage(size uint64, buildMessageFn func(*Builder))
  • func (mq *MessageQueue) Shutdown()
  • func (mq *MessageQueue) Startup()
• type Metadata
• type Topic

Types

type Allocator

type Allocator interface {
	AllocateBlockMemory(p peer.ID, amount uint64) <-chan error
	ReleasePeerMemory(p peer.ID) error
	ReleaseBlockMemory(p peer.ID, amount uint64) error
}

type Builder

type Builder struct {
	*gsmsg.Builder
	// contains filtered or unexported fields
}

Builder wraps a message builder with additional functions related to metadata and notifications in the message queue

func NewBuilder

func NewBuilder(ctx context.Context, topic Topic) *Builder

NewBuilder sets up a new builder for the given topic

func (*Builder) AddBlockData

func (b *Builder) AddBlockData(requestID graphsync.RequestID, blockData graphsync.BlockData)

AddBlockData add the given block metadata for this message to pass into notifications

func (*Builder) BlockData

func (b *Builder) BlockData() map[graphsync.RequestID][]graphsync.BlockData

BlockData inspects current block data

func (*Builder) Context

func (b *Builder) Context() context.Context

func (*Builder) ResponseStreams

func (b *Builder) ResponseStreams() map[graphsync.RequestID]io.Closer

ResponseStreams inspect current response stream state

func (*Builder) ScrubResponses

func (b *Builder) ScrubResponses(requestIDs []graphsync.RequestID) uint64

ScrubResponse removes the given responses from the message and metadata

func (*Builder) SetResponseStream

func (b *Builder) SetResponseStream(requestID graphsync.RequestID, stream io.Closer)

SetResponseStream sets the given response stream to close should the message fail to send

func (*Builder) SetSubscriber

func (b *Builder) SetSubscriber(requestID graphsync.RequestID, subscriber notifications.Subscriber)

SetSubscriber sets the given subscriber to get notified as events occur for this message

func (*Builder) Subscribers

func (b *Builder) Subscribers() map[graphsync.RequestID]notifications.Subscriber

Subscribers inspect current subscribers

type Event

type Event struct {
	Name     EventName
	Err      error
	Metadata Metadata
}

type EventName

type EventName uint64

const (
	Queued EventName = iota
	Sent
	Error
)

type MessageNetwork

type MessageNetwork interface {
	NewMessageSender(context.Context, peer.ID, gsnet.MessageSenderOpts) (gsnet.MessageSender, error)
	ConnectTo(context.Context, peer.ID) error
}

MessageNetwork is any network that can connect peers and generate a message sender.

type MessageQueue

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

MessageQueue implements queue of want messages to send to peers.

func New

func New(ctx context.Context, p peer.ID, network MessageNetwork, allocator Allocator, maxRetries int, sendMessageTimeout time.Duration, onShutdown func(peer.ID)) *MessageQueue

New creats a new MessageQueue.

func (*MessageQueue) AllocateAndBuildMessage

func (mq *MessageQueue) AllocateAndBuildMessage(size uint64, buildMessageFn func(*Builder))

AllocateAndBuildMessage allows you to work modify the next message that is sent in the queue. If blkSize > 0, message building may block until enough memory has been freed from the queues to allocate the message.

func (*MessageQueue) Shutdown

func (mq *MessageQueue) Shutdown()

Shutdown stops the processing of messages for a message queue.

func (*MessageQueue) Startup

func (mq *MessageQueue) Startup()

Startup starts the processing of messages, and creates an initial message based on the given initial wantlist.

type Metadata

type Metadata struct {
	BlockData     map[graphsync.RequestID][]graphsync.BlockData
	ResponseCodes map[graphsync.RequestID]graphsync.ResponseStatusCode
}

type Topic

type Topic uint64