Experimental Kotlin Flow System in Go

flow.go

package main

import (
	"context"
	"fmt"
)

type Producer[T any] interface {
	Produce(ctx context.Context) (T, bool)
}

type sliceProducer[T any] struct {
	s []T
	i int
}

func (s *sliceProducer[T]) Produce(ctx context.Context) (v T, ok bool) {
	if s.i >= len(s.s) {
		return v, false
	}

	i := s.i
	s.i++
	return s.s[i], true
}

type unchangedFilter[T comparable] struct {
	p    Producer[T]
	prev *T
}

func (f *unchangedFilter[T]) Produce(ctx context.Context) (v T, ok bool) {
	for {
		v, ok = f.p.Produce(ctx)
		if !ok {
			return v, false
		}
		if (f.prev != nil) && (*f.prev == v) {
			continue
		}

		f.prev = &v
		return v, true
	}
}

type Flow[T any] struct {
	p Producer[T]
}

func NewFlow[T any](p Producer[T]) Flow[T] {
	return Flow[T]{p: p}
}

func FlowOf[T any](v ...T) Flow[T] {
	return Flow[T]{p: &sliceProducer[T]{s: v}}
}

func FilterUnchanged[T comparable](f Flow[T]) Flow[T] {
	return Flow[T]{p: &unchangedFilter[T]{p: f.p}}
}

func (flow Flow[T]) Collect(ctx context.Context, f func(T)) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			v, ok := flow.p.Produce(ctx)
			if !ok {
				return
			}
			f(v)
		}
	}
}

func main() {
	FilterUnchanged(FlowOf(3, 5, 5, 2, 2, 2)).Collect(context.Background(), func(v int) {
		fmt.Println(v)
	})
}