back round 0 view raw

knotclient: introduce EventSource and Message #234

deleted

opened by

oppi.li 6 months ago targeting master from push-qoplqnlvlqqo

EventSource: a knot's identity (its hostname currently)
Message: message struct that follows the event structure to deserialize just rkey and nsid upfront, this could allow consumers to filter by nsid or configure cursor ranges to listen to

Signed-off-by: oppiliappan me@oppi.li

options

unified split

Changed files

+54 -6

knotclient

events.go

+54 -6

knotclient/events.go

···

       2
        
       

     

       3
        
       import (

     

       4
        
       	"context"

     

       0
        
       
     

       5
        
       	"log/slog"

     

       6
        
       	"math/rand"

     

       7
        
       	"net/url"

     
···

       13
        
       	"github.com/gorilla/websocket"

     

       14
        
       )

     

       15
        
       

     

       16
       -
       type ProcessFunc func(source string, message []byte) error

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       17
        
       

     

       18
        
       type ConsumerConfig struct {

     

       19
       -
       	Sources           []string

     

       20
        
       	ProcessFunc       ProcessFunc

     

       21
        
       	RetryInterval     time.Duration

     

       22
        
       	MaxRetryInterval  time.Duration

     
···

       24
        
       	WorkerCount       int

     

       25
        
       	QueueSize         int

     

       26
        
       	Logger            *slog.Logger

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       27
        
       }

     

       28
        
       

     

       29
        
       type EventConsumer struct {

     
···

       36
        
       	randSource *rand.Rand

     

       37
        
       }

     

       38
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       39
        
       type job struct {

     

       40
       -
       	source  string

     

       41
        
       	message []byte

     

       42
        
       }

     

       43
        
       

     
···

       104
        
       			if !ok {

     

       105
        
       				return

     

       106
        
       			}

     

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

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

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

     

       109
        
       			}

     

       110
        
       		}

     

       111
        
       	}

     

       112
        
       }

     

       113
        
       

     

       114
       -
       func (c *EventConsumer) startConnectionLoop(ctx context.Context, source string) {

     

       115
        
       	defer c.wg.Done()

     

       116
        
       	retryInterval := c.cfg.RetryInterval

     

       117
        
       	for {

     
···

       144
        
       	}

     

       145
        
       }

     

       146
        
       

     

       147
       -
       func (c *EventConsumer) runConnection(ctx context.Context, source string) error {

     

       148
        
       	connCtx, cancel := context.WithTimeout(ctx, c.cfg.ConnectionTimeout)

     

       149
        
       	defer cancel()

     

       150
        
       

     

       151
        
       	u, err := url.Parse(source)

     

       0
        
       
     

       0
        
       
     

       152
        
       	if err != nil {

     

       153
        
       		return err

     

       154
        
       	}

     

       155
        
       

     

       0
        
       
     

       156
        
       	conn, _, err := c.dialer.DialContext(connCtx, u.String(), nil)

     

       157
        
       	if err != nil {

     

       158
        
       		return err

···

       2
        
       

     

       3
        
       import (

     

       4
        
       	"context"

     

       5
       +
       	"encoding/json"

     

       6
        
       	"log/slog"

     

       7
        
       	"math/rand"

     

       8
        
       	"net/url"

     
···

       14
        
       	"github.com/gorilla/websocket"

     

       15
        
       )

     

       16
        
       

     

       17
       +
       type ProcessFunc func(source EventSource, message Message) error

     

       18
       +
       

     

       19
       +
       type Message struct {

     

       20
       +
       	Rkey string

     

       21
       +
       	Nsid string

     

       22
       +
       	// do not full deserialize this portion of the message, processFunc can do that

     

       23
       +
       	EventJson json.RawMessage

     

       24
       +
       }

     

       25
        
       

     

       26
        
       type ConsumerConfig struct {

     

       27
       +
       	Sources           []EventSource

     

       28
        
       	ProcessFunc       ProcessFunc

     

       29
        
       	RetryInterval     time.Duration

     

       30
        
       	MaxRetryInterval  time.Duration

     
···

       32
        
       	WorkerCount       int

     

       33
        
       	QueueSize         int

     

       34
        
       	Logger            *slog.Logger

     

       35
       +
       	Dev               bool

     

       36
       +
       }

     

       37
       +
       

     

       38
       +
       type EventSource struct {

     

       39
       +
       	Knot string

     

       40
       +
       }

     

       41
       +
       

     

       42
       +
       func NewEventSource(knot string) EventSource {

     

       43
       +
       	return EventSource{

     

       44
       +
       		Knot: knot,

     

       45
       +
       	}

     

       46
        
       }

     

       47
        
       

     

       48
        
       type EventConsumer struct {

     
···

       55
        
       	randSource *rand.Rand

     

       56
        
       }

     

       57
        
       

     

       58
       +
       func (e *EventConsumer) buildUrl(s EventSource, cursor string) (*url.URL, error) {

     

       59
       +
       	scheme := "wss"

     

       60
       +
       	if e.cfg.Dev {

     

       61
       +
       		scheme = "ws"

     

       62
       +
       	}

     

       63
       +
       

     

       64
       +
       	u, err := url.Parse(scheme + "://" + s.Knot + "/events")

     

       65
       +
       	if err != nil {

     

       66
       +
       		return nil, err

     

       67
       +
       	}

     

       68
       +
       

     

       69
       +
       	if cursor != "" {

     

       70
       +
       		query := url.Values{}

     

       71
       +
       		query.Add("cursor", cursor)

     

       72
       +
       		u.RawQuery = query.Encode()

     

       73
       +
       	}

     

       74
       +
       	return u, nil

     

       75
       +
       }

     

       76
       +
       

     

       77
        
       type job struct {

     

       78
       +
       	source  EventSource

     

       79
        
       	message []byte

     

       80
        
       }

     

       81
        
       

     
···

       142
        
       			if !ok {

     

       143
        
       				return

     

       144
        
       			}

     

       145
       +
       

     

       146
       +
       			var msg Message

     

       147
       +
       			err := json.Unmarshal(j.message, &msg)

     

       148
       +
       			if err != nil {

     

       149
       +
       				c.logger.Error("error deserializing message", "source", j.source.Knot, "err", err)

     

       150
       +
       				return

     

       151
       +
       			}

     

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

     

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

     

       154
        
       			}

     

       155
        
       		}

     

       156
        
       	}

     

       157
        
       }

     

       158
        
       

     

       159
       +
       func (c *EventConsumer) startConnectionLoop(ctx context.Context, source EventSource) {

     

       160
        
       	defer c.wg.Done()

     

       161
        
       	retryInterval := c.cfg.RetryInterval

     

       162
        
       	for {

     
···

       189
        
       	}

     

       190
        
       }

     

       191
        
       

     

       192
       +
       func (c *EventConsumer) runConnection(ctx context.Context, source EventSource) error {

     

       193
        
       	connCtx, cancel := context.WithTimeout(ctx, c.cfg.ConnectionTimeout)

     

       194
        
       	defer cancel()

     

       195
        
       

     

       196
        
       	u, err := url.Parse(source)

     

       197
       +
       

     

       198
       +
       	u, err := c.buildUrl(source, cursor)

     

       199
        
       	if err != nil {

     

       200
        
       		return err

     

       201
        
       	}

     

       202
        
       

     

       203
       +
       	c.logger.Info("connecting", "url", u.String())

     

       204
        
       	conn, _, err := c.dialer.DialContext(connCtx, u.String(), nil)

     

       205
        
       	if err != nil {

     

       206
        
       		return err