ref(async): define seq no in job (#7)

async/pool-defs.go

@@ -15,8 +15,9 @@ const (
 //
 
 type Job[I any] struct {
-	ID    string
-	Input I
+	ID         string
+	Input      I
+	SequenceNo int
 }
 
 type ExecutiveFunc[I, R any] func(j Job[I]) (JobResult[R], error)

async/worker-pool.go

@@ -130,7 +130,10 @@ func (p *WorkerPool[I, R]) run(
 				}
 				select {
 				case forwardChOut <- job:
-					fmt.Printf("===> 🧊 WorkerPool.run - forwarded job 🧿🧿🧿(%v)\n", job.ID)
+					fmt.Printf("===> 🧊 WorkerPool.run - forwarded job 🧿🧿🧿(%v) [Seq: %v]\n",
+						job.ID,
+						job.SequenceNo,
+					)
 				case <-ctx.Done(): // ☣️☣️☣️ CHECK THIS, IT MIGHT BE INVALID
 					fmt.Printf("===> 🧊 (#workers: '%v') WorkerPool.run - done received ☢️☢️☢️\n",
 						len(p.private.pool),

async/worker-pool_test.go

@@ -33,30 +33,29 @@ const (
 	Delay        = 750
 )
 
-var audience = []string{
-	"👻 caspar",
-	"🧙 gandalf",
-	"😺 garfield",
-	"👺 gobby",
-	"👿 nick",
-	"👹 ogre",
-	"👽 paul",
-	"🦄 pegasus",
-	"💩 poo",
-	"🤖 rusty",
-	"💀 skeletor",
-	"🐉 smaug",
-	"🧛‍♀️ vampire",
-	"👾 xenomorph",
-}
+var (
+	audience = []string{
+		"👻 caspar",
+		"🧙 gandalf",
+		"😺 garfield",
+		"👺 gobby",
+		"👿 nick",
+		"👹 ogre",
+		"👽 paul",
+		"🦄 pegasus",
+		"💩 poo",
+		"🤖 rusty",
+		"💀 skeletor",
+		"🐉 smaug",
+		"🧛‍♀️ vampire",
+		"👾 xenomorph",
+	}
 
-type TestJobInput struct {
-	sequenceNo int // allocated by observer
-	Recipient  string
-}
+	noOp = func(_ context.Context, _ time.Duration, _ ...context.CancelFunc) {}
+)
 
-func (i TestJobInput) SequenceNo() int {
-	return i.sequenceNo
+type TestJobInput struct {
+	Recipient string
 }
 
 type TestJobResult = string
@@ -69,7 +68,7 @@ var greeter = func(j async.Job[TestJobInput]) (async.JobResult[TestJobResult], e
 
 	result := async.JobResult[TestJobResult]{
 		Payload: fmt.Sprintf("			---> 🍉🍉🍉 [Seq: %v] Hello: '%v'",
-			j.Input.SequenceNo(), j.Input.Recipient,
+			j.SequenceNo, j.Input.Recipient,
 		),
 	}
 
@@ -91,12 +90,8 @@ type pipeline[I, R any] struct {
 func start[I, R any]() *pipeline[I, R] {
 	pipe := &pipeline[I, R]{
 		resultsCh: make(chan async.JobResult[R], ResultChSize),
-		stop: func(_ context.Context, _ time.Duration, _ ...context.CancelFunc) {
-			// no-op
-		},
-		cancel: func(_ context.Context, _ time.Duration, _ ...context.CancelFunc) {
-			// no-op
-		},
+		stop:      noOp,
+		cancel:    noOp,
 	}
 
 	return pipe
@@ -160,13 +155,10 @@ var _ = Describe("WorkerPool", func() {
 				pipe := start[TestJobInput, TestJobResult]()
 
 				By("👾 WAIT-GROUP ADD(producer)")
-				sequence := 0
 				pipe.produce(ctx, func() TestJobInput {
 					recipient := rand.Intn(len(audience)) //nolint:gosec // trivial
-					sequence++
 					return TestJobInput{
-						sequenceNo: sequence,
-						Recipient:  audience[recipient],
+						Recipient: audience[recipient],
 					}
 				})
 
@@ -200,13 +192,11 @@ var _ = Describe("WorkerPool", func() {
 				cancellations := []context.CancelFunc{cancel}
 
 				By("👾 WAIT-GROUP ADD(producer)")
-				sequence := 0
 				pipe.produce(ctxCancel, func() TestJobInput {
 					recipient := rand.Intn(len(audience)) //nolint:gosec // trivial
-					sequence++
+
 					return TestJobInput{
-						sequenceNo: sequence,
-						Recipient:  audience[recipient],
+						Recipient: audience[recipient],
 					}
 				})
 

internal/helpers/test-producer.go

@@ -79,11 +79,15 @@ func (p *Producer[I, R]) run(ctx context.Context) {
 }
 
 func (p *Producer[I, R]) item(ctx context.Context) bool {
+	p.sequenceNo++
+	p.Count++
+
 	result := true
 	i := p.provider()
 	j := async.Job[I]{
-		ID:    fmt.Sprintf("JOB-ID:%v", uuid.NewString()),
-		Input: i,
+		ID:         fmt.Sprintf("JOB-ID:%v", uuid.NewString()),
+		Input:      i,
+		SequenceNo: p.sequenceNo,
 	}
 
 	fmt.Printf(">>>> ✨ producer.item, 🟠 waiting to post item: '%+v'\n", i)
@@ -96,7 +100,6 @@ func (p *Producer[I, R]) item(ctx context.Context) bool {
 
 	case p.JobsCh <- j:
 	}
-	p.Count++
 
 	if result {
 		fmt.Printf(">>>> ✨ producer.item, 🟢 posted item: '%+v'\n", i)