update

Author: vm-001

pkg/queue/queue.go

@@ -11,7 +11,7 @@ type Message struct {
 	WorkspaceID string
 }
 
-type HandleFunc func(ctx context.Context, messages []*Message) error
+type HandlerFunc func(ctx context.Context, messages []*Message) error
 
 type Queue interface {
 	Producer
@@ -20,9 +20,9 @@ type Queue interface {
 }
 
 type Producer interface {
-	WriteMessage(ctx context.Context, message *Message) error
+	Enqueue(ctx context.Context, message *Message) error
 }
 
 type Consumer interface {
-	StartListen(ctx context.Context, handle HandleFunc)
+	StartListen(ctx context.Context, handler HandlerFunc)
 }

pkg/queue/redis/redis.go

@@ -41,7 +41,7 @@ func NewRedisQueue(opts Options, logger *zap.SugaredLogger) (queue.Queue, error)
 	return q, nil
 }
 
-func (q *RedisQueue) WriteMessage(ctx context.Context, message *queue.Message) error {
+func (q *RedisQueue) Enqueue(ctx context.Context, message *queue.Message) error {
 	ctx, span := tracing.Start(ctx, "redis.queue.enqueue", trace.WithSpanKind(trace.SpanKindServer))
 	defer span.End()
 
@@ -117,15 +117,15 @@ func (q *RedisQueue) delete(ctx context.Context, xmessages []redis.XMessage) err
 	return err
 }
 
-func (q *RedisQueue) StartListen(ctx context.Context, handle queue.HandleFunc) {
+func (q *RedisQueue) StartListen(ctx context.Context, handler queue.HandlerFunc) {
 	q.log.Infof("starting %d listeners", q.opts.Listeners)
 	for i := 0; i < q.opts.Listeners; i++ {
-		go q.listen(ctx, handle)
+		go q.listen(ctx, handler)
 	}
 	go q.process(ctx)
 }
 
-func (q *RedisQueue) listen(ctx context.Context, handle queue.HandleFunc) {
+func (q *RedisQueue) listen(ctx context.Context, handler queue.HandlerFunc) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -146,7 +146,7 @@ func (q *RedisQueue) listen(ctx context.Context, handle queue.HandleFunc) {
 				messages = append(messages, toMessage(msg.Values))
 			}
 
-			err = handle(ctx, messages)
+			err = handler(ctx, messages)
 			if err != nil {
 				q.log.Warnf("failed to handle message: %v", err)
 				continue

proxy/gateway.go

@@ -294,7 +294,7 @@ func (gw *Gateway) ingestEvent(ctx context.Context, async bool, event *entities.
 			Time:        time.Now(),
 			WorkspaceID: event.WorkspaceId,
 		}
-		return gw.queue.WriteMessage(ctx, &msg)
+		return gw.queue.Enqueue(ctx, &msg)
 	}
 
 	return gw.dispatch(ctx, []*entities.Event{event})