add handler mux and enqueue client

Author: MohamedAklamaash

client.go

@@ -0,0 +1,79 @@
+package relay
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/MohamedAklamaash/relay/internal/base"
+	"github.com/MohamedAklamaash/relay/internal/rdb"
+	"github.com/google/uuid"
+)
+
+var (
+	ErrDuplicateTask  = rdb.ErrDuplicateTask
+	ErrTaskIDConflict = rdb.ErrTaskIDConflict
+)
+
+type Client struct {
+	rdb *rdb.RDB
+}
+
+func NewClient(r RedisConnOpt) *Client {
+	return &Client{rdb: rdb.New(r.MakeClient())}
+}
+
+func (c *Client) Close() error {
+	return c.rdb.Close()
+}
+
+func (c *Client) Enqueue(task *Task, opts ...Option) (*TaskInfo, error) {
+	return c.EnqueueContext(context.Background(), task, opts...)
+}
+
+func (c *Client) EnqueueContext(ctx context.Context, task *Task, opts ...Option) (*TaskInfo, error) {
+	if task == nil {
+		return nil, errors.New("relay: cannot enqueue nil task")
+	}
+	o := composeOptions(defaultOptions(), append(task.opts, opts...)...)
+
+	id := o.taskID
+	if id == "" {
+		id = uuid.NewString()
+	}
+
+	msg := &base.TaskMessage{
+		ID:         id,
+		Type:       task.typename,
+		Payload:    task.payload,
+		Queue:      o.queue,
+		MaxRetry:   o.maxRetry,
+		Timeout:    int64(o.timeout.Seconds()),
+		Retention:  int64(o.retention.Seconds()),
+		EnqueuedAt: time.Now().Unix(),
+	}
+	if !o.deadline.IsZero() {
+		msg.Deadline = o.deadline.Unix()
+	}
+	if o.uniqueTTL > 0 {
+		msg.UniqueHash = base.HashUnique(o.queue, task.typename, task.payload)
+	}
+
+	if err := c.rdb.Enqueue(ctx, msg, o.processAt, o.uniqueTTL); err != nil {
+		return nil, err
+	}
+
+	state := "pending"
+	if o.processAt.After(time.Now()) {
+		state = "scheduled"
+	}
+	return &TaskInfo{
+		ID:            msg.ID,
+		Queue:         msg.Queue,
+		Type:          msg.Type,
+		Payload:       msg.Payload,
+		State:         state,
+		MaxRetry:      msg.MaxRetry,
+		NextProcessAt: o.processAt,
+	}, nil
+}

handler.go

@@ -0,0 +1,52 @@
+package relay
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+type Handler interface {
+	ProcessTask(context.Context, *Task) error
+}
+
+type HandlerFunc func(context.Context, *Task) error
+
+func (f HandlerFunc) ProcessTask(ctx context.Context, t *Task) error {
+	return f(ctx, t)
+}
+
+type ServeMux struct {
+	mu sync.RWMutex
+	m  map[string]Handler
+}
+
+func NewServeMux() *ServeMux {
+	return &ServeMux{m: make(map[string]Handler)}
+}
+
+func (mux *ServeMux) Handle(pattern string, h Handler) {
+	mux.mu.Lock()
+	defer mux.mu.Unlock()
+	if pattern == "" {
+		panic("relay: invalid pattern")
+	}
+	if h == nil {
+		panic("relay: nil handler")
+	}
+	mux.m[pattern] = h
+}
+
+func (mux *ServeMux) HandleFunc(pattern string, fn func(context.Context, *Task) error) {
+	mux.Handle(pattern, HandlerFunc(fn))
+}
+
+func (mux *ServeMux) ProcessTask(ctx context.Context, t *Task) error {
+	mux.mu.RLock()
+	h, ok := mux.m[t.Type()]
+	mux.mu.RUnlock()
+	if !ok {
+		return fmt.Errorf("relay: no handler registered for %q", t.Type())
+	}
+	return h.ProcessTask(ctx, t)
+}