commit 92de01576d1162667ba7db28f19f7aa043a75672 · julien.rbrt.fr/tangled-core

+17 -7

knotclient/events.go

···

       17
        
       	"github.com/gorilla/websocket"

     

       18
        
       )

     

       19
        
       

     

       20
       -
       type ProcessFunc func(source EventSource, message Message) error

     

       21
        
       

     

       22
        
       type Message struct {

     

       23
        
       	Rkey string

     
···

       39
        
       	CursorStore       CursorStore

     

       40
        
       }

     

       41
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       42
        
       type EventSource struct {

     

       43
        
       	Knot string

     

       44
        
       }

     
···

       50
        
       }

     

       51
        
       

     

       52
        
       type EventConsumer struct {

     

       53
       -
       	cfg        ConsumerConfig

     

       54
        
       	wg         sync.WaitGroup

     

       55
        
       	dialer     *websocket.Dialer

     

       56
        
       	connMap    sync.Map

     
···

       58
        
       	logger     *slog.Logger

     

       59
        
       	randSource *rand.Rand

     

       60
        
       

     

       61
       -
       	// rw lock over edits to consumer config

     

       62
       -
       	mu sync.RWMutex

     

       0
        
       
     

       63
        
       }

     

       64
        
       

     

       65
        
       type CursorStore interface {

     
···

       202
        
       }

     

       203
        
       

     

       204
        
       func (c *EventConsumer) AddSource(ctx context.Context, s EventSource) {

     

       205
       -
       	c.mu.Lock()

     

       206
        
       	c.cfg.Sources[s] = struct{}{}

     

       207
        
       	c.wg.Add(1)

     

       208
        
       	go c.startConnectionLoop(ctx, s)

     

       209
       -
       	c.mu.Unlock()

     

       210
        
       }

     

       211
        
       

     

       212
        
       func (c *EventConsumer) worker(ctx context.Context) {

     
···

       230
        
       			// update cursor

     

       231
        
       			c.cfg.CursorStore.Set(j.source.Knot, time.Now().Unix())

     

       232
        
       

     

       233
       -
       			if err := c.cfg.ProcessFunc(j.source, msg); err != nil {

     

       234
        
       				c.logger.Error("error processing message", "source", j.source, "err", err)

     

       235
        
       			}

     

       236
        
       		}

···

       17
        
       	"github.com/gorilla/websocket"

     

       18
        
       )

     

       19
        
       

     

       20
       +
       type ProcessFunc func(ctx context.Context, source EventSource, message Message) error

     

       21
        
       

     

       22
        
       type Message struct {

     

       23
        
       	Rkey string

     
···

       39
        
       	CursorStore       CursorStore

     

       40
        
       }

     

       41
        
       

     

       42
       +
       func NewConsumerConfig() *ConsumerConfig {

     

       43
       +
       	return &ConsumerConfig{

     

       44
       +
       		Sources: make(map[EventSource]struct{}),

     

       45
       +
       	}

     

       46
       +
       }

     

       47
       +
       

     

       48
       +
       func (cc *ConsumerConfig) AddEventSource(es EventSource) {

     

       49
       +
       	cc.Sources[es] = struct{}{}

     

       50
       +
       }

     

       51
       +
       

     

       52
        
       type EventSource struct {

     

       53
        
       	Knot string

     

       54
        
       }

     
···

       60
        
       }

     

       61
        
       

     

       62
        
       type EventConsumer struct {

     

       0
        
       
     

       63
        
       	wg         sync.WaitGroup

     

       64
        
       	dialer     *websocket.Dialer

     

       65
        
       	connMap    sync.Map

     
···

       67
        
       	logger     *slog.Logger

     

       68
        
       	randSource *rand.Rand

     

       69
        
       

     

       70
       +
       	// rw lock over edits to ConsumerConfig

     

       71
       +
       	cfgMu sync.RWMutex

     

       72
       +
       	cfg   ConsumerConfig

     

       73
        
       }

     

       74
        
       

     

       75
        
       type CursorStore interface {

     
···

       212
        
       }

     

       213
        
       

     

       214
        
       func (c *EventConsumer) AddSource(ctx context.Context, s EventSource) {

     

       215
       +
       	c.cfgMu.Lock()

     

       216
        
       	c.cfg.Sources[s] = struct{}{}

     

       217
        
       	c.wg.Add(1)

     

       218
        
       	go c.startConnectionLoop(ctx, s)

     

       219
       +
       	c.cfgMu.Unlock()

     

       220
        
       }

     

       221
        
       

     

       222
        
       func (c *EventConsumer) worker(ctx context.Context) {

     
···

       240
        
       			// update cursor

     

       241
        
       			c.cfg.CursorStore.Set(j.source.Knot, time.Now().Unix())

     

       242
        
       

     

       243
       +
       			if err := c.cfg.ProcessFunc(ctx, j.source, msg); err != nil {

     

       244
        
       				c.logger.Error("error processing message", "source", j.source, "err", err)

     

       245
        
       			}

     

       246
        
       		}