// 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 }