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
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
        
       	"context"

     

       5
       5
       +
       	"encoding/json"

     

       5
       6
        
       	"log/slog"

     

       6
       7
        
       	"math/rand"

     

       7
       8
        
       	"net/url"

     
···

       13
       14
        
       	"github.com/gorilla/websocket"

     

       14
       15
        
       )

     

       15
       16
        
       

     

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

     

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

     

       18
       18
       +
       

     

       19
       19
       +
       type Message struct {

     

       20
       20
       +
       	Rkey string

     

       21
       21
       +
       	Nsid string

     

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

     

       23
       23
       +
       	EventJson json.RawMessage

     

       24
       24
       +
       }

     

       17
       25
        
       

     

       18
       26
        
       type ConsumerConfig struct {

     

       19
       19
       -
       	Sources           []string

     

       27
       27
       +
       	Sources           []EventSource

     

       20
       28
        
       	ProcessFunc       ProcessFunc

     

       21
       29
        
       	RetryInterval     time.Duration

     

       22
       30
        
       	MaxRetryInterval  time.Duration

     
···

       24
       32
        
       	WorkerCount       int

     

       25
       33
        
       	QueueSize         int

     

       26
       34
        
       	Logger            *slog.Logger

     

       35
       35
       +
       	Dev               bool

     

       36
       36
       +
       }

     

       37
       37
       +
       

     

       38
       38
       +
       type EventSource struct {

     

       39
       39
       +
       	Knot string

     

       40
       40
       +
       }

     

       41
       41
       +
       

     

       42
       42
       +
       func NewEventSource(knot string) EventSource {

     

       43
       43
       +
       	return EventSource{

     

       44
       44
       +
       		Knot: knot,

     

       45
       45
       +
       	}

     

       27
       46
        
       }

     

       28
       47
        
       

     

       29
       48
        
       type EventConsumer struct {

     
···

       36
       55
        
       	randSource *rand.Rand

     

       37
       56
        
       }

     

       38
       57
        
       

     

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

     

       59
       59
       +
       	scheme := "wss"

     

       60
       60
       +
       	if e.cfg.Dev {

     

       61
       61
       +
       		scheme = "ws"

     

       62
       62
       +
       	}

     

       63
       63
       +
       

     

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

     

       65
       65
       +
       	if err != nil {

     

       66
       66
       +
       		return nil, err

     

       67
       67
       +
       	}

     

       68
       68
       +
       

     

       69
       69
       +
       	if cursor != "" {

     

       70
       70
       +
       		query := url.Values{}

     

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

     

       72
       72
       +
       		u.RawQuery = query.Encode()

     

       73
       73
       +
       	}

     

       74
       74
       +
       	return u, nil

     

       75
       75
       +
       }

     

       76
       76
       +
       

     

       39
       77
        
       type job struct {

     

       40
       40
       -
       	source  string

     

       78
       78
       +
       	source  EventSource

     

       41
       79
        
       	message []byte

     

       42
       80
        
       }

     

       43
       81
        
       

     
···

       104
       142
        
       			if !ok {

     

       105
       143
        
       				return

     

       106
       144
        
       			}

     

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

     

       145
       145
       +
       

     

       146
       146
       +
       			var msg Message

     

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

     

       148
       148
       +
       			if err != nil {

     

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

     

       150
       150
       +
       				return

     

       151
       151
       +
       			}

     

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

     

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

     

       109
       154
        
       			}

     

       110
       155
        
       		}

     

       111
       156
        
       	}

     

       112
       157
        
       }

     

       113
       158
        
       

     

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

     

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

     

       115
       160
        
       	defer c.wg.Done()

     

       116
       161
        
       	retryInterval := c.cfg.RetryInterval

     

       117
       162
        
       	for {

     
···

       144
       189
        
       	}

     

       145
       190
        
       }

     

       146
       191
        
       

     

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

     

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

     

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

     

       149
       194
        
       	defer cancel()

     

       150
       195
        
       

     

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

     

       197
       197
       +
       

     

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

     

       152
       199
        
       	if err != nil {

     

       153
       200
        
       		return err

     

       154
       201
        
       	}

     

       155
       202
        
       

     

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

     

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

     

       157
       205
        
       	if err != nil {

     

       158
       206
        
       		return err