kern/v4/lev2/lti/kbus_base/kbus_base.go

// package kbus_base -- базовая часть шины данных.
package kbus_base

import (
	"sync"

	"gitp78su.ipnodns.ru/svi/kern/v4/lev0"
	mL1 "gitp78su.ipnodns.ru/svi/kern/v4/lev1"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev1/comp_spec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kern_ent/stream_name"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/kspec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/bus_ent"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/bus_spec"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/kbus/dict_topic_serve"
	"gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/kbus/dict_topic_sub"
)

var (
	busBaseStreamName = stream_name.NewAStreamName(lev0.LetTxt("bus_base"))
)

// KBusBase -- базовая часть шины данных.
type KBusBase struct {
	KCtx_     kspec.IKernelCtx
	IsWork_   lev0.ISafeBool
	lCtx      comp_spec.ILocalCtx
	log       comp_spec.ILogBuf
	dictSub   bus_spec.IDictTopicSub
	dictServe bus_spec.IDictTopicServe
}

var (
	Bus_  *KBusBase
	block sync.Mutex
)

// GetKernelBusBase -- возвращает базовую шину сообщений.
func GetKernelBusBase() *KBusBase {
	block.Lock()
	defer block.Unlock()
	if Bus_ != nil {
		return Bus_
	}
	kCtx := kctx.GetKernelCtx()
	lCtx := mL1.NewLocalCtx(kCtx.Ctx())
	Bus_ = &KBusBase{
		KCtx_:     kCtx,
		IsWork_:   mL1.NewSafeBool(false),
		dictSub:   dict_topic_sub.NewDictTopicSub(),
		dictServe: dict_topic_serve.NewDictServe(),
		lCtx:      lCtx,
	}
	Bus_.log = Bus_.lCtx.Log()
	go Bus_.close()
	go Bus_.run()
	Bus_.IsWork_.Set()
	Bus_.KCtx_.Wg().Add(busBaseStreamName)
	Bus_.KCtx_.Set("kernBusBase", Bus_, "base of data bus")
	_ = bus_spec.IKernelBus(Bus_)
	return Bus_
}

// Log -- возвращает лог шины.
func (sf *KBusBase) Log() comp_spec.ILogBuf {
	return sf.log
}

func (sf *KBusBase) run() {
	sf.log.Debug("KBusBase.run()")
	for {
		break
	}
}

// Unsubscribe -- отписывает обработчик от топика.
func (sf *KBusBase) Unsubscribe(handler bus_spec.IBusHandlerSubscribe) {
	msg := lev0.LetTxt("KBusBase.Unsubscribe(): handler='%v'",
		handler.Name())
	sf.log.Debug(msg.Get())
	sf.dictSub.Unsubscribe(handler)
}

// Subscribe -- подписывает обработчик на топик.
func (sf *KBusBase) Subscribe(handler bus_spec.IBusHandlerSubscribe) *lev0.Result[*lev0.LBool] {
	msg := lev0.LetTxt("KBusBase.Subscribe(): handler='%v'",
		handler.Name()).Get()
	sf.log.Debug(msg)
	lev0.If(!sf.IsWork_.Get()).Hassert("KBusBase.Subscribe():  handler='%v', bus already closed",
		[]any{handler.Name()})
	sf.dictSub.Subscribe(handler)
	return lev0.ResOk(lev0.LetBool(true))
}

// SendRequest -- отправляет запрос в шину данных.
func (sf *KBusBase) SendRequest(topic bus_ent.ITopicFix, binReq []byte) *lev0.Result[[]byte] {
	msg := lev0.LetTxt("KBusBase.SendRequest(): topic='%v'", topic.Get()).Get()
	sf.log.Debug(msg)
	if !sf.IsWork_.Get() {
		err := lev0.LetErr("KBusBase.SendRequest():  topic='%v', bus already closed", topic.Get())
		sf.log.Err(err)
		return lev0.ResErr[[]byte](err)
	}
	res := sf.dictServe.SendRequest(topic, binReq)
	if res.IsErr() {
		err := lev0.LetErr("KBusBase.SendRequest(): topic='%v', err=\n\t%w", topic.Get(), res.Err())
		sf.log.Err(err)
		return lev0.ResErr[[]byte](err)
	}
	return res
}

// RegisterServe -- регистрирует обработчики входящих запросов.
func (sf *KBusBase) RegisterServe(handler bus_spec.IBusHandlerServe) *lev0.Result[lev0.Bool] {
	if handler == nil {
		err := lev0.LetErr("KBusBase.RegisterServe(): IBusHandlerServe==nil")
		return lev0.ResErr[*lev0.LBool](err)
	}
	msg := lev0.LetTxt("KBusBase.RegisterServe(): handler='%v'", handler.Name()).Get()
	sf.log.Debug(msg)
	res := sf.dictServe.Register(handler)
	if res.IsErr() {
		err := lev0.LetErr("KBusBase.RegisterServe(): handler='%v', err=\n\t%w", handler.Name(), res.Err())
		sf.log.Err(err)
		return lev0.ResErr[*lev0.LBool](err)
	}
	return lev0.ResOk(lev0.LetBool(true))
}

// Publish -- публикует сообщение в шину.
func (sf *KBusBase) Publish(topic bus_ent.ITopicFix, binMsg []byte) *lev0.Result[lev0.Bool] {
	msg := lev0.LetTxt("KBusBase.Publish(): topic='%v'", topic).Get()
	sf.log.Debug(msg)
	if !sf.IsWork_.Get() {
		err := lev0.LetErr("KBusBase.Publish(): topic='%v',bus already closed", topic)
		sf.log.Err(err)
		return lev0.ResErr[lev0.Bool](err)
	}
	// Асинхронный запуск чтения
	go sf.dictSub.Read(topic, binMsg)
	return lev0.ResOk(lev0.Bool(true))
}

// IsWork -- возвращает признак работы шины.
func (sf *KBusBase) IsWork() *lev0.LBool {
	return lev0.LetBool(sf.IsWork_.Get())
}

// Ожидает закрытия шины в отдельном потоке.
func (sf *KBusBase) close() {
	sf.KCtx_.Wait()
	sf.KCtx_.Lock()
	defer sf.KCtx_.Unlock()
	if !sf.IsWork_.Get() {
		return
	}
	sf.IsWork_.Reset()
	sf.KCtx_.Wg().Done(busBaseStreamName)
	sf.log.Debug("KBusBase.close(): done")
}