dispatcher

Documentation

Index

• func NewBlockEvent(block *cb.Block, sourceURL string) *fab.BlockEvent
• func NewChaincodeEvent(chaincodeID, eventName, txID string, payload []byte, blockNum uint64, ...) *fab.CCEvent
• func NewFilteredBlockEvent(fblock *pb.FilteredBlock, sourceURL string) *fab.FilteredBlockEvent
• func NewTxStatusEvent(txID string, txValidationCode pb.TxValidationCode, blockNum uint64, ...) *fab.TxStatusEvent
• func WithEventConsumerBufferSize(value uint) options.Opt
• func WithEventConsumerTimeout(value time.Duration) options.Opt
• type BlockReg
• type ChaincodeReg
• type Dispatcher
  • func New(opts ...options.Opt) *Dispatcher
  • func (ed *Dispatcher) EventCh() (chan<- interface{}, error)
  • func (ed *Dispatcher) HandleBlock(block *cb.Block, sourceURL string)
  • func (ed *Dispatcher) HandleFilteredBlock(fblock *pb.FilteredBlock, sourceURL string)
  • func (ed *Dispatcher) HandleStopEvent(e Event)
  • func (ed *Dispatcher) LastBlockNum() uint64
  • func (ed *Dispatcher) RegisterHandler(t interface{}, h Handler)
  • func (ed *Dispatcher) RegisterHandlers()
  • func (p *Dispatcher) SetEventConsumerBufferSize(value uint)
  • func (p *Dispatcher) SetEventConsumerTimeout(value time.Duration)
  • func (ed *Dispatcher) Start() error
• type Event
• type FilteredBlockReg
• type Handler
• type HandlerRegistry
• type RegisterBlockEvent
  • func NewRegisterBlockEvent(filter fab.BlockFilter, eventch chan<- *fab.BlockEvent, ...) *RegisterBlockEvent
• type RegisterChaincodeEvent
  • func NewRegisterChaincodeEvent(ccID, eventFilter string, eventch chan<- *fab.CCEvent, ...) *RegisterChaincodeEvent
• type RegisterEvent
  • func NewRegisterEvent(respch chan<- fab.Registration, errCh chan<- error) RegisterEvent
• type RegisterFilteredBlockEvent
  • func NewRegisterFilteredBlockEvent(eventch chan<- *fab.FilteredBlockEvent, respch chan<- fab.Registration, ...) *RegisterFilteredBlockEvent
• type RegisterTxStatusEvent
  • func NewRegisterTxStatusEvent(txID string, eventch chan<- *fab.TxStatusEvent, respch chan<- fab.Registration, ...) *RegisterTxStatusEvent
• type RegistrationInfo
• type RegistrationInfoEvent
  • func NewRegistrationInfoEvent(regInfoCh chan<- *RegistrationInfo) *RegistrationInfoEvent
• type StopEvent
  • func NewStopEvent(errch chan<- error) *StopEvent
• type TxStatusReg
• type UnregisterEvent
  • func NewUnregisterEvent(reg fab.Registration) *UnregisterEvent

Functions

func NewBlockEvent

func NewBlockEvent(block *cb.Block, sourceURL string) *fab.BlockEvent

NewBlockEvent creates a new BlockEvent

func NewChaincodeEvent

func NewChaincodeEvent(chaincodeID, eventName, txID string, payload []byte, blockNum uint64, sourceURL string) *fab.CCEvent

NewChaincodeEvent creates a new ChaincodeEvent

func NewFilteredBlockEvent

func NewFilteredBlockEvent(fblock *pb.FilteredBlock, sourceURL string) *fab.FilteredBlockEvent

NewFilteredBlockEvent creates a new FilteredBlockEvent

func NewTxStatusEvent

func NewTxStatusEvent(txID string, txValidationCode pb.TxValidationCode, blockNum uint64, sourceURL string) *fab.TxStatusEvent

NewTxStatusEvent creates a new TxStatusEvent

func WithEventConsumerBufferSize

func WithEventConsumerBufferSize(value uint) options.Opt

WithEventConsumerBufferSize sets the size of the registered consumer's event channel.

func WithEventConsumerTimeout

func WithEventConsumerTimeout(value time.Duration) options.Opt

WithEventConsumerTimeout is the timeout when sending events to a registered consumer. If < 0, if buffer full, unblocks immediately and does not send. If 0, if buffer full, will block and guarantee the event will be sent out. If > 0, if buffer full, blocks util timeout.

Types

type BlockReg

type BlockReg struct {
	Filter  fab.BlockFilter
	Eventch chan<- *fab.BlockEvent
}

BlockReg contains the data for a block registration

type ChaincodeReg

type ChaincodeReg struct {
	ChaincodeID string
	EventFilter string
	EventRegExp *regexp.Regexp
	Eventch     chan<- *fab.CCEvent
}

ChaincodeReg contains the data for a chaincode registration

type Dispatcher

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

Dispatcher is responsible for handling all events, including connection and registration events originating from the client, and events originating from the channel event service. All events are processed in a single Go routine in order to avoid any race conditions and to ensure that events are processed in the order in which they are received. This also avoids the need for synchronization.

func New

func New(opts ...options.Opt) *Dispatcher

New creates a new Dispatcher.

func (*Dispatcher) EventCh

func (ed *Dispatcher) EventCh() (chan<- interface{}, error)

EventCh returns the channel to which events may be posted

func (*Dispatcher) HandleBlock

func (ed *Dispatcher) HandleBlock(block *cb.Block, sourceURL string)

HandleBlock handles a block event

func (*Dispatcher) HandleFilteredBlock

func (ed *Dispatcher) HandleFilteredBlock(fblock *pb.FilteredBlock, sourceURL string)

HandleFilteredBlock handles a filtered block event

func (*Dispatcher) HandleStopEvent

func (ed *Dispatcher) HandleStopEvent(e Event)

HandleStopEvent stops the dispatcher and unregisters all event registration. The Dispatcher is no longer usable.

func (*Dispatcher) LastBlockNum

func (ed *Dispatcher) LastBlockNum() uint64

LastBlockNum returns the block number of the last block for which an event was received.

func (*Dispatcher) RegisterHandler

func (ed *Dispatcher) RegisterHandler(t interface{}, h Handler)

RegisterHandler registers an event handler

func (*Dispatcher) RegisterHandlers

func (ed *Dispatcher) RegisterHandlers()

RegisterHandlers registers all of the handlers by event type

func (*Dispatcher) SetEventConsumerBufferSize

func (p *Dispatcher) SetEventConsumerBufferSize(value uint)

func (*Dispatcher) SetEventConsumerTimeout

func (p *Dispatcher) SetEventConsumerTimeout(value time.Duration)

func (*Dispatcher) Start

func (ed *Dispatcher) Start() error

Start starts dispatching events as they arrive. All events are processed in a single Go routine in order to avoid any race conditions

type Event

type Event interface{}

Event is an event that's sent to the dispatcher. This includes client registration requests or events that come from an event producer.

type FilteredBlockReg

type FilteredBlockReg struct {
	Eventch chan<- *fab.FilteredBlockEvent
}

FilteredBlockReg contains the data for a filtered block registration

type Handler

type Handler func(Event)

Handler is the handler for a given event type.

type HandlerRegistry

type HandlerRegistry map[reflect.Type]Handler

HandlerRegistry contains the handlers for each type of event

type RegisterBlockEvent

type RegisterBlockEvent struct {
	RegisterEvent
	Reg *BlockReg
}

RegisterBlockEvent registers for block events

func NewRegisterBlockEvent

func NewRegisterBlockEvent(filter fab.BlockFilter, eventch chan<- *fab.BlockEvent, respch chan<- fab.Registration, errCh chan<- error) *RegisterBlockEvent

NewRegisterBlockEvent creates a new RegisterBlockEvent

type RegisterChaincodeEvent

type RegisterChaincodeEvent struct {
	RegisterEvent
	Reg *ChaincodeReg
}

RegisterChaincodeEvent registers for chaincode events

func NewRegisterChaincodeEvent

func NewRegisterChaincodeEvent(ccID, eventFilter string, eventch chan<- *fab.CCEvent, respch chan<- fab.Registration, errCh chan<- error) *RegisterChaincodeEvent

NewRegisterChaincodeEvent creates a new RegisterChaincodeEvent

type RegisterEvent

type RegisterEvent struct {
	RegCh chan<- fab.Registration
	ErrCh chan<- error
}

RegisterEvent is the base for all registration events.

func NewRegisterEvent

func NewRegisterEvent(respch chan<- fab.Registration, errCh chan<- error) RegisterEvent

NewRegisterEvent creates a new RgisterEvent

type RegisterFilteredBlockEvent

type RegisterFilteredBlockEvent struct {
	RegisterEvent
	Reg *FilteredBlockReg
}

RegisterFilteredBlockEvent registers for filtered block events

func NewRegisterFilteredBlockEvent

func NewRegisterFilteredBlockEvent(eventch chan<- *fab.FilteredBlockEvent, respch chan<- fab.Registration, errCh chan<- error) *RegisterFilteredBlockEvent

NewRegisterFilteredBlockEvent creates a new RegisterFilterBlockEvent

type RegisterTxStatusEvent

type RegisterTxStatusEvent struct {
	RegisterEvent
	Reg *TxStatusReg
}

RegisterTxStatusEvent registers for transaction status events

func NewRegisterTxStatusEvent

func NewRegisterTxStatusEvent(txID string, eventch chan<- *fab.TxStatusEvent, respch chan<- fab.Registration, errCh chan<- error) *RegisterTxStatusEvent

NewRegisterTxStatusEvent creates a new RegisterTxStatusEvent

type RegistrationInfo

type RegistrationInfo struct {
	TotalRegistrations            int
	NumBlockRegistrations         int
	NumFilteredBlockRegistrations int
	NumCCRegistrations            int
	NumTxStatusRegistrations      int
}

RegistrationInfo contains a snapshot of the current event registrations

type RegistrationInfoEvent

type RegistrationInfoEvent struct {
	RegInfoCh chan<- *RegistrationInfo
}

RegistrationInfoEvent requests registration information

func NewRegistrationInfoEvent

func NewRegistrationInfoEvent(regInfoCh chan<- *RegistrationInfo) *RegistrationInfoEvent

NewRegistrationInfoEvent returns a new RegistrationInfoEvent

type StopEvent

type StopEvent struct {
	ErrCh chan<- error
}

StopEvent tells the dispatcher to stop processing

func NewStopEvent

func NewStopEvent(errch chan<- error) *StopEvent

NewStopEvent creates a new StopEvent

type TxStatusReg

type TxStatusReg struct {
	TxID    string
	Eventch chan<- *fab.TxStatusEvent
}

TxStatusReg contains the data for a transaction status registration

type UnregisterEvent

type UnregisterEvent struct {
	Reg fab.Registration
}

UnregisterEvent unregisters a registration

func NewUnregisterEvent

func NewUnregisterEvent(reg fab.Registration) *UnregisterEvent

NewUnregisterEvent creates a new UnregisterEvent