package fpgo

import (

"errors"

"sync"

"time"

)

// Queue Queue inspired by Collection utils

type Queue[T any] interface {

}

// Stack Stack inspired by Collection utils

type Stack[T any] interface {

Push(val T) error

Pop() (T, error)

}

var (

// ErrQueueIsEmpty Queue Is Empty

ErrQueueIsEmpty = errors.New("queue is empty")

// ErrQueueIsFull Queue Is Full

ErrQueueIsFull = errors.New("queue is full")

// ErrQueueIsClosed Queue Is Closed

ErrQueueIsClosed = errors.New("queue is closed")

// ErrQueueTakeTimeout Queue Take Timeout

ErrQueueTakeTimeout = errors.New("queue take timeout")

// ErrQueuePutTimeout Queue Put Timeout

ErrQueuePutTimeout = errors.New("queue put timeout")

// ErrStackIsEmpty Stack Is Empty

ErrStackIsEmpty = errors.New("stack is empty")

// ErrStackIsFull Stack Is Full

ErrStackIsFull = errors.New("stack is full")

)

// ConcurrentQueue & ConcurrentStack

// ConcurrentQueue ConcurrentQueue inspired by Collection utils

type ConcurrentQueue[T any] struct {

lock sync.RWMutex

queue Queue[T]

}

// NewConcurrentQueue New ConcurrentQueue instance from a Queue[T]

func NewConcurrentQueue[T any](queue Queue[T]) *ConcurrentQueue[T] {

}

// Put Put the T val(probably blocking)

func (q *ConcurrentQueue[T]) Put(val T) error {

q.lock.Lock()

defer q.lock.Unlock()

return q.queue.Put(val)

}

// Take Take the T val(probably blocking)

func (q *ConcurrentQueue[T]) Take() (T, error) {

q.lock.RLock()

defer q.lock.RUnlock()

return q.queue.Take()

}

// Offer Offer the T val(non-blocking)

func (q *ConcurrentQueue[T]) Offer(val T) error {

q.lock.Lock()

defer q.lock.Unlock()

return q.queue.Offer(val)

}

// Poll Poll the T val(non-blocking)

func (q *ConcurrentQueue[T]) Poll() (T, error) {

q.lock.RLock()

defer q.lock.RUnlock()

return q.queue.Poll()

}

// ConcurrentStack

// ConcurrentStack ConcurrentStack inspired by Collection utils

type ConcurrentStack[T any] struct {

lock sync.RWMutex

stack Stack[T]

}

// NewConcurrentStack New ConcurrentStack instance from a Stack[T]

func NewConcurrentStack[T any](stack Stack[T]) *ConcurrentStack[T] {

}

// Put Put the T val(probably blocking)

func (q *ConcurrentStack[T]) Push(val T) error {

q.lock.Lock()

defer q.lock.Unlock()

return q.stack.Push(val)

}

// Take Take the T val(probably blocking)

func (q *ConcurrentStack[T]) Pop() (T, error) {

q.lock.RLock()

defer q.lock.RUnlock()

return q.stack.Pop()

}

// ChannelQueue

// ChannelQueue ChannelQueue inspired by Collection utils

type ChannelQueue[T any] chan T

// NewChannelQueue New ChannelQueue instance with capacity

func NewChannelQueue[T any](capacity int) ChannelQueue[T] {

return make(ChannelQueue[T], capacity)

}

// Put Put the T val(blocking)

func (q ChannelQueue[T]) Put(val T) error {

q <- val

return nil

}

// PutWithTimeout Put the T val(blocking), with timeout

func (q ChannelQueue[T]) PutWithTimeout(val T, timeout time.Duration) error {

select {

case q <- val:

return nil

case <-time.After(timeout):

return ErrQueuePutTimeout

}

}

// Take Take the T val(blocking)

func (q ChannelQueue[T]) Take() (T, error) {

val, ok := <-q

if !ok {

return *new(T), ErrQueueIsClosed

}

return val, nil

}

// TakeWithTimeout Take the T val(blocking), with timeout

func (q ChannelQueue[T]) TakeWithTimeout(timeout time.Duration) (T, error) {

select {

case val, ok := <-q:

if !ok {

return *new(T), ErrQueueIsClosed

}

return val, nil

case <-time.After(timeout):

return *new(T), ErrQueueTakeTimeout

}

}

// Offer Offer the T val(non-blocking)

func (q ChannelQueue[T]) Offer(val T) error {

select {

case q <- val:

return nil

default:

return ErrQueueIsFull

}

}

// Poll Poll the T val(non-blocking)

func (q ChannelQueue[T]) Poll() (T, error) {

select {

case val := <-q:

return val, nil

default:

return *new(T), ErrQueueIsEmpty

}

}

// LinkedList & DoublyLinkedList

// LinkedListItem LinkedListItem inspired by Collection utils

type LinkedListItem[T any] struct {

}

// Count Count items

func (listItem *LinkedListItem[T]) Count() int {

count := 1

first := listItem

for first.Next != nil {

count++

first = first.Next

}

return count

}

// Last Get the Last one

func (listItem *LinkedListItem[T]) Last() *LinkedListItem[T] {

last := listItem

for last.Next != nil {

last = last.Next

}

return last

}

// AddLast Add the input item to the last

func (listItem *LinkedListItem[T]) AddLast(input *LinkedListItem[T]) *LinkedListItem[T] {

last := listItem.Last()

last.Next = input

return last

}

// DoublyListItem DoublyListItem inspired by Collection utils

type DoublyListItem[T any] struct {

}

// Count Count items

func (listItem *DoublyListItem[T]) Count() int {

count := 1

first := listItem.First()

for first.Next != nil {

count++

first = first.Next

}

return count

}

// Last Get the Last one

func (listItem *DoublyListItem[T]) Last() *DoublyListItem[T] {

last := listItem

for last.Next != nil {

last = last.Next

}

return last

}

// First Get the First one

func (listItem *DoublyListItem[T]) First() *DoublyListItem[T] {

first := listItem

for first.Prev != nil {

first = first.Prev

}

return first

}

// AddLast Add the input item to the last

func (listItem *DoublyListItem[T]) AddLast(input *DoublyListItem[T]) *DoublyListItem[T] {

last := listItem.Last()

first := input.First()

last.Next = first

first.Prev = last

return last

}

// AddFirst Add the input item to the first position

func (listItem *DoublyListItem[T]) AddFirst(input *DoublyListItem[T]) *DoublyListItem[T] {

last := input.Last()

first := listItem.First()

last.Next = first

first.Prev = last

return first

}

// LinkedListQueue LinkedListQueue inspired by Collection utils

type LinkedListQueue[T any] struct {

}

// NewLinkedListQueue New LinkedListQueue instance

func NewLinkedListQueue[T any]() *LinkedListQueue[T] {

}

func (q *LinkedListQueue[T]) putAllIntoPool(first *DoublyListItem[T]) {

for first != nil {

node := first

first = first.Next

node.Val = nil

node.Prev = nil

node.Next = nil

q.nodeGCPool.Put(node)

}

}

// Count Count Items

func (q *LinkedListQueue[T]) Count() int {

return q.count

}

// ClearNodePool Clear cached LinkedListItem nodes in nodePool

func (q *LinkedListQueue[T]) ClearNodePool() {

q.putAllIntoPool(q.nodePoolFirst)

q.nodeCount = 0

q.nodePoolFirst = nil

}

// KeepNodePoolCount Decrease/Increase LinkedListItem nodes to n items

func (q *LinkedListQueue[T]) KeepNodePoolCount(n int) {

if n <= 0 {

q.ClearNodePool()

return

}

q.nodeCount = n

n--

last := q.nodePoolFirst

if last == nil {

last = q.nodeGCPool.Get().(*DoublyListItem[T])

q.nodePoolFirst = last

}

for n > 0 {

n--

if last.Next == nil {

last.Next = q.nodeGCPool.Get().(*DoublyListItem[T])

}

last = last.Next

}

q.putAllIntoPool(last.Next)

last.Next = nil

}

// CLear Clear all data

func (q *LinkedListQueue[T]) Clear() {

q.nodePoolFirst = q.first

q.nodeCount = q.count

node := q.nodePoolFirst

for node != nil {

node.Val = nil

node.Prev = nil

node = node.Next

}

q.first = nil

q.last = nil

q.count = 0

}

// Put Put the T val to the last position(no-blocking)

func (q *LinkedListQueue[T]) Put(val T) error {

return q.Offer(val)

}

// Take Take the T val from the first position(no-blocking)

func (q *LinkedListQueue[T]) Take() (T, error) {

return q.Poll()

}

// Offer Offer the T val to the last position(non-blocking)

func (q *LinkedListQueue[T]) Offer(val T) error {

// Try get from pool or new one

node := q.generateNode()

node.Val = &val

q.count++

if q.first == nil {

q.first = node

}

last := q.last

if last != nil {

last.Next = node

node.Prev = last

}

q.last = node

return nil

}

// Poll Poll the T val from the first position(non-blocking)

func (q *LinkedListQueue[T]) Poll() (T, error) {

return q.Shift()

}

// Peek Peek the T val from the first position without removing it (non-blocking)

func (q *LinkedListQueue[T]) Peek() (T, error) {

node := q.first

if node == nil {

return *new(T), ErrQueueIsEmpty

}

return *node.Val, nil

}

// Shift Shift the T val from the first position (non-blocking)

func (q *LinkedListQueue[T]) Shift() (T, error) {

node := q.first

if node == nil {

return *new(T), ErrQueueIsEmpty

}

// Remove the first item

q.count--

q.first = node.Next

if q.first == nil {

q.last = nil

}

val := *node.Val

q.recycleNode(node)

return val, nil

}

// Unshift Unshift the T val to the first position(non-blocking)

func (q *LinkedListQueue[T]) Unshift(val T) error {

// Try get from pool or new one

node := q.generateNode()

node.Val = &val

q.count++

if q.last == nil {

q.last = node

}

first := q.first

q.first = node

node.Next = first

if first != nil {

first.Prev = node

}

return nil

}

// Pop Pop the data from the last position(non-blocking)

func (q *LinkedListQueue[T]) Pop() (T, error) {

node := q.last

if node == nil {

return *new(T), ErrStackIsEmpty

}

// Remove the last item

q.count--

q.last = node.Prev

if q.last == nil {

q.first = nil

}

val := *node.Val

q.recycleNode(node)

return val, nil

}

// Push Push the data to the last position(non-blocking)

func (q *LinkedListQueue[T]) Push(val T) error {

// return q.Unshift(val)

return q.Offer(val)

}

func (q *LinkedListQueue[T]) generateNode() *DoublyListItem[T] {

node := q.nodePoolFirst

if node == nil {

node = q.nodeGCPool.Get().(*DoublyListItem[T])

} else {

q.nodeCount--

q.nodePoolFirst = node.Next

node.Next = nil

node.Prev = nil

}

return node

}

func (q *LinkedListQueue[T]) recycleNode(node *DoublyListItem[T]) {

if node == nil {

return

}

// Recycle

q.nodeCount++

node.Val = nil

node.Next = q.nodePoolFirst

node.Prev = nil

q.nodePoolFirst = node

}

// BufferedChannelQueue BlockingQueue with ChannelQueue & scalable pool, inspired by Collection utils

type BufferedChannelQueue[T any] struct {

}

// NewBufferedChannelQueue New BufferedChannelQueue instance from a Queue[T]

func NewBufferedChannelQueue[T any](channelCapacity int, bufferSizeMaximum int, nodeHookPoolSize int) *BufferedChannelQueue[T] {

}

func (q *BufferedChannelQueue[T]) freeNodePool() {

for range q.freeNodeWorkerCh {

time.Sleep(q.freeNodeHookPoolIntervalDuration)

if q.isClosed.Get() {

break

}

q.lock.Lock()

if q.pool.nodeCount > q.nodeHookPoolSize {

q.pool.KeepNodePoolCount(q.nodeHookPoolSize)

}

q.lock.Unlock()

}

}

func (q *BufferedChannelQueue[T]) loadFromPool() {

for range q.loadWorkerCh {

if q.isClosed.Get() {

break

}

q.lock.Lock()

var val T

var pollErr, offerErr error

for q.pool.Count() > 0 {

// Try poll from the pool

val, pollErr = q.pool.Poll()

if pollErr != nil {

break

}

offerErr = q.blockingQueue.Offer(val)

// If failed, unshift it back

if offerErr != nil {

q.pool.Unshift(val)

break

}

}

q.lock.Unlock()

time.Sleep(q.loadFromPoolDuration)

}

}

func (q *BufferedChannelQueue[T]) notifyWorkers() {

q.loadWorkerCh.Offer(1)

q.freeNodeWorkerCh.Offer(1)

}

// SetBufferSizeMaximum Set MaximumBufferSize(maximum number of buffered items outside the ChannelQueue)

func (q *BufferedChannelQueue[T]) SetBufferSizeMaximum(size int) *BufferedChannelQueue[T] {

q.bufferSizeMaximum = size

return q

}

// SetNodeHookPoolSize Set nodeHookPoolSize(the buffering node hooks ideal size)

func (q *BufferedChannelQueue[T]) SetNodeHookPoolSize(size int) *BufferedChannelQueue[T] {

q.nodeHookPoolSize = size

return q

}

// SetLoadFromPoolDuration Set loadFromPoolDuration(the interval to take buffered items into the ChannelQueue)

func (q *BufferedChannelQueue[T]) SetLoadFromPoolDuration(duration time.Duration) *BufferedChannelQueue[T] {

q.loadFromPoolDuration = duration

return q

}

// SetFreeNodeHookPoolIntervalDuration Set freeNodeHookPoolIntervalDuration(the interval to clear buffering node hooks down to nodeHookPoolSize)

func (q *BufferedChannelQueue[T]) SetFreeNodeHookPoolIntervalDuration(duration time.Duration) *BufferedChannelQueue[T] {

q.freeNodeHookPoolIntervalDuration = duration

return q

}

// GetBufferSizeMaximum Get MaximumBufferSize(maximum number of buffered items outside the ChannelQueue)

func (q *BufferedChannelQueue[T]) GetBufferSizeMaximum() int {

return q.bufferSizeMaximum

}

// GetNodeHookPoolSize Get nodeHookPoolSize(the buffering node hooks ideal size)

func (q *BufferedChannelQueue[T]) GetNodeHookPoolSize() int {

return q.nodeHookPoolSize

}

// GetLoadFromPoolDuration Get loadFromPoolDuration(the interval to take buffered items into the ChannelQueue)

func (q *BufferedChannelQueue[T]) GetLoadFromPoolDuration() time.Duration {

return q.loadFromPoolDuration

}

// GetFreeNodeHookPoolIntervalDuration Get freeNodeHookPoolIntervalDuration(the interval to clear buffering node hooks down to nodeHookPoolSize)

func (q *BufferedChannelQueue[T]) GetFreeNodeHookPoolIntervalDuration() time.Duration {

return q.freeNodeHookPoolIntervalDuration

}

// GetChannel Get Channel(for Selecting channels usages)

func (q *BufferedChannelQueue[T]) GetChannel() chan T {

q.notifyWorkers()

return q.blockingQueue

}

// Count Count items

func (q *BufferedChannelQueue[T]) Count() int {

if q.isClosed.Get() {

return 0

}

q.lock.RLock()

defer q.lock.RUnlock()

return len(q.blockingQueue) + q.pool.Count()

}

// IsClosed Is the BufferedChannelQueue closed

func (q *BufferedChannelQueue[T]) IsClosed() bool {

return q.isClosed.Get()

}

// Close Close the BufferedChannelQueue

func (q *BufferedChannelQueue[T]) Close() {

q.lock.Lock()

defer q.lock.Unlock()

q.isClosed.Set(true)

close(q.loadWorkerCh)

close(q.blockingQueue)

}

// Put Put the T val(non-blocking)

func (q *BufferedChannelQueue[T]) Put(val T) error {

// q.lock.Lock()

// defer q.lock.Unlock()

//

// if q.isClosed.Get() {

// return ErrQueueIsClosed

// }

//

// q.lock.Lock()

// poolCount := q.pool.Count()

//

// // If appearing nothing in the pool

// if poolCount == 0 {

// defer q.lock.Unlock()

// // Try channel

// return q.blockingQueue.Put(val)

// }

// q.lock.Unlock()

return q.Offer(val)

}

// // PutWithTimeout Put the T val(blocking), with timeout

// func (q BufferedChannelQueue[T]) PutWithTimeout(val T, timeout time.Duration) error {

// // q.lock.Lock()

// // defer q.lock.Unlock()

//

// if q.isClosed.Get() {

// return ErrQueueIsClosed

// }

//

// return q.blockingQueue.PutWithTimeout(val, timeout)

// }

// Take Take the T val(blocking)

func (q *BufferedChannelQueue[T]) Take() (T, error) {

if q.isClosed.Get() {

return *new(T), ErrQueueIsClosed

}

q.notifyWorkers()

return q.blockingQueue.Take()

}

// TakeWithTimeout Take the T val(blocking), with timeout

func (q *BufferedChannelQueue[T]) TakeWithTimeout(timeout time.Duration) (T, error) {

if q.isClosed.Get() {

return *new(T), ErrQueueIsClosed

}

q.notifyWorkers()

return q.blockingQueue.TakeWithTimeout(timeout)

}

// Offer Offer the T val(non-blocking)

func (q *BufferedChannelQueue[T]) Offer(val T) error {

q.lock.Lock()

defer q.lock.Unlock()

if q.isClosed.Get() {

return ErrQueueIsClosed

}

poolCount := q.pool.Count()

// If appearing nothing in the pool

if poolCount == 0 {

// Try channel

err := q.blockingQueue.Offer(val)

if err == nil {

// Success

return nil

} else if err == ErrQueueIsFull {

// Do nothing and let pool.Offer(val)

} else {

// Other

return err

}

}

// Before +1: >=, After +1: >

if poolCount >= q.bufferSizeMaximum {

return ErrQueueIsFull

}

q.pool.Offer(val)

q.loadWorkerCh.Offer(1)

return nil

}

// Poll Poll the T val(non-blocking)

func (q *BufferedChannelQueue[T]) Poll() (T, error) {

if q.isClosed.Get() {

return *new(T), ErrQueueIsClosed

}

q.notifyWorkers()

return q.blockingQueue.Poll()

}