// Dispatch dispatch a event to specific topic, there may be time duration between executing and dispatching
Dispatch dispatch a event to specific topic, there may be time duration between executing and dispatching
// Dispatch dispatch a event to specific topic there may be time duration between executing and dispatching
func (d *localDispatcher) Dispatch(ctx context.Context, topic Topic, evt Event) error {
	if d.isStopped() {
		return ErrDispatcherNotWorking
	}
	if d.total.Load() >= d.options.MaxQueuedEvents {
		return ErrMaxQueuedEventsReached
	}
	events := d.events[topic]
	if nil == events || len(events.handlers) == 0 {
		return ErrNoHandlerFound
	}
	emitAt := time.Now()
	if delayedEvt, ok := evt.(DelayedEvent); ok {
		emitAt = emitAt.Add(delayedEvt.Delayed())
	}
	events.PushEvent(ctx, &DispatchedEvent{Event: evt, EmitAt: emitAt})
	d.total.Inc()
	d.changed <- events
	return nil
}
func ( d * localDispatcher ) Dispatch ( ctx context . Context , topic Topic , evt Event ) error { if d . isStopped ( ) { return ErrDispatcherNotWorking } if d . total . Load ( ) >= d . options . MaxQueuedEvents { return ErrMaxQueuedEventsReached } events := d . events [ topic ] if nil == events || len ( events . handlers ) == 0 { return ErrNoHandlerFound } emitAt := time . Now ( ) if delayedEvt , ok := evt . ( DelayedEvent ) ; ok { emitAt = emitAt . Add ( delayedEvt . Delayed ( ) ) } events . PushEvent ( ctx , & DispatchedEvent { Event : evt , EmitAt : emitAt } ) d . total . Inc ( ) d . changed <- events return nil }
Dispatch dispatch a event to specific topic, there may be time duration between executing and dispatching
Dispatch dispatch a event to specific topic there may be time duration between executing and dispatching
//check max queued size //local localDispatcher should register handlers first //add emit time