new object

Author: sjnam

README.md

@@ -17,7 +17,11 @@ import (
 )
 
 func main() {
-	inputStream := func() <-chan string {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	my := oproc.NewOrderedProc[string /*input param*/, string /*output param*/](ctx)
+	my.InputStream = func() <-chan string {
 		ch := make(chan string)
 		go func() {
 			defer close(ch)
@@ -28,20 +32,16 @@ func main() {
 			}
 		}()
 		return ch
-	}
-
-	doWork := func(str string) string {
+	}()
+	my.DoWork = func(str string) string {
 		// sleep instead of fetching
 		time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
 		return fmt.Sprintf("%s ... is fetched!", str)
 	}
 
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-
 	start := time.Now()
 
-	for s := range oproc.OrderedProc(ctx, inputStream(), doWork /*, 10 optional # of goroutines */) {
+	for s := range my.Process() {
 		fmt.Println(s)
 	}
 

examples/ctan/main.go

@@ -23,7 +23,11 @@ type item struct {
 }
 
 func main() {
-	inputStream := func() <-chan item {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	my := oproc.NewOrderedProc[item /*input param*/, item /*output param*/](ctx)
+	my.InputStream = func() <-chan item {
 		valStream := make(chan item)
 		go func() {
 			defer close(valStream)
@@ -43,9 +47,8 @@ func main() {
 			}
 		}()
 		return valStream
-	}
-
-	doWork := func(o item) item {
+	}()
+	my.DoWork = func(o item) item {
 		resp, err := http.Get("https://ctan.org/json/2.0/pkg/" + o.Key)
 		if err != nil {
 			log.Fatal(err)
@@ -57,11 +60,9 @@ func main() {
 		}
 		return o
 	}
+	my.Size = 20
 
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-
-	for s := range oproc.OrderedProc(ctx, inputStream(), doWork, 20) {
+	for s := range my.Process() {
 		b, err := json.Marshal(s)
 		if err == nil {
 			fmt.Println(string(b))

examples/ordered/main.go

@@ -10,7 +10,11 @@ import (
 )
 
 func main() {
-	inputStream := func() <-chan string {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	my := oproc.NewOrderedProc[string /*input param*/, string /*output param*/](ctx)
+	my.InputStream = func() <-chan string {
 		ch := make(chan string)
 		go func() {
 			defer close(ch)
@@ -21,20 +25,16 @@ func main() {
 			}
 		}()
 		return ch
-	}
-
-	doWork := func(str string) string {
+	}()
+	my.DoWork = func(str string) string {
 		// sleep instead of fetching
 		time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
 		return fmt.Sprintf("%s ... is fetched!", str)
 	}
 
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-
 	start := time.Now()
 
-	for s := range oproc.OrderedProc(ctx, inputStream(), doWork) {
+	for s := range my.Process() {
 		fmt.Println(s)
 	}
 

oproc.go

@@ -5,76 +5,80 @@ import (
 	"runtime"
 )
 
-func OrderedProc[T, V any](
-	ctx context.Context,
-	inStream <-chan V,
-	doWork func(V) T,
-	size ...int,
-) <-chan T {
-	lvl := runtime.NumCPU()
-	if len(size) > 0 {
-		lvl = size[0]
+type OrderedProc[TI, TO any] struct {
+	Ctx         context.Context
+	InputStream <-chan TI
+	DoWork      func(TI) TO
+	Size        int
+}
+
+func NewOrderedProc[TI, TO any](ctx context.Context) *OrderedProc[TI, TO] {
+	return &OrderedProc[TI, TO]{
+		Ctx:  ctx,
+		Size: runtime.NumCPU(),
 	}
+}
 
-	orDone := func(c <-chan T) <-chan T {
-		ch := make(chan T)
+func (o *OrderedProc[TI, TO]) Process() <-chan TO {
+	orDone := func(c <-chan TO) <-chan TO {
+		ch := make(chan TO)
 		go func() {
 			defer close(ch)
 			for {
 				select {
-				case <-ctx.Done():
+				case <-o.Ctx.Done():
 					return
 				case v, ok := <-c:
 					if !ok {
 						return
 					}
 					select {
 					case ch <- v:
-					case <-ctx.Done():
+					case <-o.Ctx.Done():
 					}
 				}
 			}
 		}()
 		return ch
 	}
 
-	chanchan := func() <-chan <-chan T {
-		chch := make(chan (<-chan T), lvl)
+	chanchan := func() <-chan <-chan TO {
+		chch := make(chan (<-chan TO), o.Size)
 		go func() {
 			defer close(chch)
-			for v := range inStream {
-				ch := make(chan T)
+			for v := range o.InputStream {
+				ch := make(chan TO)
 				chch <- ch
 
 				go func() {
 					defer close(ch)
-					ch <- doWork(v)
+					ch <- o.DoWork(v)
 				}()
 			}
 		}()
 		return chch
 	}
 
 	// bridge-channel
-	return func(chch <-chan <-chan T) <-chan T {
-		vch := make(chan T)
+	return func(chch <-chan <-chan TO) <-chan TO {
+		vch := make(chan TO)
 		go func() {
 			defer close(vch)
 			for {
-				var ch <-chan T
+				var ch <-chan TO
 				select {
 				case maybe, ok := <-chch:
 					if !ok {
 						return
 					}
 					ch = maybe
-				case <-ctx.Done():
+				case <-o.Ctx.Done():
 					return
 				}
 				for v := range orDone(ch) {
 					select {
 					case vch <- v:
-					case <-ctx.Done():
+					case <-o.Ctx.Done():
 					}
 				}
 			}

oproc_test.go

@@ -6,7 +6,11 @@ import (
 )
 
 func ExampleOrderedProc() {
-	inputStream := func() <-chan string {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	my := NewOrderedProc[string /*input param*/, string /*output param*/](ctx)
+	my.InputStream = func() <-chan string {
 		ch := make(chan string)
 		go func() {
 			defer close(ch)
@@ -15,16 +19,12 @@ func ExampleOrderedProc() {
 			}
 		}()
 		return ch
-	}
-
-	doWork := func(str string) string {
+	}()
+	my.DoWork = func(str string) string {
 		return fmt.Sprintf("line:%s", str)
 	}
 
-	ctx, cancel := context.WithCancel(context.TODO())
-	defer cancel()
-
-	for s := range OrderedProc(ctx, inputStream(), doWork) {
+	for s := range my.Process() {
 		fmt.Println(s)
 		// Output:
 		// line:0