package core import ( "context" "errors" "fmt" "reflect" "sync" "sync/atomic" ) const maxHandlerParams = 2 var ctxInterfaceType = reflect.TypeFor[context.Context]() var errInterfaceType = reflect.TypeFor[error]() type eventListener struct { id uint64 fnVal reflect.Value numIn int hasCtx bool hasPayload bool argType reflect.Type returnsErr bool } // EventBus is a thread-safe, strongly-typed in-process domain event bus. type EventBus struct { mu sync.RWMutex nextID atomic.Uint64 handlers map[string][]eventListener } // NewEventBus creates a new EventBus instance. func NewEventBus() *EventBus { return &EventBus{ handlers: make(map[string][]eventListener), } } // On registers an event handler for the given topic. // // Supported handler signatures: // - func(ctx context.Context, event T) error // - func(ctx context.Context, event T) // - func(event T) error // - func(event T) // - func(ctx context.Context) error // - func(ctx context.Context) // - func() error // - func() // // Returns a Disposer function that unregisters the handler when called. func (b *EventBus) On(topic string, handler any) Disposer { if handler == nil { panic("core/events: handler cannot be nil") } fnVal := reflect.ValueOf(handler) fnType := fnVal.Type() if fnType.Kind() != reflect.Func { panic(fmt.Sprintf("core/events: expected func, got %s", fnType.Kind())) } numIn := fnType.NumIn() if numIn > maxHandlerParams { panic(fmt.Sprintf("core/events: handler has %d parameters, maximum 2 supported (ctx, event)", numIn)) } numOut := fnType.NumOut() if numOut > 1 { panic(fmt.Sprintf("core/events: handler has %d return values, maximum 1 supported (error)", numOut)) } returnsErr := false if numOut == 1 { outType := fnType.Out(0) if !outType.Implements(errInterfaceType) { panic(fmt.Sprintf("core/events: handler return type must be error, got %v", outType)) } returnsErr = true } listener := eventListener{ id: b.nextID.Add(1), fnVal: fnVal, numIn: numIn, returnsErr: returnsErr, } switch numIn { case 0: // func() or func() error case 1: in0 := fnType.In(0) if in0.Implements(ctxInterfaceType) { listener.hasCtx = true } else { listener.hasPayload = true listener.argType = in0 } case 2: in0 := fnType.In(0) if !in0.Implements(ctxInterfaceType) { panic(fmt.Sprintf("core/events: first parameter must implement context.Context, got %v", in0)) } listener.hasCtx = true listener.hasPayload = true listener.argType = fnType.In(1) } b.mu.Lock() b.handlers[topic] = append(b.handlers[topic], listener) b.mu.Unlock() listenerID := listener.id var disposed atomic.Bool return func() error { if disposed.Swap(true) { return nil } b.mu.Lock() defer b.mu.Unlock() list := b.handlers[topic] for i, l := range list { if l.id == listenerID { b.handlers[topic] = append(list[:i], list[i+1:]...) break } } if len(b.handlers[topic]) == 0 { delete(b.handlers, topic) } return nil } } // Subscribe registers a strongly-typed generic event listener on the given EventBus. func Subscribe[T any](bus *EventBus, topic string, handler func(ctx context.Context, event T) error) Disposer { if bus == nil { panic("core/events: nil EventBus provided to Subscribe") } return bus.On(topic, handler) } // Emit publishes an event to all subscribers of the specified topic. // Handlers are executed synchronously. If any handler panics or returns an error, // the error is collected and returned via errors.Join. // //nolint:contextcheck func (b *EventBus) Emit(ctx context.Context, topic string, payload any) error { if ctx == nil { ctx = context.Background() } b.mu.RLock() rawListeners := b.handlers[topic] if len(rawListeners) == 0 { b.mu.RUnlock() return nil } listeners := make([]eventListener, len(rawListeners)) copy(listeners, rawListeners) b.mu.RUnlock() var payloadVal reflect.Value if payload != nil { payloadVal = reflect.ValueOf(payload) } var errs []error for _, l := range listeners { args := b.buildArgs(ctx, l, payloadVal) err := func() (resErr error) { defer func() { if r := recover(); r != nil { resErr = fmt.Errorf("core/events: panic in handler for topic %q: %v", topic, r) } }() results := l.fnVal.Call(args) if l.returnsErr && len(results) > 0 && !results[0].IsNil() { resErr = results[0].Interface().(error) } return resErr }() if err != nil { errs = append(errs, err) } } return errors.Join(errs...) } func (b *EventBus) buildArgs(ctx context.Context, l eventListener, payloadVal reflect.Value) []reflect.Value { if l.numIn == 0 { return nil } args := make([]reflect.Value, 0, l.numIn) if l.hasCtx { args = append(args, reflect.ValueOf(ctx)) } if l.hasPayload { arg := b.convertPayload(payloadVal, l.argType) args = append(args, arg) } return args } func (b *EventBus) convertPayload(payloadVal reflect.Value, targetType reflect.Type) reflect.Value { if !payloadVal.IsValid() { return reflect.Zero(targetType) } valType := payloadVal.Type() // 1. Direct assignable if valType.AssignableTo(targetType) { return payloadVal } // 2. Direct convertible if valType.ConvertibleTo(targetType) { return payloadVal.Convert(targetType) } // 3. Payload is pointer *T, target expects T if valType.Kind() == reflect.Pointer && valType.Elem().AssignableTo(targetType) { if !payloadVal.IsNil() { return payloadVal.Elem() } return reflect.Zero(targetType) } // 4. Payload is value T, target expects *T if targetType.Kind() == reflect.Pointer && valType.AssignableTo(targetType.Elem()) { ptr := reflect.New(valType) ptr.Elem().Set(payloadVal) return ptr } // Fallback to zero value of targetType return reflect.Zero(targetType) } // Listeners returns the number of active listeners for a topic. func (b *EventBus) Listeners(topic string) int { b.mu.RLock() defer b.mu.RUnlock() return len(b.handlers[topic]) } // Topics returns all topics that have registered listeners. func (b *EventBus) Topics() []string { b.mu.RLock() defer b.mu.RUnlock() topics := make([]string, 0, len(b.handlers)) for t := range b.handlers { topics = append(topics, t) } return topics }