kern/v4/d0/defs/safe_chan/safe_chan.go

// package safe_chan -- умный безопасный канал
package safe_chan

import (
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/defs/def_err"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/defs/def_spec_base"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/defs/safe_bool"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/defs/safe_mutex"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/defs/table/owner"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/helpers"
	"gitp78su.ipnodns.ru/svi/kern/v4/d0/result"
)

// SafeChanParam -- параметры умного безопасного канала.
type SafeChanParam[T any] struct {
	Limit_   def_spec_base.Num // лимит на размер канала, не может быть пустым
	OnWrite_ func(T)           // Обратный вызов при записи в канал
	OnRead_  func(T)           // Обратный вызов при чтении из канала
	OnClose_ func()            // Обратный вызов при закрытии канала
	OnLimit_ func()            // Обратный вызов при достижении лимита
	Owner_   *owner.LOwner     // Владелец канала
}

// SelfCheck -- проверка корректности правильности параметров умного канала.
func (sf *SafeChanParam[T]) SelfCheck() {
	helpers.If(sf.Limit_ <= 0).
		Hassert("SafeChanParam[T].SelfCheck(): limit=%v, канал должен иметь положительный лимит",
			sf.Limit_)
	helpers.If(sf.Owner_ == nil).Hassert("SafeChanParam[T].SelfCheck(): Owner_==nil")
}

// SafeChan -- умный безопасный канал
//
// Канал является резиновым, но лимит можно поставить сверху.
// При необходимости лимит ёмкости можно поднять, но не снизить.
// Умный канал можно безопасно закрывать многократно.
// Умный канал можно использовать неблокирующим способом.
// Умный канал точно знает своё состояние (длина, закрыт и т.п.)
// При передаче инстанса канала -- из него можно только читать.
// Умный канал строго типизирован и его тип видно из сигнатуры.
// Кроме того, можно повесить хуки на события записи, чтения и закрытия
// (например, для целей валидации).
type SafeChan[T any] struct {
	*SafeChanParam[T]
	block    *safe_mutex.MSafeMutex
	isClosed *safe_bool.MSafeBool
	lstMsg   []T
}

var msg2 = def_spec_base.Txt("NewKChan: param==nil")

// MutSafeChan -- создаёт новый безопасный умный канал.
func MutSafeChan[T any](param *SafeChanParam[T]) *SafeChan[T] {
	helpers.If(param == nil).Hassert(msg2)
	param.SelfCheck()
	sf := &SafeChan[T]{
		SafeChanParam: param,
		block:         safe_mutex.MutSafeMutex(),
		isClosed:      safe_bool.MutSafeBool(param.Owner_, true),
	}
	return sf
}

// Read -- возвращает первый элемент очереди (если есть).
func (sf *SafeChan[T]) Read() *result.Result[T] {
	sf.block.Lock()
	defer sf.block.Unlock()
	if len(sf.lstMsg) == 0 {
		err := def_err.LetErr("KChan[T].Read(): empty list msg")
		return result.ResErr[T](err)
	}

	msg := sf.lstMsg[0]
	sf.lstMsg = sf.lstMsg[1:]
	return result.ResOk(msg)
}

// Limit -- ограничение размера канала.
func (sf *SafeChan[T]) Limit() def_spec_base.Num {
	sf.block.RLock()
	defer sf.block.RUnlock()
	return def_spec_base.Num(len(sf.lstMsg))
}

// Len -- возвращает количество элементов в канале.
func (sf *SafeChan[T]) Len() def_spec_base.Num {
	sf.block.RLock()
	defer sf.block.RUnlock()
	return def_spec_base.Num(len(sf.lstMsg))
}

// IsClosed -- возвращает признак закрытия канала.
func (sf *SafeChan[T]) IsClosed() def_spec_base.Bool {
	return sf.isClosed.IsOk()
}

// Close -- закрывает канал.
func (sf *SafeChan[T]) Close() {
	sf.block.Lock()
	defer sf.block.Unlock()
	sf.isClosed.Set()
}