Merge pull request #53 from reugn/develop

Use generic transformation functions in Flows

Author: reugn

.github/workflows/build.yml

@@ -7,15 +7,18 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.17.x]
+        go-version: [1.19]
     steps:
     - name: Setup Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: ${{ matrix.go-version }}
+
     - name: Checkout code
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
+
     - name: Run coverage
       run: go test ./... -coverprofile=coverage.out -covermode=atomic
+
     - name: Upload coverage to Codecov
       run: bash <(curl -s https://codecov.io/bash)

.github/workflows/golangci-lint.yml

@@ -10,6 +10,8 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - name: Checkout code
+        uses: actions/checkout@v3
+
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v2
+        uses: golangci/golangci-lint-action@v3

README.md

@@ -20,11 +20,13 @@ Flow capabilities ([flow](https://github.com/reugn/go-streams/tree/master/flow)
 * Map
 * FlatMap
 * Filter
+* Reduce
 * PassThrough
 * Split
 * FanOut
 * RoundRobin
 * Merge
+* Flatten
 * Throttler
 * SlidingWindow
 * TumblingWindow

flow/filter.go

@@ -5,7 +5,7 @@ import (
 )
 
 // FilterPredicate represents a filter predicate (boolean-valued function).
-type FilterPredicate func(interface{}) bool
+type FilterPredicate[T any] func(T) bool
 
 // Filter filters incoming elements using a filter predicate.
 // If an element matches the predicate, the element is passed downstream.
@@ -16,22 +16,22 @@ type FilterPredicate func(interface{}) bool
 // [ -------- FilterPredicate -------- ]
 //
 // out -- 1 -- 2 ------------------ 5 --
-type Filter struct {
-	filterPredicate FilterPredicate
+type Filter[T any] struct {
+	filterPredicate FilterPredicate[T]
 	in              chan interface{}
 	out             chan interface{}
 	parallelism     uint
 }
 
 // Verify Filter satisfies the Flow interface.
-var _ streams.Flow = (*Filter)(nil)
+var _ streams.Flow = (*Filter[any])(nil)
 
 // NewFilter returns a new Filter instance.
 //
 // filterPredicate is the boolean-valued filter function.
 // parallelism is the flow parallelism factor. In case the events order matters, use parallelism = 1.
-func NewFilter(filterPredicate FilterPredicate, parallelism uint) *Filter {
-	filter := &Filter{
+func NewFilter[T any](filterPredicate FilterPredicate[T], parallelism uint) *Filter[T] {
+	filter := &Filter[T]{
 		filterPredicate: filterPredicate,
 		in:              make(chan interface{}),
 		out:             make(chan interface{}),
@@ -43,44 +43,44 @@ func NewFilter(filterPredicate FilterPredicate, parallelism uint) *Filter {
 }
 
 // Via streams data through the given flow
-func (f *Filter) Via(flow streams.Flow) streams.Flow {
+func (f *Filter[T]) Via(flow streams.Flow) streams.Flow {
 	go f.transmit(flow)
 	return flow
 }
 
 // To streams data to the given sink
-func (f *Filter) To(sink streams.Sink) {
+func (f *Filter[T]) To(sink streams.Sink) {
 	f.transmit(sink)
 }
 
 // Out returns an output channel for sending data
-func (f *Filter) Out() <-chan interface{} {
+func (f *Filter[T]) Out() <-chan interface{} {
 	return f.out
 }
 
 // In returns an input channel for receiving data
-func (f *Filter) In() chan<- interface{} {
+func (f *Filter[T]) In() chan<- interface{} {
 	return f.in
 }
 
-func (f *Filter) transmit(inlet streams.Inlet) {
-	for elem := range f.Out() {
-		inlet.In() <- elem
+func (f *Filter[T]) transmit(inlet streams.Inlet) {
+	for element := range f.Out() {
+		inlet.In() <- element
 	}
 	close(inlet.In())
 }
 
 // doStream discards items that don't match the filter predicate.
-func (f *Filter) doStream() {
+func (f *Filter[T]) doStream() {
 	sem := make(chan struct{}, f.parallelism)
 	for elem := range f.in {
 		sem <- struct{}{}
-		go func(e interface{}) {
+		go func(element T) {
 			defer func() { <-sem }()
-			if f.filterPredicate(e) {
-				f.out <- e
+			if f.filterPredicate(element) {
+				f.out <- element
 			}
-		}(elem)
+		}(elem.(T))
 	}
 	for i := 0; i < int(f.parallelism); i++ {
 		sem <- struct{}{}

flow/flat_map.go

@@ -5,7 +5,7 @@ import (
 )
 
 // FlatMapFunction represents a FlatMap transformation function.
-type FlatMapFunction func(interface{}) []interface{}
+type FlatMapFunction[T, R any] func(T) []R
 
 // FlatMap takes one element and produces zero, one, or more elements.
 //
@@ -14,22 +14,22 @@ type FlatMapFunction func(interface{}) []interface{}
 // [ -------- FlatMapFunction -------- ]
 //
 // out -- 1' - 2' -------- 4'- 4" - 5' -
-type FlatMap struct {
-	flatMapFunction FlatMapFunction
+type FlatMap[T, R any] struct {
+	flatMapFunction FlatMapFunction[T, R]
 	in              chan interface{}
 	out             chan interface{}
 	parallelism     uint
 }
 
 // Verify FlatMap satisfies the Flow interface.
-var _ streams.Flow = (*FlatMap)(nil)
+var _ streams.Flow = (*FlatMap[any, any])(nil)
 
 // NewFlatMap returns a new FlatMap instance.
 //
 // flatMapFunction is the FlatMap transformation function.
 // parallelism is the flow parallelism factor. In case the events order matters, use parallelism = 1.
-func NewFlatMap(flatMapFunction FlatMapFunction, parallelism uint) *FlatMap {
-	flatMap := &FlatMap{
+func NewFlatMap[T, R any](flatMapFunction FlatMapFunction[T, R], parallelism uint) *FlatMap[T, R] {
+	flatMap := &FlatMap[T, R]{
 		flatMapFunction: flatMapFunction,
 		in:              make(chan interface{}),
 		out:             make(chan interface{}),
@@ -41,44 +41,44 @@ func NewFlatMap(flatMapFunction FlatMapFunction, parallelism uint) *FlatMap {
 }
 
 // Via streams data through the given flow
-func (fm *FlatMap) Via(flow streams.Flow) streams.Flow {
+func (fm *FlatMap[T, R]) Via(flow streams.Flow) streams.Flow {
 	go fm.transmit(flow)
 	return flow
 }
 
 // To streams data to the given sink
-func (fm *FlatMap) To(sink streams.Sink) {
+func (fm *FlatMap[T, R]) To(sink streams.Sink) {
 	fm.transmit(sink)
 }
 
 // Out returns an output channel for sending data
-func (fm *FlatMap) Out() <-chan interface{} {
+func (fm *FlatMap[T, R]) Out() <-chan interface{} {
 	return fm.out
 }
 
 // In returns an input channel for receiving data
-func (fm *FlatMap) In() chan<- interface{} {
+func (fm *FlatMap[T, R]) In() chan<- interface{} {
 	return fm.in
 }
 
-func (fm *FlatMap) transmit(inlet streams.Inlet) {
-	for elem := range fm.Out() {
-		inlet.In() <- elem
+func (fm *FlatMap[T, R]) transmit(inlet streams.Inlet) {
+	for element := range fm.Out() {
+		inlet.In() <- element
 	}
 	close(inlet.In())
 }
 
-func (fm *FlatMap) doStream() {
+func (fm *FlatMap[T, R]) doStream() {
 	sem := make(chan struct{}, fm.parallelism)
 	for elem := range fm.in {
 		sem <- struct{}{}
-		go func(e interface{}) {
+		go func(element T) {
 			defer func() { <-sem }()
-			trans := fm.flatMapFunction(e)
-			for _, item := range trans {
+			result := fm.flatMapFunction(element)
+			for _, item := range result {
 				fm.out <- item
 			}
-		}(elem)
+		}(elem.(T))
 	}
 	for i := 0; i < int(fm.parallelism); i++ {
 		sem <- struct{}{}