Merge pull request #1 from DumbMachine/feature/impl-get-updates

Minor docs update and added GetUpdates method

Author: DumbMachine

README.md

@@ -33,6 +33,63 @@ For successful cancellations it is important that `ProcessingFunc`, the function
 
 # Task Options Walkthrough
 
+## Watch for updates
+
+( Introduced in v0.3 )
+
+Listen for updates to task metadata
+
+```go
+func main() {
+	client := nq.NewPublishClient(nq.NatsClientOpt{
+		Addr: "nats://127.0.0.1:4222",
+	}, nq.NoAuthentcation(),
+	)
+
+	defer client.Close()
+
+	bytesPayload1, err := json.Marshal(UrlPayload{Url: "https://httpstat.us/200?sleep=10000"})
+	if err != nil {
+		panic(err)
+	}
+
+	var wg sync.WaitGroup
+	task1 := nq.NewTask(QueueDev, bytesPayload1)
+	if ack, err := client.Enqueue(task1); err == nil {
+		log.Printf("Watching updates queue=%s taskID=%s payload=%s", ack.Queue, ack.ID, ack.Payload)
+		wg.Add(1)
+		updates, err := client.GetUpdates(ack.ID)
+		if err != nil {
+			panic(err)
+		}
+		// listening for updates
+		go func() {
+			defer wg.Done()
+
+			for {
+				msg, ok := <-updates
+				if !ok {
+					// channel closed
+					return
+				}
+				log.Printf("Change detected, status=%s", msg.GetStatus())
+			}
+		}()
+	} else {
+		log.Printf("err=%s", err)
+	}
+	wg.Wait()
+}
+
+```
+
+```
+2022/08/29 22:17:15 Watching updates queue=scrap-url-dev taskID=yzaKwBIcbGEt8sMGgMJcZ0 payload={"url":"https://httpstat.us/200?sleep=10000"}
+2022/08/29 22:17:15 Change detected, status=pending
+2022/08/29 22:17:16 Change detected, status=processing
+2022/08/29 22:17:28 Change detected, status=completed
+```
+
 ## Retrying
 
 By default `task` is submitted for retry, if it returns non-nil error.

nq.go

@@ -1,7 +1,3 @@
-// Copyright 2022 Ratin Kumar. All rights reserved.
-// Use of this source code is governed by a MIT license
-// that can be found in the LICENSE file.
-
 // nq provides a go package to publish/process tasks via nats
 package nq
 
@@ -13,20 +9,16 @@ import (
 	"github.com/nats-io/nuid"
 )
 
+// Task is a representation work to be performed by a worker
 type Task struct {
 	// Stream subject
 	queue string
 	// Payload for task
 	payload []byte
-	opts    []TaskOption
-	// Result writer for retention
+	// Options to configure task processing behavior
+	opts []TaskOption
 }
 
-// !TODO: Remove this later
-// func (p *PublishClient) Subscribe(pattern string, handler ProcessingFunc, contextMap ContextStore) error {
-// 	// return p.broker.Subscribe(pattern, handler, contextMap)
-// }
-
 // Value zero indicates no timeout and no deadline.
 var (
 	noTimeout  time.Duration = 0
@@ -51,22 +43,27 @@ type TaskOption interface {
 	Value() interface{}
 }
 
-// Internal option representations.
 type (
 	retryOption    int
 	taskIDOption   string
 	timeoutOption  time.Duration
 	deadlineOption time.Time
 )
 
+// Returns an options to specify maximum number of times a task will be retried before being marked as failed.
+//
+// -ve retry count is assigned defaultRetry ( 0 )
 func Retry(n int) TaskOption {
+	if n < 0 {
+		return retryOption(defaultMaxRetry)
+	}
 	return retryOption(n)
 }
 func (n retryOption) String() string       { return fmt.Sprintf("MaxRetry(%d)", int(n)) }
 func (n retryOption) Type() TaskOptionType { return MaxRetryOpt }
 func (n retryOption) Value() interface{}   { return int(n) }
 
-// TaskID returns an option to specify the task ID.
+// TaskID returns an option to specify the task ID
 func TaskID(id string) TaskOption {
 	return taskIDOption(id)
 }
@@ -75,11 +72,11 @@ func (id taskIDOption) String() string       { return fmt.Sprintf("TaskID(%q)",
 func (id taskIDOption) Type() TaskOptionType { return TaskIDOpt }
 func (id taskIDOption) Value() interface{}   { return string(id) }
 
-// Timeout returns an option to specify how long a task may run.
+// Timeout returns an option to specify how long a task can run before being cancelled.
 //
-// Zero duration means no limit ( math.MaxInt64 is chosen )
+// Zero duration means no limit ( math.MaxInt32 )
 //
-// If there's a conflicting Deadline option, whichever comes earliest
+// If both Deadline and Timeout options are set, whichever comes earliest
 // will be used.
 func Timeout(d time.Duration) TaskOption {
 	return timeoutOption(d)
@@ -90,10 +87,8 @@ func (d timeoutOption) Type() TaskOptionType { return TimeoutOpt }
 func (d timeoutOption) Value() interface{}   { return time.Duration(d) }
 
 // Deadline returns an option to specify the deadline for the given task.
-// If it reaches the deadline before the Handler returns, then the task
-// will be retried.
 //
-// If there's a conflicting Timeout option, whichever comes earliest
+// If both Deadline and Timeout options are set, whichever comes earliest
 // will be used.
 func Deadline(t time.Time) TaskOption {
 	return deadlineOption(t)
@@ -116,7 +111,7 @@ type option struct {
 func withDefaultOptions(opts ...TaskOption) (option, error) {
 	res := option{
 		timeout:  0,
-		retry:    0,
+		retry:    defaultMaxRetry,
 		deadline: time.Time{},
 		// TODO: store generator per server
 		taskID: nuid.New().Next(),
@@ -144,7 +139,9 @@ func withDefaultOptions(opts ...TaskOption) (option, error) {
 	return res, nil
 }
 
-// payload is jsonified data of whatever the ProcessingFunc expects
+// NewTask returns a new Task given queue and byte payload
+//
+// TaskOption can be used to configure task processing
 func NewTask(queue string, payload []byte, opts ...TaskOption) *Task {
 	return &Task{
 		queue:   queue,

options.go

@@ -13,6 +13,7 @@ const (
 	defaultKVName                   = "nq-store"
 	defaultReconnectWait            = time.Second * 10
 	defaultAuthenticationType       = NoAuthenticationOpt
+	defaultMaxRetry                 = 0
 )
 
 var defaultNatsOptions = make([]nats.Option, 0)

publisher.go

@@ -4,19 +4,44 @@ import (
 	"context"
 	"encoding/json"
 	"log"
-
-	"github.com/nats-io/nats.go"
 )
 
 // Signature for function executed by a worker.
 // `ProcessingFunc` type are be registered to subjects, process messages published by client
 type ProcessingFunc func(context.Context, *TaskPayload) error
 
+type ListenUpdates struct {
+	m map[string]chan string
+}
+
+type Inspector struct {
+	broker    *NatsBroker
+	listeners *ListenUpdates
+}
+
+func NewInspector(broker *NatsBroker) *Inspector {
+	return &Inspector{
+		broker:    broker,
+		listeners: &ListenUpdates{make(map[string]chan string)},
+	}
+}
+
+func (i *Inspector) AddAnother(queue string, sendUpdatesTo chan string) error {
+	return nil
+}
+func (i *Inspector) Servers() {}
+func (i *Inspector) Queues()  {}
+
+// Client responsible for interaction with nq tasks
+//
+// Client is used to enqueue / cancel tasks or fetch metadata for tasks
 type PublishClient struct {
 	broker *NatsBroker
 	kv     ResultHandlerIFACE
 }
 
+// NewPublishClient returns a new Client instance, given nats connection options, to interact with nq tasks
+//
 func NewPublishClient(config NatsClientOpt, opts ...ClientConnectionOption) *PublishClient {
 	opt, err := withDefaultClientOptions(opts...)
 	if err != nil {
@@ -40,12 +65,6 @@ func NewPublishClient(config NatsClientOpt, opts ...ClientConnectionOption) *Pub
 	return &PublishClient{broker: broker, kv: kv}
 }
 
-type PackagePubAck struct {
-	// ID assigned to published message
-	ID string
-	*nats.PubAck
-}
-
 func (p *PublishClient) Stats(queue string) error {
 	return p.broker.Stats(NewQueue(queue))
 }
@@ -68,8 +87,7 @@ func (p *PublishClient) publishMessage(msg *TaskMessage) (*TaskMessage, error) {
 	}
 }
 
-// Publish a TaskMessage into a stream
-func (p *PublishClient) PublishToSubject(task *Task, opts ...TaskOption) (*TaskMessage, error) {
+func (p *PublishClient) publishToSubject(task *Task, opts ...TaskOption) (*TaskMessage, error) {
 	opts = append(task.opts, opts...)
 	opt, err := withDefaultOptions(opts...)
 	if err != nil {
@@ -98,14 +116,29 @@ func (p *PublishClient) PublishToSubject(task *Task, opts ...TaskOption) (*TaskM
 	return p.publishMessage(taskMessage)
 }
 
+// GetUpdates can be used get changes to a task's metadata
+//
+// Returns error if failed to start watching for changes
+// Channel is closed, once task reaches terminal state
+func (p *PublishClient) GetUpdates(taskID string) (chan *TaskMessage, error) {
+	if status, err := p.kv.Watch(taskID); err != nil {
+		return nil, err
+	} else {
+		return status, err
+	}
+}
+
+// Enqueue can be used to enqueu given task to a queue
+//
+// Returns TaskMessage and nil error is enqueued successfully, else non-nill error
 func (p *PublishClient) Enqueue(task *Task, opts ...TaskOption) (*TaskMessage, error) {
 	q := NewQueue(task.queue)
 	p.broker.ConnectoQueue(q)
 
-	return p.PublishToSubject(task, opts...)
+	return p.publishToSubject(task, opts...)
 }
 
-// Fetch qname from kv store instead
+// Cancel sends `cancel` request for given task to workers
 func (p *PublishClient) Cancel(id string) error {
 	if taskInfo, err := p.kv.Get(id); err != nil {
 		return ErrTaskNotFound
@@ -140,22 +173,31 @@ func (p *PublishClient) CancelInQueue(id string, qname string) error {
 	return p.cancelInStream(id, q)
 }
 
-func (p *PublishClient) createStream(streamName, subject string, policy nats.RetentionPolicy) error {
-	if err := p.broker.AddStream(nats.StreamConfig{
-		Name:      streamName,
-		Subjects:  []string{subject},
-		Retention: policy,
-	}); err != nil {
-		return err
-	}
-	return nil
-}
-
+// Delete a queue
+//
+// Deletes underlying nats stream assosociated with a queue
 func (p *PublishClient) DeleteQueue(qname string) {
 	q := NewQueue(qname)
+	// delete task stream
 	if err := p.broker.DeleteStream(q.stream); err != nil {
 		log.Printf("error deleting stream=%s", q.stream)
 	}
+	// delete cancellation stream
+	if err := p.broker.DeleteStream(q.cancelStream); err != nil {
+		log.Printf("error deleting stream=%s", q.stream)
+	}
+}
+
+// Fetch fetches TaskMessage for given task
+//
+func (p *PublishClient) Fetch(id string) (*TaskMessage, error) {
+	return p.kv.Get(id)
+}
+
+// Close closes the connection with nats
+func (p *PublishClient) Close() error {
+	defer p.broker.Close()
+	return nil
 }
 
 func (p *PublishClient) cancelInStream(id string, q *Queue) error {
@@ -174,13 +216,3 @@ func (p *PublishClient) cancelInStream(id string, q *Queue) error {
 		return nil
 	}
 }
-
-func (p *PublishClient) Fetch(id string) (*TaskMessage, error) {
-	return p.kv.Get(id)
-}
-
-// Also delete stream for cleanup
-func (p *PublishClient) Close() error {
-	defer p.broker.Close()
-	return nil
-}

utils.go

@@ -49,24 +49,8 @@ func CancelStreamNameToStreamName(stream, subject string) string {
 
 //
 //
-// Internally `Queue`s represent an abstraction over a nats stream -> subject
+// Internal `Queue`s represent an abstraction over a nats stream -> subject
 type Queue struct {
-	// // Name of this queue. User facing
-	// //
-	// // In nats terminalogy this corresponds to stream/subject
-	// name string
-
-	// streamName string
-
-	// // // Durable name for a stream subscription
-	// // // Used to maintain sequence number after reconnection to nats-server
-	// // durableName string
-
-	// cancelStreamName string
-
-	// // Name of cancel subject
-	// cancelName string
-
 	stream        string
 	subject       string
 	cancelStream  string
@@ -79,13 +63,6 @@ func NewQueue(name string) *Queue {
 		subject:       fmt.Sprintf("%s.task", name),
 		cancelStream:  fmt.Sprintf("%s/cancel", name),
 		cancelSubject: fmt.Sprintf("%s.cancel", name),
-		// // Stream and subject for recieving tasks
-		// streamName: fmt.Sprintf("stream/%s", name),
-		// name:       name,
-
-		// // Stream and subject for recieving cancellations
-		// cancelName:       fmt.Sprintf("cancel/%s", name),
-		// cancelStreamName: fmt.Sprintf("stream/cancel/%s", name),
 	}
 }