kern/v4/d2/lti/kbus/dict_topic_serve/dict_topic_serve.go

// package dict_topic_serve -- словарь топиков обработчиков запросов.
package dict_topic_serve

import (
	"context"
	"time"

	"gitp78su.ipnodns.ru/svi/kern/v4/d0"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/kern_ctx"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_ent"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_ent/topic"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_spec"
)

// dictServe -- потокобезопасный словарь обработчиков запросов.
//
// Допускается только один обработчик запросов на один топик.
type dictServe struct {
	*d0.MSafeMutex
	kCtx      *kern_ctx.KernCtx
	owner     *d0.LOwner
	dictServe map[bus_ent.ATopic]bus_spec.IBusHandlerServe
}

// NewDictServe -- возвращает потокобезопасный словарь обработчиков запросов.
func NewDictServe(owner *d0.LOwner) *dictServe {
	sf := &dictServe{
		MSafeMutex: d0.MutSafeMutex(),
		kCtx:       kern_ctx.GetKernCtx(),
		owner:      owner,
		dictServe:  map[bus_ent.ATopic]bus_spec.IBusHandlerServe{},
	}
	return sf
}

// Register -- регистрирует обработчик запросов.
func (sf *dictServe) Register(handler bus_spec.IBusHandlerServe) *d0.Result[d0.Bool] {
	sf.Lock()
	defer sf.Unlock()
	if handler == nil {
		err := d0.LetErr(sf.owner, "dictServe.Register(): IBusHandlerSubscribe==nil")
		return d0.ResErr[d0.Bool](err)
	}
	topic := handler.Topic()
	isTwinRegister := sf.register(handler)
	if isTwinRegister {
		err := d0.LetErr(sf.owner, "dictServe.Register(): handler of topic (%v) already register", topic.Get())
		return d0.ResErr[d0.Bool](err)
	}
	return d0.ResOk(d0.Bool(true))
}

// Unregister -- удаляет обработчик запросов из словаря.
func (sf *dictServe) Unregister(handler bus_spec.IBusHandlerServe) {
	sf.Lock()
	defer sf.Unlock()
	if handler == nil {
		err := d0.LetErr(sf.owner, "dictServe.Unregister(): IBusHandlerSubscribe==nil")
		sf.kCtx.Log().Err(err)
		return
	}
	delete(sf.dictServe, handler.Topic().Get())
}

// SendRequest -- вызывает обработчик при поступлении запроса.
func (sf *dictServe) SendRequest(topic *topic.LTopic, binReq []byte) *d0.Result[[]byte] {
	sf.RLock()
	defer sf.RUnlock()
	handler, isOk := sf.dictServe[topic.Get()]
	if !isOk {
		err := d0.LetErr(sf.owner,
			"dictServe.SendRequest(): handler for topic (%v) not exists", topic.Get())
		return d0.ResErr[[]byte](err)
	}
	var (
		chRes = make(chan *d0.Result[[]byte], 2)
	)
	ctx, fnCancel := context.WithTimeout(sf.kCtx.SelfCtx(), time.Millisecond*time.Duration(TimeoutDefault))
	defer fnCancel()
	fnCall := func() {
		defer close(chRes)
		res := handler.FnBack(binReq)
		chRes <- res
	}
	go fnCall()
	select {
	case <-ctx.Done():
		err := d0.LetErr(sf.owner,
			"dictServe.SendRequest(): in call for topic (%v), err=\n\t%w", topic.Get(), ctx.Err())
		return d0.ResErr[[]byte](err)
	case res := <-chRes:
		return res
	}
}

var TimeoutDefault = 15000

// регистрирует обработчик запросов.
func (sf *dictServe) register(handler bus_spec.IBusHandlerServe) bool {
	topic := handler.Topic()
	_, isOk := sf.dictServe[topic.Get()]
	if isOk {
		return true
	}
	sf.dictServe[topic.Get()] = handler
	return false
}