eventbus

README

EventBus

Usage

package main

import (
	"context"
	"fmt"

	"github.com/go-kratos-ecosystem/components/v2/eventbus"
)

type Event struct {
	ID int
}

type Listener struct{}

var _ eventbus.Handler[*Event] = (*Listener)(nil)

func NewListener() *Listener {
	return &Listener{}
}

func (l Listener) Handle(ctx context.Context, msg *Event) error {
	fmt.Println("Listener", msg.ID)
	return nil
}

func main() {
	event := eventbus.NewEvent[*Event]()

	// on with HandlerFunc
	listener := event.On(eventbus.HandlerFunc[*Event](func(_ context.Context, msg *Event) error {
		fmt.Println("HandlerFunc", msg.ID)
		return nil
	}))
	// on with Listener
	event.On(NewListener())

	// emit
	_ = event.Emit(context.Background(), &Event{2})
	// Output:
	// HandlerFunc 2
	// Listener 2

	// emit with async
	_ = event.EmitAsync(context.Background(), &Event{2})
	_ = event.Emit(context.Background(), &Event{2}, eventbus.WithEmitAsync())

	// emit with skip errors
	_ = event.Emit(context.Background(), &Event{2}, eventbus.WithEmitSkipErrors())

	// off listener
	_ = listener.Off()

	// off all listeners
	event.OffAll()

	// listeners
	event.Listeners()

	// listeners count
	event.ListenersCount()
}

Documentation

Index

• Variables
• type EmitOption
  • func WithEmitAsync() EmitOption
  • func WithEmitSkipErrors() EmitOption
• type Event
  • func NewEvent[T any]() *Event[T]
  • func (t *Event[T]) Emit(ctx context.Context, msg T, opts ...EmitOption) error
  • func (t *Event[T]) EmitAsync(ctx context.Context, msg T, opts ...EmitOption) error
  • func (t *Event[T]) Listeners() []*Listener[T]
  • func (t *Event[T]) ListenersCount() int
  • func (t *Event[T]) Off(listener *Listener[T]) error
  • func (t *Event[T]) OffAll()
  • func (t *Event[T]) On(handler Handler[T]) *Listener[T]
• type Handler
• type HandlerFunc
  • func (f HandlerFunc[T]) Handle(ctx context.Context, msg T) error
• type Listener
  • func (l *Listener[T]) Off() error

Variables

var ErrListenerNotFound = errors.New("[event_bus] listener not found")

Types

type EmitOption

type EmitOption func(*emitOptions)

func WithEmitAsync

func WithEmitAsync() EmitOption

func WithEmitSkipErrors

func WithEmitSkipErrors() EmitOption

type Event

type Event[T any] struct {
	// contains filtered or unexported fields
}

func NewEvent

func NewEvent[T any]() *Event[T]

func (*Event[T]) Emit

func (t *Event[T]) Emit(ctx context.Context, msg T, opts ...EmitOption) error

func (*Event[T]) EmitAsync

func (t *Event[T]) EmitAsync(ctx context.Context, msg T, opts ...EmitOption) error

func (*Event[T]) Listeners

func (t *Event[T]) Listeners() []*Listener[T]

func (*Event[T]) ListenersCount

func (t *Event[T]) ListenersCount() int

func (*Event[T]) Off

func (t *Event[T]) Off(listener *Listener[T]) error

func (*Event[T]) OffAll

func (t *Event[T]) OffAll()

func (*Event[T]) On

func (t *Event[T]) On(handler Handler[T]) *Listener[T]

type Handler

type Handler[T any] interface {
	Handle(ctx context.Context, msg T) error
}

type HandlerFunc

type HandlerFunc[T any] func(ctx context.Context, msg T) error

func (HandlerFunc[T]) Handle

func (f HandlerFunc[T]) Handle(ctx context.Context, msg T) error

type Listener

type Listener[T any] struct {
	// contains filtered or unexported fields
}

func (*Listener[T]) Off

func (l *Listener[T]) Off() error