dispatcher

Documentation

Index

• type Dispatcher
  • func New() (Dispatcher, error)
• type DispatcherImpl
  • func (d *DispatcherImpl) GetRegister() register.Register
  • func (d *DispatcherImpl) GetSubscribers() map[string]subscriber.Subscriber
  • func (d *DispatcherImpl) GetSubscribersPool() map[string]*goroutinepool.GoroutinePool
  • func (d *DispatcherImpl) RegisterInput(i input.Input)
  • func (d *DispatcherImpl) RegisterSubscriber(s subscriber.Subscriber)
  • func (d *DispatcherImpl) Start()
  • func (d *DispatcherImpl) Stop()
• type Router
  • func NewRouter(dispatcher *DispatcherImpl) (*Router, error)
  • func (r *Router) GetFilters() []filters.Filter
  • func (r *Router) RegisterFilter(f filters.Filter)
  • func (r *Router) Route(m *types.Message) *errors.DispatchError

Types

type Dispatcher

type Dispatcher interface {
	RegisterSubscriber(s subscriber.Subscriber)
	RegisterInput(input.Input)
	Start() // block until stopped
	Stop()  // block until stopped
}

func New

func New() (Dispatcher, error)

type DispatcherImpl

type DispatcherImpl struct {
	// contains filtered or unexported fields
}

func (*DispatcherImpl) GetRegister

func (d *DispatcherImpl) GetRegister() register.Register

func (*DispatcherImpl) GetSubscribers

func (d *DispatcherImpl) GetSubscribers() map[string]subscriber.Subscriber

func (*DispatcherImpl) GetSubscribersPool

func (d *DispatcherImpl) GetSubscribersPool() map[string]*goroutinepool.GoroutinePool

func (*DispatcherImpl) RegisterInput

func (d *DispatcherImpl) RegisterInput(i input.Input)

func (*DispatcherImpl) RegisterSubscriber

func (d *DispatcherImpl) RegisterSubscriber(s subscriber.Subscriber)

func (*DispatcherImpl) Start

func (d *DispatcherImpl) Start()

func (*DispatcherImpl) Stop

func (d *DispatcherImpl) Stop()

1. 关闭所有输入端 (httpserver, inputs) 2. 等待 pool 里的所有消息发送完 3. 关闭 pool 4. 关闭 register

type Router

type Router struct {
	// contains filtered or unexported fields
}

+-------------------------+
|     Router              |
|                         |
|     +-------------+     |      +-------------+
|     |             |     |      |  backend    |

src ----------+---->| A +-----+------> message |

|     |             |     |      |  consumer   |
|     +-------------+     |      | e.g.dingding|
+-------------------------+      +-------------+

A: []filter

[]filter:

+---------------+  +---------------+  +----------------+	 +-----------------+
| unifylabels   +--> registerlabel +--> webhookfilter  +-->  lastfilter     |
|               |  |               |  |                |	 |                 |
+---------------+  +---------------+  +----------------+	 +-----------------+

func NewRouter

func NewRouter(dispatcher *DispatcherImpl) (*Router, error)

func (*Router) GetFilters

func (r *Router) GetFilters() []filters.Filter

func (*Router) RegisterFilter

func (r *Router) RegisterFilter(f filters.Filter)

func (*Router) Route

func (r *Router) Route(m *types.Message) *errors.DispatchError