Method run

pubsub/bus.go:108–178

()

}

func (b *bus) run() {
	defer b.lc.ShutdownCompleted()

	var outch chan<- Event
	var curev Event

loop:
	for {

		if b.eventch != nil && len(b.evbuf) > 0 {
			// If we're emitting events (Subscriber mode) and there
			// are events to emit, set up the output channel and output
			// event accordingly.
			outch = b.eventch
			curev = b.evbuf[0]
		} else {
			// otherwise block the output (sending to a nil channel always blocks)
			outch = nil
		}

		select {
		case err := <-b.lc.ShutdownRequest():
			b.lc.ShutdownInitiated(err)
			break loop

		case outch <- curev:
			// Event was emitted. Shrink current event buffer.
			b.evbuf = b.evbuf[1:]

		case ev := <-b.pubch:
			// publish event

			// Buffer event.
			if b.eventch != nil {
				b.evbuf = append(b.evbuf, ev)
			}

			// Publish to children.
			for sub := range b.subscriptions {
				if err := sub.Publish(ev); err != nil && !errors.Is(err, ErrNotRunning) {
					panic(err)
				}
			}

		case ch := <-b.subch:
			// new subscription

			sub := newSubscriber(b)
			b.subscriptions[sub] = true

			ch <- sub

		case sub := <-b.unsubch:
			// subscription closed
			delete(b.subscriptions, sub)
		}
	}