src/cracha.go: Use PoolRunner() and Result type from gobang

1 files changed, 2 insertions, 93 deletions

diff --git a/src/cracha.go b/src/cracha.go
index 5547d6e..7a20449 100644
--- a/src/cracha.go
+++ b/src/cracha.go
@@ -1069,97 +1069,6 @@ func unregisterConsumers(queue q.IQueue, consumers []consumerT, prefix string) {
 	}
 }
 
-type resultT[T any] struct{
-	value T
-	err   error
-}
-
-type taggedT[T any] struct{
-	id    int
-	value T
-}
-
-func startRunner[A any, B any](
-	in  <-chan taggedT[A],
-	out chan<- taggedT[B],
-	fn func(A) B,
-	done func(),
-) {
-	for input := range in {
-		out <- taggedT[B]{
-			id:    input.id,
-			value: fn(input.value),
-		}
-	}
-	done()
-}
-
-func makePoolRunner[A any, B any](count int, fn func(A) B) (func(A) B, func()) {
-	var wg sync.WaitGroup
-	wg.Add(count)
-
-	in  := make(chan taggedT[A])
-	out := make(chan taggedT[B])
-
-	for _ = range count {
-		go startRunner(in, out, fn, wg.Done)
-	}
-
-	var mutex sync.Mutex
-	m := map[int]chan B{}
-	id := 0
-	go func() {
-		for output := range out {
-			mutex.Lock()
-			defer mutex.Unlock()
-			m[output.id] <- output.value
-			close(m[output.id])
-			delete(m, output.id)
-		}
-	}()
-
-	poolRunFn := func(input A) B {
-		c := make(chan B)
-		{
-			mutex.Lock()
-			defer mutex.Unlock()
-			m[id] = c
-			id++
-		}
-
-		in <- taggedT[A]{
-			id:    id,
-			value: input,
-		}
-		return <- c
-	}
-
-	close := func() {
-		close(in)
-		wg.Wait()
-		close(out)
-	}
-
-	return poolRunFn, close
-}
-
-func asResult[A any, B any](fn func(A) (B, error)) func(A) resultT[B] {
-	return func(input A) resultT[B] {
-		output, err := fn(input)
-		return resultT[B]{
-			value: output,
-			err:   err,
-		}
-	}
-}
-
-func unwrapResult[A any, B any](fn func(A) resultT[B]) func(A) (B, error) {
-	return func(input A) (B, error) {
-		result := fn(input)
-		return result.value, result.err
-	}
-}
-
 func NewWithPrefix(databasePath string, prefix string) (IAuth, error) {
 	queue, err := q.New(databasePath)
 	if err != nil {
@@ -1177,7 +1086,7 @@ func NewWithPrefix(databasePath string, prefix string) (IAuth, error) {
 	}
 
 	numCPU := runtime.NumCPU()
-	hasher, closeHasher := makePoolRunner(numCPU, asResult(scrypt.Hash))
+	hasher, closeHasher := g.PoolRunner(numCPU, g.Resultify(scrypt.Hash))
 
 	closeFn := func() {
 		unregisterConsumers(queue, consumers, prefix)
@@ -1188,7 +1097,7 @@ func NewWithPrefix(databasePath string, prefix string) (IAuth, error) {
 		queue:   queue,
 		db:      db,
 		queries: queries,
-		hasher:  unwrapResult(hasher),
+		hasher:  g.Unresultify(hasher),
 		close:   closeFn,
 	}