feat: dispatcher and long poller options

This commit is contained in:
abelith committed 2026-10-05 14:15:20 +03:00
1 parent b5f1b426af
commit fd2ae3db72
3 files changed
+109 -23

No files matched your search

+78 -13
View File
@@ -5,14 +5,13 @@ import (
"errors" "errors"
mcontext "github.com/abelith/gomax/context" mcontext "github.com/abelith/gomax/context"
"github.com/abelith/gomax/state" "github.com/abelith/gomax/state"
"github.com/abelith/gomax/state/memory"
maxbot "github.com/max-messenger/max-bot-api-client-go/v2" maxbot "github.com/max-messenger/max-bot-api-client-go/v2"
"github.com/max-messenger/max-bot-api-client-go/v2/model" "github.com/max-messenger/max-bot-api-client-go/v2/model"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
"sync" "sync"
) )
type DispatcherOption interface{}
type EventIsolation interface{} type EventIsolation interface{}
type FSMStorage interface { type FSMStorage interface {
@@ -30,29 +29,53 @@ type Poller interface {
Start(ctx context.Context) (<-chan model.Update, error) Start(ctx context.Context) (<-chan model.Update, error)
} }
type DispatcherOption interface {
apply(d *dispatcherOptions)
}
type Dispatcher struct { type Dispatcher struct {
api *maxbot.Api api *maxbot.Api
poller Poller poller Poller
handler UpdateHandler
fsmStorage FSMStorage fsmStorage FSMStorage
eventIsolation EventIsolation eventIsolation EventIsolation
rt *Router
wg sync.WaitGroup wg sync.WaitGroup
} }
func NewDispatcher(api *maxbot.Api, options ...DispatcherOption) *Dispatcher { type dispatcherOptions struct {
d := &Dispatcher{api: api} poller Poller
fsmStorage FSMStorage
eventIsolation EventIsolation
}
// todo: apply options type dispatcherOptionWrapper struct {
fn func(options *dispatcherOptions)
}
func NewDispatcher(api *maxbot.Api, options ...DispatcherOption) *Dispatcher {
d := &Dispatcher{
api: api,
rt: &Router{},
}
opts := newDispatcherOptions(api)
for _, opt := range options {
opt.apply(opts)
}
d.poller = opts.poller
d.fsmStorage = opts.fsmStorage
d.eventIsolation = opts.eventIsolation
return d return d
} }
func (d *Dispatcher) handleUpdates(ch <-chan model.Update) { func (d *Dispatcher) HandleUpdate(c *mcontext.Context) error {
for update := range ch { return d.rt.HandleUpdate(c)
d.wg.Go(func() { }
d.handleUpdate(update)
}) func (d *Dispatcher) Use(mw ...Middleware) {
} d.rt.Use(mw...)
} }
func (d *Dispatcher) Start(ctx context.Context) error { func (d *Dispatcher) Start(ctx context.Context) error {
@@ -67,6 +90,14 @@ func (d *Dispatcher) Start(ctx context.Context) error {
return nil return nil
} }
func (d *Dispatcher) handleUpdates(ch <-chan model.Update) {
for update := range ch {
d.wg.Go(func() {
d.handleUpdate(update)
})
}
}
func (d *Dispatcher) handleUpdate(upd model.Update) { func (d *Dispatcher) handleUpdate(upd model.Update) {
ctx := context.Background() ctx := context.Background()
var st *state.Context var st *state.Context
@@ -84,7 +115,7 @@ func (d *Dispatcher) handleUpdate(upd model.Update) {
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.rt.HandleUpdate(c); err != nil {
log.Error().Err(err).Send() log.Error().Err(err).Send()
return return
} }
@@ -103,3 +134,37 @@ func (d *Dispatcher) handleUpdate(upd model.Update) {
} }
} }
} }
func WithPoller(poller Poller) DispatcherOption {
return newOption(func(opts *dispatcherOptions) {
opts.poller = poller
})
}
func WithFSMStorage(storage FSMStorage) DispatcherOption {
return newOption(func(opts *dispatcherOptions) {
opts.fsmStorage = storage
})
}
func WithEventIsolation(eventIsolation EventIsolation) DispatcherOption {
return newOption(func(opts *dispatcherOptions) {
opts.eventIsolation = eventIsolation
})
}
func newDispatcherOptions(api *maxbot.Api) *dispatcherOptions {
return &dispatcherOptions{
poller: NewLongPoller(api),
fsmStorage: memory.NewMemoryFSMStorage(),
eventIsolation: memory.NewMemoryEventIsolation(),
}
}
func newOption(fn func(options *dispatcherOptions)) DispatcherOption {
return dispatcherOptionWrapper{fn: fn}
}
func (dow dispatcherOptionWrapper) apply(opts *dispatcherOptions) {
dow.fn(opts)
}
+19 -4
View File
@@ -12,7 +12,9 @@ import (
const defaultInterval = 200 * time.Millisecond const defaultInterval = 200 * time.Millisecond
type PollerOption interface{} type LongPollerOption interface {
apply(opts *longPollerOptions)
}
type Webhook struct{} type Webhook struct{}
@@ -23,19 +25,26 @@ type LongPoller struct {
type TestPoller struct{} type TestPoller struct{}
type longPollerOptions struct {
pollInterval time.Duration
}
func (w *Webhook) Start(_ context.Context) (<-chan model.Update, error) { func (w *Webhook) Start(_ context.Context) (<-chan model.Update, error) {
return nil, fmt.Errorf("unimplemented") return nil, fmt.Errorf("unimplemented")
} }
func NewLongPoller(api *maxbot.Api, options ...PollerOption) *LongPoller { func NewLongPoller(api *maxbot.Api, options ...LongPollerOption) *LongPoller {
lp := &LongPoller{ lp := &LongPoller{
api: api, api: api,
} }
if lp.pollInterval == 0 { opts := newLongPollerOptions()
lp.pollInterval = defaultInterval for _, opt := range options {
opt.apply(opts)
} }
lp.pollInterval = opts.pollInterval
return lp return lp
} }
@@ -84,3 +93,9 @@ func (lp *LongPoller) Start(ctx context.Context) (<-chan model.Update, error) {
func (tp *TestPoller) Start(_ context.Context) (<-chan model.Update, error) { func (tp *TestPoller) Start(_ context.Context) (<-chan model.Update, error) {
return nil, fmt.Errorf("unimplemented") return nil, fmt.Errorf("unimplemented")
} }
func newLongPollerOptions() *longPollerOptions {
return &longPollerOptions{
pollInterval: defaultInterval,
}
}
+12 -6
View File
@@ -10,25 +10,31 @@ type FSMStorage struct {
m sync.Map m sync.Map
} }
type EventIsolation struct{}
func NewMemoryFSMStorage() *FSMStorage { func NewMemoryFSMStorage() *FSMStorage {
return &FSMStorage{} return &FSMStorage{}
} }
func (f *FSMStorage) Get(_ context.Context, userID int) (*state.Snapshot, error) { func (f *FSMStorage) Get(_ context.Context, userID int) (state.Snapshot, error) {
st, ok := f.m.Load(userID) st, ok := f.m.Load(userID)
if !ok { if !ok {
return nil, state.ErrStateNotFound return state.Snapshot{}, state.ErrStateNotFound
} }
return st.(*state.Snapshot), nil return st.(state.Snapshot), nil
} }
func (f *FSMStorage) Put(_ context.Context, snapshot *state.Snapshot) error { func (f *FSMStorage) Put(_ context.Context, snapshot state.Snapshot) error {
f.m.Store(snapshot.ID, snapshot) f.m.Store(snapshot.ID, snapshot)
return nil return nil
} }
func (f *FSMStorage) Delete(_ context.Context, snapshot *state.Snapshot) error { func (f *FSMStorage) Delete(_ context.Context, userID int) error {
f.m.Delete(snapshot.ID) f.m.Delete(userID)
return nil return nil
} }
func NewMemoryEventIsolation() *EventIsolation {
return &EventIsolation{}
}