package transport

import (
	"context"

	"goflink/core"
)

type LocalReceiveChannel[T any] struct {
	recordCh <-chan core.StreamRecord[T]
}

func NewLocalReceiveChannel[T any](ch <-chan core.StreamRecord[T]) *LocalReceiveChannel[T] {
	return &LocalReceiveChannel[T]{
		recordCh: ch,
	}
}

func (c *LocalReceiveChannel[T]) Receive(ctx context.Context) (core.StreamRecord[T], bool) {
	select {
	case val, ok := <-c.recordCh:
		return val, ok
	case <-ctx.Done():
		var zero core.StreamRecord[T]
		return zero, false
	}
}

func (c *LocalReceiveChannel[T]) Close() error {
	// The emitter will close
	return nil
}

type LocalEmitChannel[T any] struct {
	recordCh chan<- core.StreamRecord[T]
}

func NewLocalEmitChannel[T any](ch chan<- core.StreamRecord[T]) *LocalEmitChannel[T] {
	return &LocalEmitChannel[T]{
		recordCh: ch,
	}
}

func (l *LocalEmitChannel[T]) Emit(ctx context.Context, c core.StreamRecord[T]) error {
	select {
	case l.recordCh <- c:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (l *LocalEmitChannel[T]) Close() error {
	close(l.recordCh)
	return nil
}