feat: Poller interface for dispatcher, graceful shutdown for dispatcher

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 {
+type Poller interface {
-	PollInterval time.Duration
+	Start(ctx context.Context) (<-chan model.Update, error)
-	Api          *maxbot.Api
-	Handler      UpdateHandler
-	FSMStorage   FSMStorage
 }
 
-func (d *Dispatcher) getUpdates(ctx context.Context) <-chan model.Update {
+type Dispatcher struct {
-	var updates []model.Update
+	api            *maxbot.Api
-	var marker int64
+	poller         Poller
-	var err error
+	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() {
+	// todo: apply options
-		defer close(ch)
 
-		for {
+	return d
-			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
 }
 
 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) {
+func (d *Dispatcher) Start(ctx context.Context) error {
-	d.handleUpdates(d.getUpdates(ctx))
+	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
 		}

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")
+}