// package safe_chan -- умный безопасный канал package safe_chan import ( "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/def_err" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/def_spec_base" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/safe_bool" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/safe_mutex" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/helpers" "gitp78su.ipnodns.ru/svi/kern/v4/lev0/result" ) // SafeChanParam -- параметры умного безопасного канала. type SafeChanParam[T any] struct { Limit_ def_spec_base.Num // лимит на размер канала, не может быть пустым OnWrite_ func(T) // Обратный вызов при записи в канал OnRead_ func(T) // Обратный вызов при чтении из канала OnClose_ func() // Обратный вызов при закрытии канала OnLimit_ func() // Обратный вызов при достижении лимита } // SelfCheck -- проверка корректности правильности параметров умного канала. func (sf *SafeChanParam[T]) SelfCheck() { helpers.If(sf.Limit_ <= 0). Hassert("SafeChanParam[T].SelfCheck(): limit=%v, канал должен иметь положительный лимит", sf.Limit_) } // 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(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.Get() } // Close -- закрывает канал. func (sf *SafeChan[T]) Close() { sf.block.Lock() defer sf.block.Unlock() sf.isClosed.Set() }