gno.land/p/g17khqpukees4237dtn3astzapmp462vjhsz6st4/message⧉

1package message23import (4	"errors"5	"strings"67	"gno.land/p/moul/ulist/v0"8	"gno.land/p/nt/bptree/v0"9)1011var (12	// ErrInvalidTopic is triggered when an invalid topic is used.13	ErrInvalidTopic = errors.New("invalid topic")1415	// ErrRequiredCallback is triggered when subscribing without a callback.16	ErrRequiredCallback = errors.New("message callback is required")1718	// ErrRequiredSubscriptionID is triggered when unsubscribing without an ID.19	ErrRequiredSubscriptionID = errors.New("message sibscription ID is required")2021	// ErrRequiredTopic is triggered when (un)subscribing without a topic.22	ErrRequiredTopic = errors.New("message topic is required")23)2425// NewBroker creates a new message broker.26func NewBroker() *Broker {27	return &Broker{28		callbacks: bptree.NewBPTree32(),29	}30}3132// Broker is a message broker that handles subscriptions and message publishing.33type Broker struct {34	callbacks *bptree.BPTree // string(topic) -> *ulist.List(Callback)35}3637// Topics returns the list of current subscription topics.38func (b *Broker) Topics() []Topic {39	var topics []Topic40	b.callbacks.Iterate("", "", func(k string, _ any) bool {41		topic := Topic(k)42		if topic == TopicAll {43			// Skip catchall topic from the list44			return false45		}4647		topics = append(topics, topic)48		return false49	})50	return topics51}5253// Subscribe subscribes to messages published for a topic.54// It returns the callback ID within the topic.55func (b *Broker) Subscribe(topic Topic, cb Callback) (id int, _ error) {56	key := strings.TrimSpace(string(topic))57	if key == "" {58		return 0, ErrRequiredTopic59	}6061	if cb == nil {62		return 0, ErrRequiredCallback63	}6465	callbacks, _ := b.callbacks.Get(key).(*ulist.List)66	if callbacks == nil {67		callbacks = ulist.New()68	}6970	callbacks.Append(cb)71	b.callbacks.Set(key, callbacks)72	return callbacks.TotalSize(), nil73}7475// Unsubscribe unsubscribes a callback from a message topic.76// ID is the callback ID within the topic, returned on subscription.77func (b *Broker) Unsubscribe(topic Topic, id int) (unsubscribed bool, _ error) {78	key := strings.TrimSpace(string(topic))79	if key == "" {80		return false, ErrRequiredTopic81	}8283	if id == 0 {84		return false, ErrRequiredSubscriptionID85	}8687	callbacks, ok := b.callbacks.Get(key).(*ulist.List)88	if !ok {89		return false, errors.New("message topic not found: " + key)90	}9192	i := id - 193	return callbacks.Delete(i) == nil, nil94}9596// Publish publishes a message for a topic.97func (b *Broker) Publish(topic Topic, data any) error {98	if topic == TopicAll {99		return ErrInvalidTopic100	}101102	key := strings.TrimSpace(string(topic))103	if key == "" {104		return ErrRequiredTopic105	}106107	iterCb := func(_ int, v any) bool {108		cb := v.(Callback)109		cb(Message{topic, data})110		return false111	}112113	// Trigger callbacks subscribed to current topic114	if callbacks, ok := b.callbacks.Get(key).(*ulist.List); ok {115		callbacks.Iterator(0, callbacks.Size(), iterCb)116	}117118	// Trigger callbacks subscribed to all topics119	if callbacks, ok := b.callbacks.Get(string(TopicAll)).(*ulist.List); ok {120		callbacks.Iterator(0, callbacks.Size(), iterCb)121	}122	return nil123}124