package transport

import "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() (core.StreamRecord[T], bool) {
	val, ok := <-c.recordCh
	return val, ok
}

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

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

func (l LocalEmitChannel[T]) Emit(c core.StreamRecord[T]) error {
	l.recordCh <- c
	return nil
}

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