SVI Исправления, добавление умного канала

v4/lev0/kspec/ichan.go

@@ -0,0 +1,15 @@
+package kspec
+
+// IChan -- умный канал
+type IChan[T any] interface {
+	// Read -- читает сообщение из канала
+	Read() IResult[T]
+	// Limit -- лимит канала
+	Limit() int
+	// Len -- количество сообщений в канале
+	Len() int
+	// Close -- закрыть канал
+	Close()
+	// IsClosed -- проверяет закрыт ли канал
+	IsClosed() bool
+}

v4/lev2/kchan/kchan.go

@@ -2,12 +2,34 @@
 package kchan
 
 import (
+	"fmt"
 	"sync"
 
-	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/etypes/ebool"
+	mL0 "gitp78su.ipnodns.ru/svi/kern/v4/lev0"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec"
+	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/safe_bool"
 )
 
+var (
+	hassert = mL0.Hassert
+)
+
+// KChanParam -- параметры умного канала
+type KChanParam[T any] struct {
+	Limit_   int             // лимит на размер канала, не может быть пустым
+	OnWrite_ func(T)         // Обратный вызов при записи в канал
+	OnRead_  func(T)         // Обратный вызов при чтении из канала
+	OnClose_ func()          // Обратный вызов при закрытии канала
+	OnLimit_ func()          // Обратный вызов при достижении лимита
+	Ctx_     kspec.ILocalCtx // Локальный контекст для контроля создателя
+}
+
+// SelfCheck -- проверка корректности правильности параметров умного канала
+func (sf *KChanParam[T]) SelfCheck() {
+	hassert(sf.Limit_ > 0, "KChanParam[T].SelfCheck(): limit=%v, канал должен иметь положительный лимит")
+	hassert(sf.Ctx_ != nil, "KChanParam[T]: ctx==nil")
+}
+
 // KChan -- умный потокобезопасный канал
 //
 // Канал является резиновым, но лимит можно поставить сверху.
@@ -20,23 +42,66 @@ import (
 // Кроме того, можно повесить хуки на события записи, чтения и закрытия
 // (например, для целей валидации).
 type KChan[T any] struct {
-	block sync.RWMutex
+	*KChanParam[T]
+	block    sync.RWMutex
 	isClosed kspec.ISafeBool
-	lstMsg []T
-	onWrite func(T)
-	onRead  func(T)
-	onClose func()
+	lstMsg   []T
 }
 
 // NewKChan -- создаёт новый безопасный умный канал
-func NewKChan[T any](fnWrite func()T) *KChan[T] {
-	sf := &KChan[T]{}
+func NewKChan[T any](param *KChanParam[T]) kspec.IChan[T] {
+	hassert(param != nil, "NewKChan: param==nil")
+	param.SelfCheck()
+	sf := &KChan[T]{
+		KChanParam: param,
+		isClosed:   safe_bool.NewSafeBool(true),
+	}
+	go sf.close()
 	return sf
 }
 
-// Close -- закрывает канал
-func (sf *KChan[T]) Close() error {
-	if sf.block.TryLock() == false {
-		return nil
+// Read -- возвращает первый элемент очереди (если есть)
+func (sf *KChan[T]) Read() kspec.IResult[T] {
+	sf.block.Lock()
+	defer sf.block.Unlock()
+	if len(sf.lstMsg) == 0 {
+		err := fmt.Errorf("KChan[T].Read(): empty list msg")
+		return mL0.NewErr[T](err)
 	}
+
+	msg := sf.lstMsg[0]
+	sf.lstMsg = sf.lstMsg[1:]
+	return mL0.NewOk(msg)
+}
+
+// Limit -- ограничение размера канала
+func (sf *KChan[T]) Limit() int {
+	sf.block.RLock()
+	defer sf.block.RUnlock()
+	return len(sf.lstMsg)
+}
+
+// Len -- возвращает количество элементов в канале
+func (sf *KChan[T]) Len() int {
+	sf.block.RLock()
+	defer sf.block.RUnlock()
+	return len(sf.lstMsg)
+}
+
+// IsClosed -- возвращает признак закрытия канала
+func (sf *KChan[T]) IsClosed() bool {
+	return sf.isClosed.Get()
+}
+
+// Close -- закрывает канал
+func (sf *KChan[T]) Close() {
+	sf.block.Lock()
 	defer sf.block.Unlock()
+	sf.isClosed.Set()
+}
+
+// Ожидает закрытия канала по контексту
+func (sf *KChan[T]) close() {
+	sf.Ctx_.Wait()
+	sf.isClosed.Set()
+}

v4/lev2/kmodule/kmodule.go

@@ -8,6 +8,7 @@ import (
 	mKd "gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/module_name"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/etypes/ebool"
+	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec"
 	mKs "gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec"
 	mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kbus_local"
@@ -84,7 +85,7 @@ func (sf *kModule) sigLive() {
 	var (
 		topic  = mKd.NewTopic(sf.name.Get() + "_live")
 		iPhase = 0
-		res    mL0.IResult[bool]
+		res    mL0.IResult[kspec.EBool]
 	)
 	fnPhase := func() {
 		time.Sleep(time.Millisecond * time.Duration(sf.timePhase.Get()))

v4/lev2/kmonolit/kmonolit.go

@@ -51,8 +51,8 @@ func GetMonolit(name string) *kMonolit {
 		lCtx:    lCtx,
 		name:    name,
 		dict:    map[*module_name.AModuleName]mKs.IKernelModule{},
-		isWork:  mL1.NewSafeBool(),
-		isEnd:   mL1.NewSafeBool(),
+		isWork:  mL1.NewSafeBool(false),
+		isEnd:   mL1.NewSafeBool(false),
 		isLocal: ebool.NewEBool(isLocal),
 	}
 	sf.log = sf.lCtx.Log()

v4/lev2/kmonolit/kmonolit_test.go

@@ -87,7 +87,7 @@ func (sf *tester) run() {
 	sf.mon.Add(mod)
 	sf.mon.Run()
 	isWork := sf.mon.IsWork()
-	if !isWork {
+	if isWork.IsNot() {
 		sf.t.Fatalf("newGood1(): isWork==false")
 	}
 }
@@ -116,7 +116,7 @@ func (sf *tester) newGood1() {
 	}
 	sf.mon = GetMonolit("test_monolit")
 	isLocal := sf.mon.IsLocal()
-	if !isLocal {
+	if isLocal.IsNot() {
 		sf.t.Fatalf("newGood1(): isLocal==false")
 	}
 	if name := sf.mon.Name(); name != "test_monolit" {

v4/lev2/kserv_http/kserv_http.go

@@ -10,16 +10,17 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gofiber/contrib/v3/monitor"
 	"github.com/gofiber/fiber/v3"
 	"github.com/gofiber/fiber/v3/middleware/compress"
-	"github.com/gofiber/fiber/v3/middleware/filesystem"
-	"github.com/gofiber/fiber/v3/middleware/monitor"
+	"github.com/gofiber/fiber/v3/middleware/static"
 
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/defs/stream_name"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/etypes/ebool"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev0/helpers"
 	mKs "gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec"
 	mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1"
+	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/log_buf"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx"
 )
 
@@ -49,7 +50,11 @@ var (
 )
 
 func getParams() (string, mKs.ILogBuf, mKs.IKernelCtx) {
-	log := mL1.NewLogBuf(mL1.OptIsTerm(true), mL1.OptPrefix("kServHttp"))
+	param := &log_buf.LogBufParam{
+		IsTerm_: mL1.NewSafeBool(true),
+		Prefix_: "kServHttp",
+	}
+	log := mL1.NewLogBuf(param)
 	log.Debug("GetKernelServHttp(): first run")
 	strLocalUrl := os.Getenv("LOCAL_HTTP_URL")
 	hassert(strLocalUrl != "", "getParams(): env LOCAL_HTTP_URL not set")
@@ -83,19 +88,24 @@ func GetKernelServHttp() mKs.IKernelServerHttp {
 		lCtx:     mL1.NewLocalCtx(kCtx.Ctx()),
 		strUrl:   strLocalUrl,
 		fiberApp: fiber.New(confFiber),
-		isWork:   mL1.NewSafeBool(),
-		isEnd:    mL1.NewSafeBool(),
+		isWork:   mL1.NewSafeBool(false),
+		isEnd:    mL1.NewSafeBool(false),
 	}
 	sf.fiberApp.Use(compress.New(compress.Config{
 		Level: compress.LevelBestCompression, // 2
 	}))
-	sf.fiberApp.Use("/static", filesystem.New(filesystem.Config{
+	sf.fiberApp.Get("/static", static.New(static.Config{
 		Root:       http.FS(embedDirStatic),
 		PathPrefix: "static",
 		Browse:     true,
 		MaxAge:     3600 * 24,
 	}))
-	sf.fiberApp.Get("/monitor", monitor.New(monitor.Config{Title: strMonolit}))
+	// Или расширьте конфигурацию для индивидуальной настройки
+	// Назначьте промежуточное программное обеспечение для /metrics
+	// и измените заголовок на `Страница метрик MyService`
+	sf.fiberApp.Get("/metrics", monitor.New(monitor.Config{
+		Title: fmt.Sprintf("Страница метрик %q", strMonolit),
+	}))
 	sf.kCtx.Wg().Add(streamName)
 	kCtx.Set("fiberApp", sf.fiberApp, "GetKernelServHttp() internal fiber app")
 	kernServHttp = sf

v4/lev2/kserv_http_std/kserv_http_std.go

@@ -10,6 +10,8 @@ import (
 	mL0 "gitp78su.ipnodns.ru/svi/kern/v4/lev0"
 	mKs "gitp78su.ipnodns.ru/svi/kern/v4/lev0/kspec"
 	mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1"
+	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/log_buf"
+	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/safe_bool"
 	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx"
 )
 
@@ -33,7 +35,11 @@ func GetKservHttpStd() mL0.IResult[*kServHttpStd] {
 	if kServ != nil {
 		return mL0.NewOk(kServ)
 	}
-	log := mL1.NewLogBuf(mL1.OptIsTerm(true), mL1.OptPrefix("kServHttpStd"))
+	logBufParam := &log_buf.LogBufParam{
+		IsTerm_: safe_bool.NewSafeBool(true),
+		Prefix_: "kServHttpStd",
+	}
+	log := mL1.NewLogBuf(logBufParam)
 	log.Debug("GetKservHttpStd()")
 	resPort := mL1.NewSafeIntGetenv("SERV_HTTP_STD_PORT")
 	resPort.Hassert("in get env SERV_HTTP_STD_PORT")

v4/lev2/kwg/kwg_test.go

@@ -93,7 +93,7 @@ func (sf *tester) wait() {
 	sf.fnCancel()
 	for {
 		time.Sleep(time.Millisecond * 10)
-		if !sf.wg.IsWork().Get() {
+		if sf.wg.IsWork().IsNot() {
 			break
 		}
 	}
@@ -117,7 +117,7 @@ func (sf *tester) newGood1() {
 	if sf.wg == nil {
 		sf.t.Fatalf("newGood1(): KernelWg==nil")
 	}
-	if !sf.wg.IsWork().Get() {
+	if sf.wg.IsWork().IsNot() {
 		sf.t.Fatalf("wait(): isWork==false")
 	}
 	wg := GetKernelWg(sf.ctx)

v4/lev3/mod_kctx/mod_kctx_test.go

@@ -48,7 +48,7 @@ func (sf *tester) done() {
 	sf.t.Log("done")
 	sf.kCtx.Cancel()
 	sf.kCtx.Wg().Wait()
-	if isWork := sf.mod.IsWork(); isWork.Get() {
+	if isWork := sf.mod.IsWork(); isWork.IsOk() {
 		sf.t.Fatalf("newGood1(): isWork==true")
 	}
 }
@@ -74,7 +74,7 @@ func (sf *tester) newGood1() {
 	go sf.mod.Run()
 	for {
 		time.Sleep(time.Millisecond * 1)
-		if sf.mod.IsWork().Get() {
+		if sf.mod.IsWork().IsOk() {
 			break
 		}
 	}

v4/lev3/mod_keeper/mod_keeper_test.go

@@ -48,7 +48,7 @@ func (sf *tester) done() {
 	sf.t.Log("done")
 	sf.kCtx.Cancel()
 	sf.kCtx.Wg().Wait()
-	if isWork := sf.mod.IsWork(); isWork {
+	if isWork := sf.mod.IsWork(); isWork.IsOk() {
 		sf.t.Fatalf("newGood1(): isWork==true")
 	}
 }
@@ -74,7 +74,7 @@ func (sf *tester) newGood1() {
 	go sf.mod.Run()
 	for {
 		time.Sleep(time.Millisecond * 1)
-		if sf.mod.IsWork() {
+		if sf.mod.IsWork().IsOk() {
 			break
 		}
 	}

v4/lev3/mod_serv_http/mod_serv_http_test.go

@@ -53,7 +53,7 @@ func (sf *tester) done() {
 	sf.t.Log("done")
 	sf.ctx.Cancel()
 	sf.ctx.Wg().Wait()
-	if isWork := sf.mod.IsWork(); isWork {
+	if isWork := sf.mod.IsWork(); isWork.IsOk() {
 		sf.t.Fatalf("newGood1(): isWork==true")
 	}
 }
@@ -80,7 +80,7 @@ func (sf *tester) newGood1() {
 	go sf.mod.Run()
 	for {
 		time.Sleep(time.Millisecond * 1)
-		if sf.mod.IsWork() {
+		if sf.mod.IsWork().IsOk() {
 			return
 		}
 	}

v4/lev3/mod_wui/mod_wui_test.go

@@ -209,7 +209,7 @@ func (sf *tester) done() {
 	sf.kCtx.Cancel()
 	sf.kCtx.Wg().Wait()
 
-	if isWork := sf.mod.IsWork(); isWork {
+	if isWork := sf.mod.IsWork(); isWork.IsOk() {
 		sf.t.Fatalf("done(): isWork==true")
 	}
 }
@@ -240,7 +240,7 @@ func (sf *tester) newGood1() {
 	for {
 		time.Sleep(time.Millisecond * 1)
 
-		if sf.mod.IsWork() {
+		if sf.mod.IsWork().IsOk() {
 			break
 		}
 	}