collections/collection.go

package collections

import (
	"reflect"
	"sort"
	"sync"
	"sync/atomic"

	"pkg.icikowski.pl/collections/functions"
)

// Collection represents the collection of data.
//
// By default, the collection uses parallel implementation of operators.
type Collection[T any] struct {
	data     []T
	parallel bool
}

// Count counts values in the collection.
func (s *Collection[T]) Count() int {
	return len(s.data)
}

// Epty determines whether the collection is empty.
func (s *Collection[T]) Empty() bool {
	return len(s.data) == 0
}

// Parallel sets the collection to use parallel implementation of operators.
func (s *Collection[T]) Parallel() *Collection[T] {
	s.parallel = true
	return s
}

// Sequential sets the collection to use sequential implementation of operators.
func (s *Collection[T]) Sequential() *Collection[T] {
	s.parallel = false
	return s
}

func (s *Collection[T]) filterParallel(p functions.Predicate[T]) *Collection[T] {
	processed := []T{}
	mux := sync.Mutex{}
	wg := sync.WaitGroup{}

	wg.Add(len(s.data))
	for _, e := range s.data {
		e := e
		go func() {
			defer wg.Done()
			if p(e) {
				mux.Lock()
				defer mux.Unlock()
				processed = append(processed, e)
			}
		}()
	}
	wg.Wait()

	s.data = processed
	return s
}

func (s *Collection[T]) filterSequential(p functions.Predicate[T]) *Collection[T] {
	processed := []T{}
	for _, e := range s.data {
		if p(e) {
			processed = append(processed, e)
		}
	}
	s.data = processed
	return s
}

// Filter filters the collection using given [functions.Predicate].
func (s *Collection[T]) Filter(p functions.Predicate[T]) *Collection[T] {
	if s.parallel {
		return s.filterParallel(p)
	}
	return s.filterSequential(p)
}

func (s *Collection[T]) allMatchParallel(p functions.Predicate[T]) bool {
	wg := sync.WaitGroup{}
	state := atomic.Bool{}

	wg.Add(len(s.data))
	state.Store(true)

	for _, e := range s.data {
		if !state.Load() {
			break
		}
		e := e
		go func() {
			defer wg.Done()
			if state.Load() && !p(e) {
				state.Store(false)
			}
		}()
	}

	return state.Load()
}

func (s *Collection[T]) allMatchSequential(p functions.Predicate[T]) bool {
	for _, e := range s.data {
		if !p(e) {
			return false
		}
	}
	return true
}

// AllMatch checks whether all elements in collection match given [functions.Predicate].
func (s *Collection[T]) AllMatch(p functions.Predicate[T]) bool {
	if s.parallel {
		return s.allMatchParallel(p)
	}
	return s.allMatchSequential(p)
}

func (s *Collection[T]) anyMatchParallel(p functions.Predicate[T]) bool {
	wg := sync.WaitGroup{}
	state := atomic.Bool{}

	wg.Add(len(s.data))
	state.Store(false)

	for _, e := range s.data {
		if state.Load() {
			break
		}
		e := e
		go func() {
			defer wg.Done()
			if state.Load() && p(e) {
				state.Store(true)
			}
		}()
	}

	return state.Load()
}

func (s *Collection[T]) anyMatchSequential(p functions.Predicate[T]) bool {
	for _, e := range s.data {
		if p(e) {
			return true
		}
	}
	return false
}

// AnyMatch checks whether any elements in collection match given [functions.Predicate].
func (s *Collection[T]) AnyMatch(p functions.Predicate[T]) bool {
	if s.parallel {
		return s.anyMatchParallel(p)
	}
	return s.anyMatchSequential(p)
}

func (s *Collection[T]) noneMatchParallel(p functions.Predicate[T]) bool {
	wg := sync.WaitGroup{}
	state := atomic.Bool{}

	wg.Add(len(s.data))
	state.Store(true)

	for _, e := range s.data {
		if !state.Load() {
			break
		}
		e := e
		go func() {
			defer wg.Done()
			if state.Load() && p(e) {
				state.Store(false)
			}
		}()
	}

	return state.Load()
}

func (s *Collection[T]) noneMatchSequential(p functions.Predicate[T]) bool {
	for _, e := range s.data {
		if p(e) {
			return false
		}
	}
	return true
}

// NoneMatch checks whether no elements in collection match given [functions.Predicate].
func (s *Collection[T]) NoneMatch(p functions.Predicate[T]) bool {
	if s.parallel {
		return s.noneMatchParallel(p)
	}
	return s.noneMatchSequential(p)
}

// Sorted sorts the collection using given [functions.Comparator].
func (s *Collection[T]) Sorted(c functions.Comparator[T]) *Collection[T] {
	sort.SliceStable(s.data, func(i, j int) bool {
		return c(s.data[i], s.data[j])
	})
	return s
}

func (s *Collection[T]) peekParallel(c functions.Consumer[T]) *Collection[T] {
	wg := sync.WaitGroup{}
	wg.Add(len(s.data))

	for _, e := range s.data {
		e := e
		go func() {
			defer wg.Done()
			c(e)
		}()
	}

	wg.Wait()
	return s
}

func (s *Collection[T]) peekSequential(c functions.Consumer[T]) *Collection[T] {
	for _, e := range s.data {
		c(e)
	}
	return s
}

// Peek executes given [function.Consumer] on every value in collections.
func (s *Collection[T]) Peek(c functions.Consumer[T]) *Collection[T] {
	if s.parallel {
		return s.peekParallel(c)
	}
	return s.peekSequential(c)
}

func (s *Collection[T]) transformParallel(u functions.UnaryOperator[T]) *Collection[T] {
	processed := make([]T, len(s.data))
	wg := sync.WaitGroup{}

	wg.Add(len(s.data))
	for i, e := range s.data {
		i, e := i, e
		go func() {
			defer wg.Done()
			processed[i] = u(e)
		}()
	}

	wg.Wait()
	s.data = processed
	return s
}

func (s *Collection[T]) transformSequential(u functions.UnaryOperator[T]) *Collection[T] {
	processed := []T{}
	for _, e := range s.data {
		processed = append(processed, u(e))
	}
	s.data = processed
	return s
}

// Transform transforms all values in collection using given [functions.UnaryOperator].
func (s *Collection[T]) Transform(u functions.UnaryOperator[T]) *Collection[T] {
	if s.parallel {
		return s.transformParallel(u)
	}
	return s.transformSequential(u)
}

// Reduce reduces values in given collection using given [functions.BinaryOperator].
func (s *Collection[T]) Reduce(b functions.BinaryOperator[T]) T {
	processed := *new(T)
	if len(s.data) == 0 {
		processed = s.data[0]
	}

	for _, e := range s.data[1:] {
		processed = b(processed, e)
	}
	return processed
}

// Limit limits the collection to given number of values.
func (s *Collection[T]) Limit(n int) *Collection[T] {
	if len(s.data) < n {
		return s
	}

	if n < 0 {
		n = 0
	}

	s.data = s.data[:n]
	return s
}

// Skip skips given number of values in the collection.
func (s *Collection[T]) Skip(n int) *Collection[T] {
	if len(s.data) == 0 {
		return s
	}

	if n > len(s.data) {
		n = len(s.data)
	}

	s.data = s.data[n:]
	return s
}

// FindFirst returns [Optional] with first item of the collection.
func (s *Collection[T]) FindFirst() *Optional[T] {
	e, present := *new(T), false
	if len(s.data) != 0 {
		e, present = s.data[0], true
	}
	return &Optional[T]{e, present}
}

// FindFirst returns [Optional] with last item of the collection.
func (s *Collection[T]) FindLast() *Optional[T] {
	e, present := *new(T), false
	if len(s.data) != 0 {
		e, present = s.data[len(s.data)-1], true
	}
	return &Optional[T]{e, present}
}

// Min returns the lowest value from the collection using given [functions.Comparator].
func (s *Collection[T]) Min(c functions.Comparator[T]) *Optional[T] {
	return s.Sorted(c).FindFirst()
}

// Max returns the highest value from the collection using given [functions.Comparator].
func (s *Collection[T]) Max(c functions.Comparator[T]) *Optional[T] {
	return s.Sorted(c).FindLast()
}

// Distinct ensures that all elements in the collecion are unique.
func (s *Collection[T]) Distinct() *Collection[T] {
	processed := []T{}
	for i := 0; i < len(s.data); i++ {
		hasCopy := false
		for j := i + 1; j < len(s.data); j++ {
			if reflect.DeepEqual(s.data[i], s.data[j]) {
				hasCopy = true
				break
			}
		}
		if !hasCopy {
			processed = append(processed, s.data[i])
		}
	}
	s.data = processed
	return s
}

// MapCollection maps collection of values of type T to collection of values of type U using given [functions.Function].
func MapCollection[T, U any](src *Collection[T], mapper functions.Function[T, U]) *Collection[U] {
	data := []U{}
	for _, e := range src.data {
		data = append(data, mapper(e))
	}
	return &Collection[U]{data, src.parallel}
}

// Collect returns all items from collection as a slice.
func (s *Collection[T]) Collect() []T {
	return s.data
}