dict_sub_hook.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. // package dict_sub_hook -- словарь потребителей топика по подписке.
  2. package dict_sub_hook
  3. import (
  4. "sync"
  5. "gitp78su.ipnodns.ru/svi/kern/v4/lev0"
  6. "gitp78su.ipnodns.ru/svi/kern/v4/lev2/kctx"
  7. "gitp78su.ipnodns.ru/svi/kern/v4/lev2/kspec"
  8. "gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/bus_ent"
  9. "gitp78su.ipnodns.ru/svi/kern/v4/lev2/lti/bus_spec"
  10. )
  11. // dictSubHook -- словарь потребителей топика по подписке.
  12. type dictSubHook struct {
  13. kCtx kspec.IKernelCtx
  14. dict map[bus_ent.AHandlerName]struct{} // В качестве ключа -- URL веб-хука
  15. block sync.RWMutex
  16. }
  17. // NewDictSubHook -- возвращает новый словарь веб-хуков одного топика.
  18. func NewDictSubHook() bus_spec.IDictSubHook {
  19. sf := &dictSubHook{
  20. kCtx: kctx.GetKernelCtx(),
  21. dict: map[bus_ent.AHandlerName]struct{}{},
  22. }
  23. return sf
  24. }
  25. // Unsubscribe -- удаляет из словаря подписки обработчик.
  26. func (sf *dictSubHook) Unsubscribe(handler bus_spec.IBusHandlerSubscribe) {
  27. sf.block.Lock()
  28. defer sf.block.Unlock()
  29. if handler == nil {
  30. err := lev0.LetErr("dictSubHook.Unsubscribe(): handler==nil")
  31. sf.kCtx.Log().Err(err)
  32. return
  33. }
  34. handlerName := handler.Name()
  35. delete(sf.dict, handlerName.Get())
  36. sf.kCtx.Del(lev0.LetTxt(handler.Name().Get()))
  37. }
  38. // Subscribe -- добавляет в словарь подписки новый обработчик.
  39. func (sf *dictSubHook) Subscribe(handler bus_spec.IBusHandlerSubscribe) {
  40. sf.block.Lock()
  41. defer sf.block.Unlock()
  42. helpers.If(handler == nil).Hassert("dictSubHook.Subscribe(): handler==nil", []any{})
  43. handlerName := handler.Name()
  44. sf.dict[handlerName.Get()] = struct{}{}
  45. sf.kCtx.Set(handlerName.Txt(), handler, "subscribe handler")
  46. }
  47. // Read -- вызывает все обработчики словаря подписок.
  48. func (sf *dictSubHook) Read(binMsg []byte) {
  49. sf.block.RLock()
  50. defer sf.block.RUnlock()
  51. for handlerName := range sf.dict {
  52. optHand := sf.kCtx.Get(defs.TxtFix(handlerName))
  53. if optHand.IsNone() {
  54. sf.kCtx.Del(defs.TxtFix(handlerName))
  55. continue
  56. }
  57. handler := optHand.Some().Val().(bus_spec.IBusHandlerSubscribe)
  58. go handler.FnBack(binMsg)
  59. }
  60. }