Kirika/audio/queue/queue.go

package queue

import (
	"git.gammaspectra.live/S.O.N.G/Kirika/audio"
	"git.gammaspectra.live/S.O.N.G/Kirika/audio/filter"
	"log"
	"sync"
	"sync/atomic"
)

type QueueIdentifier int

type QueueEntry struct {
	Identifier     QueueIdentifier
	Source         audio.Source
	ReadSamples    atomic.Uint64
	cancel         chan bool
	StartCallback  func(q *Queue, entry *QueueEntry)
	EndCallback    func(q *Queue, entry *QueueEntry)
	RemoveCallback func(q *Queue, entry *QueueEntry)
}

type Queue struct {
	queue             []*QueueEntry
	output            audio.Source
	interrupt         chan bool
	interruptDepth    atomic.Int64
	closed            bool
	lock              sync.RWMutex
	wg                sync.WaitGroup
	identifierCounter QueueIdentifier
}

func NewQueue(format audio.SourceFormat, bitDepth, sampleRate, channels int) *Queue {
	if channels != 1 && channels != 2 {
		log.Panicf("not allowed channel number %d", channels)
	}

	q := &Queue{
		interrupt: make(chan bool, 1),
	}

	switch format {
	case audio.SourceFloat32:
		q.output = audio.NewSource[float32](bitDepth, sampleRate, channels)
		queueLoopStart[float32](q)
	case audio.SourceInt16:
		q.output = audio.NewSource[int16](bitDepth, sampleRate, channels)
		queueLoopStart[int16](q)
	case audio.SourceInt32:
		q.output = audio.NewSource[int32](bitDepth, sampleRate, channels)
		queueLoopStart[int32](q)
	default:
		log.Panicf("not found source format %d", int(format))
	}
	return q
}

func spliceHelper[T audio.AllowedSourceTypes](input audio.TypedSource[T]) (output audio.TypedSource[T], cancel chan bool) {
	cancel = make(chan bool, 1)
	output = audio.NewSource[T](input.GetBitDepth(), input.GetSampleRate(), input.GetChannels())

	bitDepth := input.GetBitDepth()

	sourceChannel := input.GetBlocks()
	go func() {
		defer output.Close()
	L1:
		for {
			select {
			case <-cancel:
				break L1
			case block, more := <-sourceChannel:
				if !more {
					//no more blocks!
					break L1
				} else {
					output.IngestNative(block, bitDepth)
				}
			}
		}

		input.Unlock()

		//sink remaining
		go audio.NewNullSink().Process(input)
	}()

	return
}

func (q *Queue) spliceSources(input audio.Source) (audio.Source, chan bool) {

	switch q.GetSource().GetFormat() {
	case audio.SourceFloat32:
		return spliceHelper(input.ToFloat32())
	case audio.SourceInt16:
		return spliceHelper(input.ToInt16())
	case audio.SourceInt32:
		return spliceHelper(input.ToInt32(input.GetBitDepth()))
	default:
		log.Panicf("not found source format %d", int(input.GetFormat()))
	}

	return nil, nil
}

func queueLoopStart[T audio.AllowedSourceTypes](q *Queue) {
	q.wg.Add(1)
	go func() {
		defer q.wg.Done()
		var current *QueueEntry
		var currentBlocks chan []T
	L1:
		for {
			q.lock.RLock()
			if q.closed {
				q.output.Close()
				break L1
			}
			if len(q.queue) == 0 { //no more entries, wait for interrupt
				q.lock.RUnlock()
				<-q.interrupt
				q.interruptDepth.Add(-1)
				continue
			}
			if current == nil || current.Identifier != q.queue[0].Identifier {
				current = q.queue[0]
				currentBlocks = current.Source.(audio.TypedSource[T]).GetBlocks()
			}
			q.lock.RUnlock()

		F1:
			for {
				select {
				case <-q.interrupt:
					q.interruptDepth.Add(-1)
					//force recheck
					break F1
				case block, more := <-currentBlocks:
					if !more {
						//no more blocks! skip
						if current.EndCallback != nil {
							current.EndCallback(q, current)
						}
						q.Remove(current.Identifier)
						break F1
					} else {
						if current.StartCallback != nil && current.ReadSamples.Load() == 0 && len(block) > 0 {
							current.StartCallback(q, current)
						}
						current.ReadSamples.Add(uint64(len(block) / current.Source.GetChannels()))
						q.output.(audio.TypedSource[T]).IngestNative(block, current.Source.GetBitDepth())
					}
				}
			}

		}

	}()
}

func (q *Queue) getFilterChain(source audio.Source) audio.Source {
	if q.GetChannels() == 1 {
		return filter.NewFilterChain(source, filter.MonoFilter{}, filter.NewResampleFilter(q.GetSampleRate(), filter.QualityFastest, 0), filter.SourceFormatFilter{
			Format: q.GetSource().GetFormat(),
		})
	} else {
		return filter.NewFilterChain(source, filter.StereoFilter{}, filter.NewResampleFilter(q.GetSampleRate(), filter.QualityFastest, 0), filter.SourceFormatFilter{
			Format: q.GetSource().GetFormat(),
		})
	}
}

func (q *Queue) AddHead(source audio.Source, startCallback func(q *Queue, entry *QueueEntry), endCallback func(q *Queue, entry *QueueEntry), removeCallback func(q *Queue, entry *QueueEntry)) (identifier QueueIdentifier) {
	q.lock.Lock()

	splicedOutput, cancel := q.spliceSources(source)
	identifier = q.identifierCounter
	if len(q.queue) > 0 {
		q.queue = append(q.queue[:1], append([]*QueueEntry{{
			Identifier:     identifier,
			Source:         q.getFilterChain(splicedOutput),
			cancel:         cancel,
			StartCallback:  startCallback,
			EndCallback:    endCallback,
			RemoveCallback: removeCallback,
		}}, q.queue[1:]...)...)
	} else {
		q.queue = append(q.queue, &QueueEntry{
			Identifier:     identifier,
			Source:         q.getFilterChain(splicedOutput),
			cancel:         cancel,
			StartCallback:  startCallback,
			EndCallback:    endCallback,
			RemoveCallback: removeCallback,
		})
	}
	q.identifierCounter++
	q.lock.Unlock()
	q.sendInterrupt()

	return
}

func (q *Queue) AddTail(source audio.Source, startCallback func(q *Queue, entry *QueueEntry), endCallback func(q *Queue, entry *QueueEntry), removeCallback func(q *Queue, entry *QueueEntry)) (identifier QueueIdentifier) {
	q.lock.Lock()

	splicedOutput, cancel := q.spliceSources(source)
	identifier = q.identifierCounter
	q.queue = append(q.queue, &QueueEntry{
		Identifier:     identifier,
		Source:         q.getFilterChain(splicedOutput),
		cancel:         cancel,
		StartCallback:  startCallback,
		EndCallback:    endCallback,
		RemoveCallback: removeCallback,
	})
	q.identifierCounter++
	q.lock.Unlock()
	q.sendInterrupt()

	return
}

func (q *Queue) IsClosed() bool {
	q.lock.RLock()
	defer q.lock.RUnlock()
	return q.closed
}

func (q *Queue) Remove(identifier QueueIdentifier) bool {
	q.lock.Lock()

	for i, e := range q.queue {
		if e.Identifier == identifier {
			q.sendInterrupt()
			e.cancel <- true

			e.Source.Unlock()

			go audio.NewNullSink().Process(e.Source)
			//delete entry
			q.queue = append(q.queue[:i], q.queue[i+1:]...)
			q.lock.Unlock()
			if e.RemoveCallback != nil {
				e.RemoveCallback(q, e)
			}
			return true
		}
	}
	q.lock.Unlock()

	return false

}

func (q *Queue) GetQueueHead() *QueueEntry {
	q.lock.RLock()
	defer q.lock.RUnlock()
	if len(q.queue) > 0 {
		return q.queue[0]
	}
	return nil
}

func (q *Queue) GetQueueTail() (index int, entry *QueueEntry) {
	q.lock.RLock()
	defer q.lock.RUnlock()
	if len(q.queue) > 0 {
		return len(q.queue) - 1, q.queue[len(q.queue)-1]
	}
	return 0, nil
}

func (q *Queue) GetQueueIndex(index int) *QueueEntry {
	q.lock.RLock()
	defer q.lock.RUnlock()
	if len(q.queue) > index {
		return q.queue[index]
	}
	return nil
}

func (q *Queue) GetQueueEntry(identifier QueueIdentifier) (index int, entry *QueueEntry) {
	q.lock.RLock()
	defer q.lock.RUnlock()
	for i, e := range q.queue {
		if e.Identifier == identifier {
			return i, e
		}
	}
	return -1, nil
}

func (q *Queue) GetQueueSize() int {
	q.lock.RLock()
	defer q.lock.RUnlock()

	return len(q.queue)
}

func (q *Queue) GetQueue() (entries []*QueueEntry) {
	q.lock.RLock()
	defer q.lock.RUnlock()

	entries = make([]*QueueEntry, len(q.queue))
	copy(entries, q.queue)

	return
}

func (q *Queue) GetSource() audio.Source {
	return q.output
}

func (q *Queue) GetSampleRate() int {
	return q.GetSource().GetSampleRate()
}

func (q *Queue) GetChannels() int {
	return q.GetSource().GetChannels()
}

func (q *Queue) sendInterrupt() {
	//TODO: maybe use len() on channel?
	if q.interruptDepth.Load() == 0 { //not waiting on interrupt
		q.interruptDepth.Add(1)
		q.interrupt <- true
	}
}

func (q *Queue) Close() {
	if !q.closed {
		q.closed = true
		q.sendInterrupt()
	}
}

func (q *Queue) Wait() {
	q.wg.Wait()
}