feat(boost): add alert func to control worker pool activity display (#12

)

boost/annotated-wait-group.go

@@ -100,9 +100,11 @@ func (a *waitGroupAnImpl) Wait(name ...GoRoutineName) {
 }
 
 func (a *waitGroupAnImpl) indicate(highlight, name, op string) {
-	fmt.Printf(
-		"		%v [[ WaitGroupAssister(%v).%v ]] - gr-name: '%v' (count: '%v') (running: '%v')\n",
-		highlight, a.waitGroupName, op, name, a.counter, a.running(),
+	Alert(
+		fmt.Sprintf(
+			"		%v [[ WaitGroupAssister(%v).%v ]] - gr-name: '%v' (count: '%v') (running: '%v')\n",
+			highlight, a.waitGroupName, op, name, a.counter, a.running(),
+		),
 	)
 }
 

boost/boost-public-api.go

@@ -62,3 +62,7 @@ func NewDuplex[T any](channel chan T) *Duplex[T] {
 		WriterCh: channel,
 	}
 }
+
+type ActivityCallback func(message string)
+
+var Alert ActivityCallback = func(message string) {}

boost/worker-pool.go

@@ -130,23 +130,27 @@ func (p *WorkerPool[I, O]) run(
 		p.private.resultOutCh <- r
 
 		p.WaitAQ.Done(p.RoutineName)
-		fmt.Printf("<--- WorkerPool.run (QUIT). 🧊🧊🧊\n")
+		Alert("<--- WorkerPool.run (QUIT). 🧊🧊🧊\n")
 	}(result)
-	fmt.Printf("===> 🧊 WorkerPool.run ...(ctx:%+v)\n", parentContext)
+	Alert(fmt.Sprintf(
+		"===> 🧊 WorkerPool.run ...(ctx:%+v)\n",
+		parentContext,
+	))
 
 	for running := true; running; {
 		select {
 		case <-parentContext.Done():
 			running = false
 
 			close(forwardChOut) // ⚠️ This is new
-			fmt.Println("===> 🧊 WorkerPool.run (source jobs chan closed) - done received ☢️☢️☢️")
+			Alert("===> 🧊 WorkerPool.run (source jobs chan closed) - done received ☢️☢️☢️")
 
 		case job, ok := <-p.sourceJobsChIn:
 			if ok {
-				fmt.Printf("===> 🧊 (#workers: '%v') WorkerPool.run - new job received\n",
+				Alert(fmt.Sprintf(
+					"===> 🧊 (#workers: '%v') WorkerPool.run - new job received",
 					len(p.private.pool),
-				)
+				))
 
 				if len(p.private.pool) < p.noWorkers {
 					p.spawn(parentContext,
@@ -159,17 +163,19 @@ func (p *WorkerPool[I, O]) run(
 				}
 				select {
 				case forwardChOut <- job:
-					fmt.Printf("===> 🧊 WorkerPool.run - forwarded job 🧿🧿🧿(%v) [Seq: %v]\n",
+					Alert(fmt.Sprintf(
+						"===> 🧊 WorkerPool.run - forwarded job 🧿🧿🧿(%v) [Seq: %v]",
 						job.ID,
 						job.SequenceNo,
-					)
+					))
 				case <-parentContext.Done():
 					running = false
 
 					close(forwardChOut) // ⚠️ This is new
-					fmt.Printf("===> 🧊 (#workers: '%v') WorkerPool.run - done received ☢️☢️☢️\n",
+					Alert(fmt.Sprintf(
+						"===> 🧊 (#workers: '%v') WorkerPool.run - done received ☢️☢️☢️",
 						len(p.private.pool),
-					)
+					))
 				}
 			} else {
 				// ⚠️ This close is essential. Since the pool acts as a bridge between
@@ -179,7 +185,7 @@ func (p *WorkerPool[I, O]) run(
 				//
 				running = false
 				close(forwardChOut)
-				fmt.Printf("===> 🚀 WorkerPool.run(source jobs chan closed) 🟥🟥🟥\n")
+				Alert("===> 🚀 WorkerPool.run(source jobs chan closed) 🟥🟥🟥")
 			}
 		}
 	}
@@ -191,14 +197,16 @@ func (p *WorkerPool[I, O]) run(
 	if err := p.drain(p.private.finishedCh); err != nil {
 		result.Error = err
 
-		fmt.Printf("===> 🧊 WorkerPool.run - drain complete with error: '%v' (workers count: '%v'). 📛📛📛\n",
+		Alert(fmt.Sprintf(
+			"===> 🧊 WorkerPool.run - drain complete with error: '%v' (workers count: '%v'). 📛📛📛",
 			err,
 			len(p.private.pool),
-		)
+		))
 	} else {
-		fmt.Printf("===> 🧊 WorkerPool.run - drain complete OK (workers count: '%v'). ☑️☑️☑️\n",
+		Alert(fmt.Sprintf(
+			"===> 🧊 WorkerPool.run - drain complete OK (workers count: '%v'). ☑️☑️☑️",
 			len(p.private.pool),
-		)
+		))
 	}
 }
 
@@ -222,14 +230,17 @@ func (p *WorkerPool[I, O]) spawn(
 
 	p.private.pool[w.core.id] = w
 	go w.core.run(parentContext, parentCancel, outputChTimeout)
-	fmt.Printf("===> 🧊 WorkerPool.spawned new worker: '%v' 🎀🎀🎀\n", w.core.id)
+	Alert(fmt.Sprintf(
+		"===> 🧊 WorkerPool.spawned new worker: '%v' 🎀🎀🎀",
+		w.core.id,
+	))
 }
 
 func (p *WorkerPool[I, O]) drain(finishedChIn finishedStreamR) error {
-	fmt.Printf(
-		"!!!! 🧊 WorkerPool.drain - waiting for remaining workers: %v (#GRs: %v); 🧊🧊🧊 \n",
+	Alert(fmt.Sprintf(
+		"!!!! 🧊 WorkerPool.drain - waiting for remaining workers: %v (#GRs: %v); 🧊🧊🧊",
 		len(p.private.pool), runtime.NumGoroutine(),
-	)
+	))
 
 	var firstError error
 
@@ -280,19 +291,21 @@ func (p *WorkerPool[I, O]) drain(finishedChIn finishedStreamR) error {
 		}
 
 		if workerResult.err != nil {
-			fmt.Printf("!!!! 🧊 WorkerPool.drain - worker (%v) 💢💢💢 finished with error: '%v'\n",
+			Alert(fmt.Sprintf(
+				"!!!! 🧊 WorkerPool.drain - worker (%v) 💢💢💢 finished with error: '%v'",
 				workerResult.id,
 				workerResult.err,
-			)
+			))
 
 			if firstError == nil {
 				firstError = workerResult.err
 			}
 		}
 
-		fmt.Printf("!!!! 🧊 WorkerPool.drain - worker-result-error(%v) finished, remaining: '%v' 🟥\n",
+		Alert(fmt.Sprintf(
+			"!!!! 🧊 WorkerPool.drain - worker-result-error(%v) finished, remaining: '%v' 🟥",
 			workerResult.err, len(p.private.pool),
-		)
+		))
 	}
 
 	return firstError

boost/worker-pool_test.go

@@ -303,7 +303,15 @@ type poolTE struct {
 	assert        assertFunc
 }
 
-var _ = Describe("WorkerPool", func() {
+func alertPrinter(message string) {
+	fmt.Println(message)
+}
+
+var _ = Describe("WorkerPool", Ordered, func() {
+	BeforeAll(func() {
+		boost.Alert = alertPrinter
+	})
+
 	DescribeTable("stream of jobs",
 		func(specContext SpecContext, entry *poolTE) {
 			defer leaktest.Check(GinkgoT())()

boost/worker.go

@@ -25,28 +25,39 @@ func (w *worker[I, O]) run(parentContext context.Context,
 	defer func(r *workerFinishedResult) {
 		w.finishedChOut <- r // ⚠️ non-pre-emptive send, but this should be ok
 
-		fmt.Printf("	<--- 🚀 worker.run(%v) (SENT FINISHED - error:'%v'). 🚀🚀🚀\n", w.id, r.err)
+		Alert(fmt.Sprintf("	<--- 🚀 worker.run(%v) (SENT FINISHED - error:'%v'). 🚀🚀🚀",
+			w.id, r.err,
+		))
 	}(&result)
 
-	fmt.Printf("	---> 🚀 worker.run(%v) ...(ctx:%+v)\n", w.id, parentContext)
+	Alert(
+		fmt.Sprintf("	---> 🚀 worker.run(%v) ...(ctx:%+v)\n", w.id, parentContext),
+	)
 
 	for running := true; running; {
 		select {
 		case <-parentContext.Done():
-			fmt.Printf("	---> 🚀 worker.run(%v)(finished) - done received 🔶🔶🔶\n", w.id)
+			Alert(fmt.Sprintf(
+				"	---> 🚀 worker.run(%v)(finished) - done received 🔶🔶🔶", w.id,
+			))
 
 			running = false
 		case job, ok := <-w.jobsChIn:
 			if ok {
-				fmt.Printf("	---> 🚀 worker.run(%v)(input:'%v')\n", w.id, job.Input)
+				Alert(fmt.Sprintf(
+					"	---> 🚀 worker.run(%v)(input:'%v')", w.id, job.Input,
+				))
+
 				err := w.invoke(parentContext, parentCancel, outputChTimeout, job)
 
 				if err != nil {
 					result.err = err
 					running = false
 				}
 			} else {
-				fmt.Printf("	---> 🚀 worker.run(%v)(jobs chan closed) 🟥🟥🟥\n", w.id)
+				Alert(fmt.Sprintf(
+					"	---> 🚀 worker.run(%v)(jobs chan closed) 🟥🟥🟥", w.id,
+				))
 
 				running = false
 			}
@@ -67,16 +78,22 @@ func (w *worker[I, O]) invoke(parentContext context.Context,
 	result, _ := w.exec(job)
 
 	if w.outputsChOut != nil {
-		fmt.Printf("	---> 🚀 worker.invoke ⏰ output timeout: '%v'\n", outputChTimeout)
+		Alert(fmt.Sprintf(
+			"	---> 🚀 worker.invoke ⏰ output timeout: '%v'", outputChTimeout,
+		))
 
 		select {
 		case w.outputsChOut <- result:
 
 		case <-parentContext.Done():
-			fmt.Printf("	---> 🚀 worker.invoke(%v)(cancel) - done received 💥💥💥\n", w.id)
+			Alert(fmt.Sprintf(
+				"	---> 🚀 worker.invoke(%v)(cancel) - done received 💥💥💥", w.id,
+			))
 
 		case <-outputContext.Done():
-			fmt.Printf("	---> 🚀 worker.invoke(%v)(cancel) - timeout on send 👿👿👿\n", w.id)
+			Alert(fmt.Sprintf(
+				"	---> 🚀 worker.invoke(%v)(cancel) - timeout on send 👿👿👿", w.id,
+			))
 
 			// ??? err = i18n.NewOutputChTimeoutError()
 			err = errors.New("timeout on send")