util.go

package api

import (
	"encoding/xml"
	sync "github.com/linkdata/deadlock"
	"reflect"
	"strconv"
	sync2 "sync"
	"sync/atomic"
)

// This is the index of the first actual subscription channel in sendCases.
// sendCases[0] is a SelectRecv case for the removeSub channel.
const firstSubSendCase = 1

// Subscription represents a stream of events. The carrier of the events is typically a
// channel, but isn't part of the interface.
//
// Subscriptions can fail while established. Failures are reported through an error
// channel. It receives a value if there is an issue with the subscription (e.g. the
// network connection delivering the events has been closed). Only one value will ever be
// sent.
//
// The error channel is closed when the subscription ends successfully (i.e. when the
// source of events is closed). It is also closed when Unsubscribe is called.
//
// The Unsubscribe method cancels the sending of events. You must call Unsubscribe in all
// cases to ensure that resources related to the subscription are released. It can be
// called any number of times.
type Subscription interface {
	Err() <-chan error // returns the error channel
	Unsubscribe()      // cancels sending of events, closing the error channel
}

// FeedOf implements one-to-many subscriptions where the carrier of events is a channel.
// Values sent to a Feed are delivered to all subscribed channels simultaneously.
//
// The zero value is ready to use.
type FeedOf[T any] struct {
	once      sync2.Once    // ensures that init only runs once
	sendLock  chan struct{} // sendLock has a one-element buffer and is empty when held.It protects sendCases.
	removeSub chan chan<- T // interrupts Send
	sendCases caseList      // the active set of select cases used by Send
	mu        sync.RWMutex  // The inbox holds newly subscribed channels until they are added to sendCases.
	inbox     caseList
}

func (f *FeedOf[T]) init() {
	f.removeSub = make(chan chan<- T)
	f.sendLock = make(chan struct{}, 1)
	f.sendLock <- struct{}{}
	f.sendCases = caseList{{Chan: reflect.ValueOf(f.removeSub), Dir: reflect.SelectRecv}}
}

// Subscribe adds a channel to the feed. Future sends will be delivered on the channel
// until the subscription is canceled.
//
// The channel should have ample buffer space to avoid blocking other subscribers. Slow
// subscribers are not dropped.
func (f *FeedOf[T]) Subscribe(channel chan<- T) Subscription {
	f.once.Do(f.init)

	chanval := reflect.ValueOf(channel)
	sub := &feedOfSub[T]{feed: f, channel: channel, err: make(chan error, 1)}

	// Add the select case to the inbox.
	// The next Send will add it to f.sendCases.
	f.mu.Lock()
	defer f.mu.Unlock()
	cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
	f.inbox = append(f.inbox, cas)
	return sub
}

func (f *FeedOf[T]) remove(sub *feedOfSub[T]) {
	// Delete from inbox first, which covers channels
	// that have not been added to f.sendCases yet.
	f.mu.Lock()
	index := f.inbox.find(sub.channel)
	if index != -1 {
		f.inbox = f.inbox.delete(index)
		f.mu.Unlock()
		return
	}
	f.mu.Unlock()

	select {
	case f.removeSub <- sub.channel:
		// Send will remove the channel from f.sendCases.
	case <-f.sendLock:
		// No Send is in progress, delete the channel now that we have the send lock.
		f.mu.Lock()
		f.sendCases = f.sendCases.delete(f.sendCases.find(sub.channel))
		f.sendLock <- struct{}{}
		f.mu.Unlock()
	}
}

// Send delivers to all subscribed channels simultaneously.
// It returns the number of subscribers that the value was sent to.
func (f *FeedOf[T]) Send(value T) (nsent int) {
	rvalue := reflect.ValueOf(value)

	f.once.Do(f.init)
	<-f.sendLock

	// Add new cases from the inbox after taking the send lock.
	f.mu.Lock()
	f.sendCases = append(f.sendCases, f.inbox...)
	f.inbox = nil
	f.mu.Unlock()

	// Set the sent value on all channels.
	for i := firstSubSendCase; i < len(f.sendCases); i++ {
		f.sendCases[i].Send = rvalue
	}

	// Send until all channels except removeSub have been chosen. 'cases' tracks a prefix
	// of sendCases. When a send succeeds, the corresponding case moves to the end of
	// 'cases' and it shrinks by one element.
	cases := f.sendCases
	for {
		// Fast path: try sending without blocking before adding to the select set.
		// This should usually succeed if subscribers are fast enough and have free
		// buffer space.
		for i := firstSubSendCase; i < len(cases); i++ {
			if cases[i].Chan.TrySend(rvalue) {
				nsent++
				cases = cases.deactivate(i)
				i--
			}
		}
		if len(cases) == firstSubSendCase {
			break
		}
		// Select on all the receivers, waiting for them to unblock.
		chosen, recv, _ := reflect.Select(cases)
		if chosen == 0 /* <-f.removeSub */ {
			index := f.sendCases.find(recv.Interface())
			f.sendCases = f.sendCases.delete(index)
			if index >= 0 && index < len(cases) {
				// Shrink 'cases' too because the removed case was still active.
				cases = f.sendCases[:len(cases)-1]
			}
		} else {
			cases = cases.deactivate(chosen)
			nsent++
		}
	}

	// Forget about the sent value and hand off the send lock.
	for i := firstSubSendCase; i < len(f.sendCases); i++ {
		f.sendCases[i].Send = reflect.Value{}
	}
	f.sendLock <- struct{}{}
	return nsent
}

type feedOfSub[T any] struct {
	feed    *FeedOf[T]
	channel chan<- T
	errOnce sync2.Once
	err     chan error
}

func (sub *feedOfSub[T]) Unsubscribe() {
	if sub == nil {
		return
	}
	sub.errOnce.Do(func() {
		sub.feed.remove(sub)
		close(sub.err)
	})
}

func (sub *feedOfSub[T]) Err() <-chan error {
	return sub.err
}
func (f *FeedOf[T]) Size() int {
	f.mu.RLock()
	ret := len(f.inbox) + len(f.sendCases)
	f.mu.RUnlock()
	return ret
}

type caseList []reflect.SelectCase

// find returns the index of a case containing the given channel.
func (cs caseList) find(channel interface{}) int {
	for i, cas := range cs {
		if cas.Chan.Interface() == channel {
			return i
		}
	}
	return -1
}

// delete removes the given case from cs.
func (cs caseList) delete(index int) caseList {
	return append(cs[:index], cs[index+1:]...)
}

// deactivate moves the case at index into the non-accessible portion of the cs slice.
func (cs caseList) deactivate(index int) caseList {
	last := len(cs) - 1
	cs[index], cs[last] = cs[last], cs[index]
	return cs[:last]
}

type AtomicUint32 uint32

func (i *AtomicUint32) Next() uint32            { return atomic.AddUint32((*uint32)(i), 1) }
func (i *AtomicUint32) Prev() uint32            { return atomic.AddUint32((*uint32)(i), ^uint32(0)) }
func (i *AtomicUint32) Load() uint32            { return atomic.LoadUint32((*uint32)(i)) }
func (i *AtomicUint32) Store(j uint32)          { atomic.StoreUint32((*uint32)(i), j) }
func (i *AtomicUint32) Add(delta uint32) uint32 { return atomic.AddUint32((*uint32)(i), delta) }
func (i *AtomicUint32) Subtract(delta uint32) uint32 {
	return atomic.AddUint32((*uint32)(i), ^(delta - 1))
}
func (i *AtomicUint32) UnmarshalXMLAttr(attr xml.Attr) error {
	val, err := strconv.ParseUint(attr.Value, 10, 32)
	i.Store(uint32(val))
	return err
}