event

Documentation

Overview

Package event implements an event bus. for a great introduction to the event bus pattern in go, see: https://levelup.gitconnected.com/lets-write-a-simple-event-bus-in-go-79b9480d8997

Example

const (
	ETMainSaidHello   = Type("main:SaidHello")
	ETMainOpSucceeded = Type("main:OperationSucceeded")
	ETMainOpFailed    = Type("main:OperationFailed")
)

ctx, done := context.WithCancel(context.Background())
defer done()

bus := NewBus(ctx)

makeDoneHandler := func(label string) Handler {
	return func(ctx context.Context, t Type, payload interface{}) error {
		fmt.Printf("%s handler called\n", label)
		return nil
	}
}

bus.Subscribe(makeDoneHandler("first"), ETMainSaidHello, ETMainOpSucceeded)
bus.Subscribe(makeDoneHandler("second"), ETMainSaidHello)
bus.Subscribe(makeDoneHandler("third"), ETMainSaidHello)

bus.Publish(ctx, ETMainSaidHello, "hello")
bus.Publish(ctx, ETMainOpSucceeded, "operation worked!")

Output:

first handler called
second handler called
third handler called
first handler called

Index

• Constants
• Variables
• type Bus
  • func NewBus(ctx context.Context) Bus
• type DsChange
• type FSICreateLinkEvent
• type Handler
• type Publisher
• type RemoteEvent
• type Type
• type WatchfsChange

Examples

• Package

Constants

const (
	// ETDatasetNameInit is when a dataset is initialized
	// payload is a DsChange
	ETDatasetNameInit = Type("dataset:Init")
	// ETDatasetCommitChange is when a dataset changes its newest commit
	// payload is a DsChange
	ETDatasetCommitChange = Type("dataset:CommitChange")
	// ETDatasetDeleteAll is when a dataset is entirely deleted
	// payload is a DsChange
	ETDatasetDeleteAll = Type("dataset:DeleteAll")
	// ETDatasetRename is when a dataset is renamed
	// payload is a DsChange
	ETDatasetRename = Type("dataset:Rename")
	// ETDatasetCreateLink is when a dataset is linked to a working directory
	// payload is a DsChange
	ETDatasetCreateLink = Type("dataset:CreateLink")
)

const (
	// ETCreatedNewFile is the event for creating a new file
	ETCreatedNewFile = Type("watchfs:CreatedNewFile")
	// ETModifiedFile is the event for modifying a file
	ETModifiedFile = Type("watchfs:ModifiedFile")
	// ETDeletedFile is the event for deleting a file
	ETDeletedFile = Type("watchfs:DeletedFile")
	// ETRenamedFolder is the event for renaming a folder
	ETRenamedFolder = Type("watchfs:RenamedFolder")
	// ETRemovedFolder is the event for removing a folder
	ETRemovedFolder = Type("watchfs:RemovedFolder")
)

const (
	// ETRemoteClientPushVersionProgress indicates a change in progress of a
	// dataset version push. Progress can fire as much as once-per-block.
	// subscriptions do not block the publisher
	// payload will be a RemoteEvent
	ETRemoteClientPushVersionProgress = Type("remoteClient:PushVersionProgress")
	// ETRemoteClientPushVersionCompleted indicates a version successfully pushed
	// to a remote.
	// payload will be a RemoteEvent
	ETRemoteClientPushVersionCompleted = Type("remoteClient:PushVersionCompleted")
	// ETRemoteClientPushDatasetCompleted indicates pushing a dataset
	// (logbook + versions) completed
	// payload will be a RemoteEvent
	ETRemoteClientPushDatasetCompleted = Type("remoteClient:PushDatasetCompleted")
	// ETRemoteClientPullVersionProgress indicates a change in progress of a
	// dataset version pull. Progress can fire as much as once-per-block.
	// subscriptions do not block the publisher
	// payload will be a RemoteEvent
	ETRemoteClientPullVersionProgress = Type("remoteClient:PullVersionProgress")
	// ETRemoteClientPullVersionCompleted indicates a version successfully pulled
	// from a remote.
	// payload will be a RemoteEvent
	ETRemoteClientPullVersionCompleted = Type("remoteClient:PullVersionCompleted")
	// ETRemoteClientPullDatasetCompleted indicates pulling a dataset
	// (logbook + versions) completed
	// payload will be a RemoteEvent
	ETRemoteClientPullDatasetCompleted = Type("remoteClient:PullDatasetCompleted")
	// ETRemoteClientRemoveDatasetCompleted indicates removing a dataset
	// (logbook + versions) remove completed
	// payload will be a RemoteEvent
	ETRemoteClientRemoveDatasetCompleted = Type("remoteClient:RemoveDatasetCompleted")
)

const (
	// ETFSICreateLinkEvent type for when FSI creates a link between a dataset
	// and working directory
	ETFSICreateLinkEvent = Type("fsi:CreateLinkEvent")
)

Variables

var (
	// ETP2PGoneOnline occurs when a p2p node opens up for peer-2-peer connections
	// payload will be []multiaddr.Addr, the listening addresses of this peer
	ETP2PGoneOnline = Type("p2p:GoneOnline")
	// ETP2PGoneOffline occurs when a p2p node has finished disconnecting from
	// a peer-2-peer network
	// payload will be nil
	ETP2PGoneOffline = Type("p2p:GoneOffline")
	// ETP2PQriPeerConnected occurs after a qri peer has connected to this node
	// payload will be a fully hydrated *profile.Profile from
	// "github.com/qri-io/qri/repo/profile"
	ETP2PQriPeerConnected = Type("p2p:QriPeerConnected")
	// ETP2PQriPeerDisconnected occurs after a qri peer has connected to this node
	// payload will be a fully hydrated *profile.Profile from
	// "github.com/qri-io/qri/repo/profile"
	// NOTE: not currently firing
	ETP2PQriPeerDisconnected = Type("p2p:QriPeerDisconnected")
	// ETP2PPeerConnected occurs after any peer has connected to this node
	// payload will be a libp2p.peerInfo
	ETP2PPeerConnected = Type("p2p:PeerConnected")
	// ETP2PPeerDisconnected occurs after any peer has connected to this node
	// payload will be a libp2p.peerInfo
	ETP2PPeerDisconnected = Type("p2p:PeerDisconnected")
	// ETP2PMessageReceived fires whenever the p2p protocol receives a message
	// from a Qri peer
	// payload will be a p2p.Message
	ETP2PMessageReceived = Type("p2p:MessageReceived")
)

var ETInstanceConstructed = Type("lib:InstanceConstructed")

ETInstanceConstructed is fired once a node is created payload is nil

var (

	// ErrBusClosed indicates the event bus is no longer coordinating events
	// because it's parent context has closed
	ErrBusClosed = fmt.Errorf("event bus is closed")
)

var NilBus = nilBus{}

NilBus replaces a nil value. it implements the bus interface, but does nothing

Types

type Bus

type Bus interface {
	// Publish an event to the bus
	Publish(ctx context.Context, t Type, data interface{}) error
	// Subscribe to one or more topics with a handler function that will be called
	// whenever the event topic is published
	Subscribe(handler Handler, topics ...Type)
	// NumSubscriptions returns the number of subscribers to the bus's events
	NumSubscribers() int
}

Bus is a central coordination point for event publication and subscription zero or more subscribers register topics to be notified of, a publisher writes a topic event to the bus, which broadcasts to all subscribers of that topic

func NewBus

func NewBus(ctx context.Context) Bus

NewBus creates a new event bus. Event busses should be instantiated as a singleton. If the passed in context is cancelled, the bus will stop emitting events and close all subscribed channels

TODO (b5) - finish context-closing cleanup

type DsChange

type DsChange struct {
	InitID     string
	TopIndex   int
	ProfileID  string
	Username   string
	PrettyName string
	HeadRef    string
	Info       *dsref.VersionInfo
	Dir        string
}

DsChange represents the result of a change to a dataset

type FSICreateLinkEvent

type FSICreateLinkEvent struct {
	FSIPath  string
	Username string
	Dsname   string
}

FSICreateLinkEvent describes an FSI created link

type Handler

type Handler func(ctx context.Context, t Type, payload interface{}) error

Handler is a function that will be called by the event bus whenever a subscribed topic is published. Handler calls are blocking, called in order of subscription. Any error returned by a handler is passed back to the event publisher. The handler context originates from the publisher, and in practice will often be scoped to a "request context" like an HTTP request or CLI command invocation. Generally, even handlers should aim to return quickly, and only delegate to goroutines when the publishing event is firing on a long-running process

type Publisher

type Publisher interface {
	Publish(ctx context.Context, t Type, payload interface{}) error
}

Publisher is an interface that can only publish an event

type RemoteEvent

type RemoteEvent struct {
	Ref        dsref.Ref
	RemoteAddr string
	Progress   dag.Completion
}

RemoteEvent encapsulates the push / pull progress of a dataset version

type Type

type Type string

Type is the set of all kinds of events emitted by the bus. Use the "Type" type to distinguish between different events. Event emitters should declare Types as constants and document the expected payload type.

type WatchfsChange

type WatchfsChange struct {
	Username    string
	Dsname      string
	Source      string
	Destination string
	Time        time.Time
}

WatchfsChange represents events for filesystem changes