func (timedQueueElement *QueueElement) Cancel() {
	timedQueueElement.timedQueue.heapMutex.Lock()
	defer timedQueueElement.timedQueue.heapMutex.Unlock()
	timedQueueElement.timedQueue.removeElement(timedQueueElement)
	close(timedQueueElement.cancel)
}
Cancel removed the given element from the queue and cancels its execution.
package timedqueue
import (
	"container/heap"
	"context"
	"sync"
	"time"
	"github.com/iotaledger/hive.go/bitmask"
)
type TimedQueue struct {
	heap      timedHeap
	heapMutex sync.RWMutex
	waitCond *sync.Cond
	maxSize int
	ctx           context.Context
	ctxCancel     context.CancelFunc
	isShutdown    bool
	shutdownFlags ShutdownFlag
	shutdownMutex sync.Mutex
}
func New(opts ...Option) (queue *TimedQueue) {
	ctx, ctxCancel := context.WithCancel(context.Background())
	queue = &TimedQueue{
		ctx:       ctx,
		ctxCancel: ctxCancel,
	}
	queue.waitCond = sync.NewCond(&queue.heapMutex)
	for _, opt := range opts {
		opt(queue)
	}
	return
}
func (t *TimedQueue) Add(value interface{}, scheduledTime time.Time) (addedElement *QueueElement) {
	if value == nil {
		panic("<nil> must not be added to the queue")
	}
	if t.IsShutdown() {
		if t.shutdownFlags.HasBits(PanicOnModificationsAfterShutdown) {
			panic("tried to modify a shutdown TimedQueue")
		}
		return
	}
	t.heapMutex.Lock()
	addedElement = &QueueElement{
		timedQueue:    t,
		Value:         value,
		ScheduledTime: scheduledTime,
		cancel:        make(chan byte),
		index:         0,
	}
	heap.Push(&t.heap, addedElement)
	if t.maxSize > 0 {
		if size := t.heap.Len(); size > t.maxSize {
			heap.Remove(&t.heap, size-1)
		}
	}
	t.heapMutex.Unlock()
	t.waitCond.Signal()
	return
}
func (t *TimedQueue) Size() int {
	t.heapMutex.RLock()
	defer t.heapMutex.RUnlock()
	return len(t.heap)
}
func (t *TimedQueue) Shutdown(optionalShutdownFlags ...ShutdownFlag) {
	t.shutdownMutex.Lock()
	if t.isShutdown {
		defer t.shutdownMutex.Unlock()
		if t.shutdownFlags.HasBits(PanicOnModificationsAfterShutdown) {
			panic("tried to shutdown and already shutdown TimedQueue")
		}
		return
	}
	t.isShutdown = true
	for _, shutdownFlag := range optionalShutdownFlags {
		t.shutdownFlags |= shutdownFlag
	}
	t.shutdownMutex.Unlock()
	t.ctxCancel()
	t.heapMutex.Lock()
	switch queuedElementsCount := len(t.heap); queuedElementsCount {
	case 0:
		t.waitCond.Broadcast()
	default:
		if t.shutdownFlags.HasBits(CancelPendingElements) {
			for i := 0; i < queuedElementsCount; i++ {
				heap.Remove(&t.heap, 0)
			}
		}
	}
	t.heapMutex.Unlock()
}
func (t *TimedQueue) IsShutdown() bool {
	t.shutdownMutex.Lock()
	defer t.shutdownMutex.Unlock()
	return t.isShutdown
}
func (t *TimedQueue) Poll(waitIfEmpty bool) interface{} {
	for {
		t.heapMutex.Lock()
		for len(t.heap) == 0 {
			if !waitIfEmpty || t.IsShutdown() {
				t.heapMutex.Unlock()
				return nil
			}
			t.waitCond.Wait()
		}
		polledElement := heap.Remove(&t.heap, 0).(*QueueElement)
		t.heapMutex.Unlock()
		select {
		case <-t.ctx.Done():
			if t.shutdownFlags.HasBits(CancelPendingElements) {
				return nil
			}
			if t.shutdownFlags.HasBits(IgnorePendingTimeouts) {
				return polledElement.Value
			}
			select {
			case <-polledElement.cancel:
				continue
			case <-time.After(time.Until(polledElement.ScheduledTime)):
				return polledElement.Value
			}
		case <-polledElement.cancel:
			continue
		case <-time.After(time.Until(polledElement.ScheduledTime)):
			return polledElement.Value
		}
	}
}
func (t *TimedQueue) removeElement(element *QueueElement) {
	if element.index == -1 {
		return
	}
	heap.Remove(&t.heap, element.index)
}
type QueueElement struct {
	Value interface{}
	ScheduledTime time.Time
	timedQueue *TimedQueue
	cancel     chan byte
	index      int
}