@@ -21,29 +21,29 @@ var (
2121
2222type (
2323 // ForEachFunc is used to do element processing, but no output.
24- ForEachFunc func (item interface {} )
24+ ForEachFunc [ T any ] func (item T )
2525 // GenerateFunc is used to let callers send elements into source.
26- GenerateFunc func (source chan <- interface {} )
26+ GenerateFunc [ T any ] func (source chan <- T )
2727 // MapFunc is used to do element processing and write the output to writer.
28- MapFunc func (item interface {} , writer Writer )
28+ MapFunc [ T , U any ] func (item T , writer Writer [ U ] )
2929 // MapperFunc is used to do element processing and write the output to writer,
3030 // use cancel func to cancel the processing.
31- MapperFunc func (item interface {} , writer Writer , cancel func (error ))
31+ MapperFunc [ T , U any ] func (item T , writer Writer [ U ] , cancel func (error ))
3232 // ReducerFunc is used to reduce all the mapping output and write to writer,
3333 // use cancel func to cancel the processing.
34- ReducerFunc func (pipe <- chan interface {} , writer Writer , cancel func (error ))
34+ ReducerFunc [ U , V any ] func (pipe <- chan U , writer Writer [ V ] , cancel func (error ))
3535 // VoidReducerFunc is used to reduce all the mapping output, but no output.
3636 // Use cancel func to cancel the processing.
37- VoidReducerFunc func (pipe <- chan interface {} , cancel func (error ))
37+ VoidReducerFunc [ U any ] func (pipe <- chan U , cancel func (error ))
3838 // Option defines the method to customize the mapreduce.
3939 Option func (opts * mapReduceOptions )
4040
41- mapperContext struct {
41+ mapperContext [ T , U any ] struct {
4242 ctx context.Context
43- mapper MapFunc
44- source <- chan interface {}
43+ mapper MapFunc [ T , U ]
44+ source <- chan T
4545 panicChan * onceChan
46- collector chan <- interface {}
46+ collector chan <- U
4747 doneChan <- chan struct {}
4848 workers int
4949 }
5454 }
5555
5656 // Writer interface wraps Write method.
57- Writer interface {
58- Write (v interface {} )
57+ Writer [ T any ] interface {
58+ Write (v T )
5959 }
6060)
6161
@@ -65,16 +65,15 @@ func Finish(fns ...func() error) error {
6565 return nil
6666 }
6767
68- return MapReduceVoid (func (source chan <- interface {} ) {
68+ return MapReduceVoid (func (source chan <- func () error ) {
6969 for _ , fn := range fns {
7070 source <- fn
7171 }
72- }, func (item interface {}, writer Writer , cancel func (error )) {
73- fn := item .(func () error )
72+ }, func (fn func () error , writer Writer [any ], cancel func (error )) {
7473 if err := fn (); err != nil {
7574 cancel (err )
7675 }
77- }, func (pipe <- chan interface {} , cancel func (error )) {
76+ }, func (pipe <- chan any , cancel func (error )) {
7877 }, WithWorkers (len (fns )))
7978}
8079
@@ -84,27 +83,26 @@ func FinishVoid(fns ...func()) {
8483 return
8584 }
8685
87- ForEach (func (source chan <- interface {} ) {
86+ ForEach (func (source chan <- func () ) {
8887 for _ , fn := range fns {
8988 source <- fn
9089 }
91- }, func (item interface {}) {
92- fn := item .(func ())
90+ }, func (fn func ()) {
9391 fn ()
9492 }, WithWorkers (len (fns )))
9593}
9694
9795// ForEach maps all elements from given generate but no output.
98- func ForEach (generate GenerateFunc , mapper ForEachFunc , opts ... Option ) {
96+ func ForEach [ T any ] (generate GenerateFunc [ T ] , mapper ForEachFunc [ T ] , opts ... Option ) {
9997 options := buildOptions (opts ... )
100- panicChan := & onceChan {channel : make (chan interface {} )}
98+ panicChan := & onceChan {channel : make (chan any )}
10199 source := buildSource (generate , panicChan )
102- collector := make (chan interface {} , options .workers )
100+ collector := make (chan any , options .workers )
103101 done := make (chan struct {})
104102
105- go executeMappers (mapperContext {
103+ go executeMappers (mapperContext [ T , any ] {
106104 ctx : options .ctx ,
107- mapper : func (item interface {} , writer Writer ) {
105+ mapper : func (item T , writer Writer [ any ] ) {
108106 mapper (item )
109107 },
110108 source : source ,
@@ -128,26 +126,26 @@ func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option) {
128126
129127// MapReduce maps all elements generated from given generate func,
130128// and reduces the output elements with given reducer.
131- func MapReduce (generate GenerateFunc , mapper MapperFunc , reducer ReducerFunc ,
132- opts ... Option ) (interface {} , error ) {
133- panicChan := & onceChan {channel : make (chan interface {} )}
129+ func MapReduce [ T , U , V any ] (generate GenerateFunc [ T ] , mapper MapperFunc [ T , U ], reducer ReducerFunc [ U , V ] ,
130+ opts ... Option ) (V , error ) {
131+ panicChan := & onceChan {channel : make (chan any )}
134132 source := buildSource (generate , panicChan )
135133 return mapReduceWithPanicChan (source , panicChan , mapper , reducer , opts ... )
136134}
137135
138136// MapReduceChan maps all elements from source, and reduce the output elements with given reducer.
139- func MapReduceChan (source <- chan interface {} , mapper MapperFunc , reducer ReducerFunc ,
140- opts ... Option ) (interface {} , error ) {
141- panicChan := & onceChan {channel : make (chan interface {} )}
137+ func MapReduceChan [ T , U , V any ] (source <- chan T , mapper MapperFunc [ T , U ], reducer ReducerFunc [ U , V ] ,
138+ opts ... Option ) (V , error ) {
139+ panicChan := & onceChan {channel : make (chan any )}
142140 return mapReduceWithPanicChan (source , panicChan , mapper , reducer , opts ... )
143141}
144142
145143// MapReduceChan maps all elements from source, and reduce the output elements with given reducer.
146- func mapReduceWithPanicChan (source <- chan interface {} , panicChan * onceChan , mapper MapperFunc ,
147- reducer ReducerFunc , opts ... Option ) (interface {}, error ) {
144+ func mapReduceWithPanicChan [ T , U , V any ] (source <- chan T , panicChan * onceChan , mapper MapperFunc [ T , U ] ,
145+ reducer ReducerFunc [ U , V ], opts ... Option ) (val V , err error ) {
148146 options := buildOptions (opts ... )
149147 // output is used to write the final result
150- output := make (chan interface {} )
148+ output := make (chan V )
151149 defer func () {
152150 // reducer can only write once, if more, panic
153151 for range output {
@@ -156,7 +154,7 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
156154 }()
157155
158156 // collector is used to collect data from mapper, and consume in reducer
159- collector := make (chan interface {} , options .workers )
157+ collector := make (chan U , options .workers )
160158 // if done is closed, all mappers and reducer should stop processing
161159 done := make (chan struct {})
162160 writer := newGuardedWriter (options .ctx , output , done )
@@ -192,9 +190,9 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
192190 reducer (collector , writer , cancel )
193191 }()
194192
195- go executeMappers (mapperContext {
193+ go executeMappers (mapperContext [ T , U ] {
196194 ctx : options .ctx ,
197- mapper : func (item interface {} , w Writer ) {
195+ mapper : func (item T , w Writer [ U ] ) {
198196 mapper (item , w , cancel )
199197 },
200198 source : source ,
@@ -207,24 +205,27 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
207205 select {
208206 case <- options .ctx .Done ():
209207 cancel (context .DeadlineExceeded )
210- return nil , context .DeadlineExceeded
208+ err = context .DeadlineExceeded
211209 case v := <- panicChan .channel :
212210 panic (v )
213211 case v , ok := <- output :
214- if err := retErr .Load (); err != nil {
215- return nil , err .(error )
212+ if e := retErr .Load (); e != nil {
213+ err = e .(error )
216214 } else if ok {
217- return v , nil
215+ val = v
218216 } else {
219- return nil , ErrReduceNoOutput
217+ err = ErrReduceNoOutput
220218 }
221219 }
220+
221+ return
222222}
223223
224224// MapReduceVoid maps all elements generated from given generate,
225225// and reduce the output elements with given reducer.
226- func MapReduceVoid (generate GenerateFunc , mapper MapperFunc , reducer VoidReducerFunc , opts ... Option ) error {
227- _ , err := MapReduce (generate , mapper , func (input <- chan interface {}, writer Writer , cancel func (error )) {
226+ func MapReduceVoid [T , U any ](generate GenerateFunc [T ], mapper MapperFunc [T , U ],
227+ reducer VoidReducerFunc [U ], opts ... Option ) error {
228+ _ , err := MapReduce (generate , mapper , func (input <- chan U , writer Writer [any ], cancel func (error )) {
228229 reducer (input , cancel )
229230 }, opts ... )
230231 if errors .Is (err , ErrReduceNoOutput ) {
@@ -261,8 +262,8 @@ func buildOptions(opts ...Option) *mapReduceOptions {
261262 return options
262263}
263264
264- func buildSource (generate GenerateFunc , panicChan * onceChan ) chan interface {} {
265- source := make (chan interface {} )
265+ func buildSource [ T any ] (generate GenerateFunc [ T ] , panicChan * onceChan ) chan T {
266+ source := make (chan T )
266267 go func () {
267268 defer func () {
268269 if r := recover (); r != nil {
@@ -278,13 +279,13 @@ func buildSource(generate GenerateFunc, panicChan *onceChan) chan interface{} {
278279}
279280
280281// drain drains the channel.
281- func drain (channel <- chan interface {} ) {
282+ func drain [ T any ] (channel <- chan T ) {
282283 // drain the channel
283284 for range channel {
284285 }
285286}
286287
287- func executeMappers (mCtx mapperContext ) {
288+ func executeMappers [ T , U any ] (mCtx mapperContext [ T , U ] ) {
288289 var wg sync.WaitGroup
289290 defer func () {
290291 wg .Wait ()
@@ -341,22 +342,21 @@ func once(fn func(error)) func(error) {
341342 }
342343}
343344
344- type guardedWriter struct {
345+ type guardedWriter [ T any ] struct {
345346 ctx context.Context
346- channel chan <- interface {}
347+ channel chan <- T
347348 done <- chan struct {}
348349}
349350
350- func newGuardedWriter (ctx context.Context , channel chan <- interface {},
351- done <- chan struct {}) guardedWriter {
352- return guardedWriter {
351+ func newGuardedWriter [T any ](ctx context.Context , channel chan <- T , done <- chan struct {}) guardedWriter [T ] {
352+ return guardedWriter [T ]{
353353 ctx : ctx ,
354354 channel : channel ,
355355 done : done ,
356356 }
357357}
358358
359- func (gw guardedWriter ) Write (v interface {} ) {
359+ func (gw guardedWriter [ T ] ) Write (v T ) {
360360 select {
361361 case <- gw .ctx .Done ():
362362 return
@@ -368,11 +368,11 @@ func (gw guardedWriter) Write(v interface{}) {
368368}
369369
370370type onceChan struct {
371- channel chan interface {}
371+ channel chan any
372372 wrote int32
373373}
374374
375- func (oc * onceChan ) write (val interface {} ) {
375+ func (oc * onceChan ) write (val any ) {
376376 if atomic .AddInt32 (& oc .wrote , 1 ) > 1 {
377377 return
378378 }
0 commit comments