[go] common concurrency model [generic version]

Conclusion in 《go The way of language concurrency 》

package common

import "sync"

// Bridge  By accepting transmission chan Of chan, Pass the value back to ( This is to read one in order channel Will choose the next channel)
func Bridge(done <-chan interface{
    }, chanStream <-chan <-chan any) <-chan any {
    
	valStream := make(chan any)
	go func() {
    
		defer close(valStream)
		for {
    
			var stream <-chan any
			select {
    
			case mybeStream, ok := <-chanStream: // Read chanStream Medium channel
				if !ok {
    
					return
				}
				stream = mybeStream
			case <-done:
				return
			}
			for val := range OrDone(done, stream) {
     // Read channel Send the content back 
				select {
    
				case <-done:
					return
				case valStream <- val:
				}
			}
		}
	}()
	return valStream
}

// Tee  Read in Data and send it to two accepted at the same time channel
func Tee(done <-chan interface{
    }, in <-chan any) (_, _ <-chan any) {
    
	out1 := make(chan any)
	out2 := make(chan any)
	go func() {
    
		defer close(out1)
		defer close(out2)
		for v := range OrDone(done, in) {
    
			var out1, out2 = out1, out2 // Local version , Hide external variables 
			for i := 0; i < 2; i++ {
        // To ensure two channel Can be written. We use two writes 
				select {
    
				case <-done:
					return
				case out1 <- v:
					out1 = nil // Close copy after simultaneous write channel To block and prevent secondary writes 
				case out2 <- v:
					out2 = nil
				}
			}
		}
	}()
	return out1, out2
}

// OrDone  adopt done To control reading chan
func OrDone(done <-chan interface{
    }, c <-chan any) <-chan any {
    
	valStream := make(chan any)
	go func() {
    
		defer close(valStream)
		for {
    
			select {
    
			case <-done:
				return
			case v, ok := <-c:
				if ok == false {
    
					return
				}
				select {
     // It can be optimized 
				case valStream <- v:
				case <-done:
				}
			}
		}
	}()
	return valStream
}

// MyInteger  Integer types , For random number type conversion 
type MyInteger interface {
    
	~int | ~int32 | ~int64
}

// MyFloat  Floating point numbers , Can be used for addition, subtraction, multiplication and division 
type MyFloat interface {
    
	~float64 | ~float32
}

// Number  Can be used for addition, subtraction, multiplication and division 
type Number interface {
    
	MyFloat | MyInteger
}

// FanIn  From many channels Merge data into one channel
func FanIn(done <-chan interface{
    }, channels []<-chan any) <-chan any {
    
	var wg sync.WaitGroup
	multiplexedStream := make(chan any)
	multiplex := func(c <-chan any) {
    
		defer wg.Done()
		for i := range c {
    
			select {
    
			case <-done:
				return
			case multiplexedStream <- i:
			}
		}
	}
	wg.Add(len(channels))
	for _, c := range channels {
    
		go multiplex(c)
	}
	go func() {
    
		wg.Wait()
		close(multiplexedStream)
	}()
	return multiplexedStream
}

// Multiply  Multiplication 
func Multiply[V Number](done <-chan interface{
    }, valueStream <-chan V, multiplier V) <-chan V {
    
	results := make(chan V)
	go func() {
    
		defer close(results)
		for v := range valueStream {
    
			select {
    
			case <-done:
				return
			case results <- v * multiplier:
			}
		}
	}()
	return results
}

// Add  Add 
func Add[V Number](done <-chan interface{
    }, valueStream <-chan V, additive V) <-chan V {
    
	results := make(chan V)
	go func() {
    
		defer close(results)
		for v := range valueStream {
    
			select {
    
			case <-done:
				return
			case results <- v + additive:
			}
		}
	}()
	return results
}

// ToType  Explicitly convert to the corresponding type 
func ToType[T any](done <-chan interface{
    }, valueStream <-chan interface{
    }) <-chan T {
    
	stringStream := make(chan T)
	go func() {
    
		defer close(stringStream)
		for v := range valueStream {
    
			select {
    
			case <-done:
				return
			case stringStream <- v.(T):
			}
		}
	}()
	return stringStream
}

// PrimeFinder  Get and judge prime numbers 
func PrimeFinder[T MyInteger](done <-chan interface{
    }, intStream <-chan T) <-chan interface{
    } {
    
	results := make(chan interface{
    })
	go func() {
    
		defer close(results)
		for v := range intStream {
    
			select {
    
			case <-done:
				return
			default:
			}
			for i := T(2); i*i < v; i++ {
    
				if v%i == 0 {
    
					goto next
				}
			}
			results <- v
		next:
		}
	}()
	return results
}

// Take  Take out num End after number 
func Take(done <-chan interface{
    }, valueStream <-chan any, num int) <-chan any {
    
	results := make(chan any)
	go func() {
    
		defer close(results)
		for i := 0; i < num; i++ {
    
			select {
    
			case <-done:
				return
			case results <- <-valueStream:
			}
		}
	}()
	return results
}

// RepeatFn  Call the function repeatedly 
func RepeatFn(done <-chan interface{
    }, fn func() interface{
    }) <-chan interface{
    } {
    
	results := make(chan interface{
    })
	go func() {
    
		defer close(results)
		for {
    
			select {
    
			case <-done:
				return
			case results <- fn():
			}
		}
	}()
	return results
}

// Repeat  Duplicate generated value 
func Repeat(done <-chan interface{
    }, values ...any) <-chan any {
    
	valueStream := make(chan any)
	go func() {
    
		defer close(valueStream)
		for {
    
			for _, v := range values {
    
				select {
    
				case <-done:
					return
				case valueStream <- v:
				}
			}
		}
	}()
	return valueStream
}

The details are in Warehouse
There is no need to use generics in many places , Unless basically written as like as two peas of wheels, it is recommended to use for many types. , Advice to see go Official instruction course , I think this generic is OK , Not ugly