diff --git a/core/dispatcher.go b/core/dispatcher.go index eb6b24e..c105975 100644 --- a/core/dispatcher.go +++ b/core/dispatcher.go @@ -8,9 +8,13 @@ import ( maxbot "github.com/max-messenger/max-bot-api-client-go/v2" "github.com/max-messenger/max-bot-api-client-go/v2/model" "github.com/rs/zerolog/log" - "time" + "sync" ) +type DispatcherOption interface{} + +type EventIsolation interface{} + type FSMStorage interface { Get(ctx context.Context, userID int) (state.Snapshot, error) Put(ctx context.Context, state state.Snapshot) error @@ -22,69 +26,52 @@ type UpdateHandler interface { Use(mw ...Middleware) } -type Dispatcher struct { - PollInterval time.Duration - Api *maxbot.Api - Handler UpdateHandler - FSMStorage FSMStorage +type Poller interface { + Start(ctx context.Context) (<-chan model.Update, error) } -func (d *Dispatcher) getUpdates(ctx context.Context) <-chan model.Update { - var updates []model.Update - var marker int64 - var err error +type Dispatcher struct { + api *maxbot.Api + poller Poller + handler UpdateHandler + fsmStorage FSMStorage + eventIsolation EventIsolation + wg sync.WaitGroup +} - ch := make(chan model.Update) +func NewDispatcher(api *maxbot.Api, options ...DispatcherOption) *Dispatcher { + d := &Dispatcher{api: api} - go func() { - defer close(ch) + // todo: apply options - for { - select { - case <-ctx.Done(): - return - case <-time.Tick(d.PollInterval): - start := time.Now() - updates, marker, err = d.Api.Subscriptions.GetUpdates(ctx, marker) - latency := time.Since(start) - if err != nil && !errors.Is(err, context.Canceled) { - log.Error().Err(err).Send() - } - log.Ctx(ctx).Debug(). - Int("received", len(updates)). - Dur("latency", latency). - Msg("received updates") - - for _, update := range updates { - select { - case <-ctx.Done(): - return - case ch <- update: - } - } - } - } - }() - - return ch + return d } func (d *Dispatcher) handleUpdates(ch <-chan model.Update) { for update := range ch { - go d.handleUpdate(update) + d.wg.Go(func() { + d.handleUpdate(update) + }) } } -func (d *Dispatcher) StartPolling(ctx context.Context) { - d.handleUpdates(d.getUpdates(ctx)) +func (d *Dispatcher) Start(ctx context.Context) error { + ch, err := d.poller.Start(ctx) + if err != nil { + return err + } + + d.handleUpdates(ch) <-ctx.Done() + d.wg.Wait() + return nil } func (d *Dispatcher) handleUpdate(upd model.Update) { ctx := context.Background() var st *state.Context - snapshot, err := d.FSMStorage.Get(ctx, int(upd.UserID)) + snapshot, err := d.fsmStorage.Get(ctx, int(upd.UserID)) if err != nil && !errors.Is(err, state.ErrStateNotFound) { log.Error().Err(err).Send() return @@ -95,22 +82,22 @@ func (d *Dispatcher) handleUpdate(upd model.Update) { st = state.Restore(snapshot) } - c := mcontext.New(ctx, upd, st, d.Api) + c := mcontext.New(ctx, upd, st, d.api) - if err := d.Handler.HandleUpdate(c); err != nil { + if err := d.handler.HandleUpdate(c); err != nil { log.Error().Err(err).Send() return } st = c.State() if st.State() == state.ClearState { - if err := d.FSMStorage.Delete(ctx, int(upd.UserID)); err != nil { + if err := d.fsmStorage.Delete(ctx, int(upd.UserID)); err != nil { log.Error().Err(err).Send() return } } if st.State() != "" { - if err := d.FSMStorage.Put(ctx, st.Snapshot()); err != nil { + if err := d.fsmStorage.Put(ctx, st.Snapshot()); err != nil { log.Error().Err(err).Send() return } diff --git a/core/pollers.go b/core/pollers.go new file mode 100644 index 0000000..74c8333 --- /dev/null +++ b/core/pollers.go @@ -0,0 +1,86 @@ +package core + +import ( + "context" + "errors" + "fmt" + maxbot "github.com/max-messenger/max-bot-api-client-go/v2" + "github.com/max-messenger/max-bot-api-client-go/v2/model" + "github.com/rs/zerolog/log" + "time" +) + +const defaultInterval = 200 * time.Millisecond + +type PollerOption interface{} + +type Webhook struct{} + +type LongPoller struct { + api *maxbot.Api + pollInterval time.Duration +} + +type TestPoller struct{} + +func (w *Webhook) Start(_ context.Context) (<-chan model.Update, error) { + return nil, fmt.Errorf("unimplemented") +} + +func NewLongPoller(api *maxbot.Api, options ...PollerOption) *LongPoller { + lp := &LongPoller{ + api: api, + } + + if lp.pollInterval == 0 { + lp.pollInterval = defaultInterval + } + + return lp +} + +func (lp *LongPoller) Start(ctx context.Context) (<-chan model.Update, error) { + var updates []model.Update + var marker int64 + var err error + + ch := make(chan model.Update) + + go func() { + defer close(ch) + + // todo: backoff on errors + + for { + select { + case <-ctx.Done(): + return + case <-time.Tick(lp.pollInterval): + start := time.Now() + updates, marker, err = lp.api.Subscriptions.GetUpdates(ctx, marker) + latency := time.Since(start) + if err != nil && !errors.Is(err, context.Canceled) { + log.Error().Err(err).Send() + } + log.Ctx(ctx).Debug(). + Int("received", len(updates)). + Dur("latency", latency). + Msg("received updates") + + for _, update := range updates { + select { + case <-ctx.Done(): + return + case ch <- update: + } + } + } + } + }() + + return ch, nil +} + +func (tp *TestPoller) Start(_ context.Context) (<-chan model.Update, error) { + return nil, fmt.Errorf("unimplemented") +}