feat: added dispatcher lifecycle hooks
This commit is contained in:
1 parent
7a4e34528f
commit
caacccec56
1 file changed
+39
-2
+39
-2
@@ -51,6 +51,8 @@ type Dispatcher struct {
|
|||||||
poller Poller
|
poller Poller
|
||||||
fsmStorage FSMStorage
|
fsmStorage FSMStorage
|
||||||
eventIsolation EventIsolation
|
eventIsolation EventIsolation
|
||||||
|
onStart func(ctx context.Context) error
|
||||||
|
onStop func(ctx context.Context) error
|
||||||
rt *Router
|
rt *Router
|
||||||
logger zerolog.Logger
|
logger zerolog.Logger
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
@@ -81,6 +83,7 @@ func NewDispatcher(api *maxbot.Api, options ...DispatcherOption) *Dispatcher {
|
|||||||
d.poller = opts.poller
|
d.poller = opts.poller
|
||||||
d.fsmStorage = opts.fsmStorage
|
d.fsmStorage = opts.fsmStorage
|
||||||
d.eventIsolation = opts.eventIsolation
|
d.eventIsolation = opts.eventIsolation
|
||||||
|
d.logger = opts.logger
|
||||||
|
|
||||||
return d
|
return d
|
||||||
}
|
}
|
||||||
@@ -107,13 +110,47 @@ func (d *Dispatcher) Start(ctx context.Context) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if d.onStart != nil {
|
||||||
|
if err := d.onStart(ctx); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
d.handleUpdates(ch)
|
d.handleUpdates(ch)
|
||||||
d.wg.Wait()
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Dispatcher) Stop(ctx context.Context) error {
|
func (d *Dispatcher) Stop(ctx context.Context) error {
|
||||||
return d.poller.Stop(ctx)
|
if err := d.poller.Stop(ctx); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
done := make(chan struct{})
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
d.wg.Wait()
|
||||||
|
done <- struct{}{}
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
case <-done:
|
||||||
|
}
|
||||||
|
|
||||||
|
if d.onStop != nil {
|
||||||
|
return d.onStop(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *Dispatcher) OnStart(f func(ctx context.Context) error) {
|
||||||
|
d.onStart = f
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *Dispatcher) OnStop(f func(ctx context.Context) error) {
|
||||||
|
d.onStop = f
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Dispatcher) handleUpdates(ch <-chan model.Update) {
|
func (d *Dispatcher) handleUpdates(ch <-chan model.Update) {
|
||||||
|
|||||||
Reference in new issue
Block a user