example-go-concunrrecy-generator-func

channel-breaking-out-from-select-and-wait.go

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int)
	ch2 := make(chan int)

	go func() {
		for i := 0; i < 5; i++ {
			ch <- i
		}
		close(ch)
	}()

	go func() {
		for i := 0; i < 5; i++ {
			ch2 <- i
		}
		close(ch2)
	}()

loop:
	for {
		select {
		case val, ok := <-ch:
			if ok {
				fmt.Println("[ch] Received value:", val)
			} else {
				fmt.Println("[ch] Channel closed")
				ch = nil

			}
		case val, ok := <-ch2:
			if ok {
				fmt.Println("[ch2] Received value:", val)
			} else {
				fmt.Println("[ch2] Channel closed")
				ch2 = nil

			}
		default:
			if ch == nil && ch2 == nil {
				break loop
			}
		}
	}

	fmt.Println("🔴 end program...")
}

generator.go

package main

import (
	"fmt"
	"sync"
)

func generator(count int) chan int {
	ch := make(chan int)
	go func() {
		for i := 0; i < count; i++ {
			ch <- i
		}
		close(ch)
	}()
	return ch
}

func main() {

	pipelineForPub := make(chan int)

	ch := generator(100)
	ch2 := generator(100)

	wg := sync.WaitGroup{}

	go func() {
		for data := range pipelineForPub {
			fmt.Printf("🔥 data :: %+v\n ", data)
			wg.Done()

		}
	}()

loop:
	for {
		select {
		case val, ok := <-ch:
			if ok {
				// fmt.Println("[ch] Received value:", val)
				wg.Add(1)
				pipelineForPub <- val
			} else {
				fmt.Println("[ch] Channel closed")
				ch = nil

			}
		case val, ok := <-ch2:
			if ok {
				// fmt.Println("[ch2] Received value:", val)
				wg.Add(1)
				pipelineForPub <- val
			} else {
				fmt.Println("[ch2] Channel closed")
				ch2 = nil

			}
		default:
			if ch == nil && ch2 == nil {
				break loop
			}
		}
	}
	fmt.Println("🔴 wait program...")
	wg.Wait()
	fmt.Println("🔴 end program...")
}

multiple-channel-wait.-consume.go

package main

import (
	"fmt"
	"sync"
)

func main() {
	ch := make(chan int)
	ch2 := make(chan int)

	pipelineForPub := make(chan int)

	go func() {
		for i := 0; i < 5; i++ {
			ch <- i
		}
		close(ch)
	}()

	go func() {
		for i := 0; i < 20; i++ {
			ch2 <- i
		}
		close(ch2)
	}()

	wg := sync.WaitGroup{}

	go func() {

		for data := range pipelineForPub {
			fmt.Printf("🔥 data :: %+v\n ", data)
			wg.Done()

		}

	}()

loop:
	for {
		select {
		case val, ok := <-ch:
			if ok {
				// fmt.Println("[ch] Received value:", val)
				wg.Add(1)
				pipelineForPub <- val
			} else {
				fmt.Println("[ch] Channel closed")
				ch = nil

			}
		case val, ok := <-ch2:
			if ok {
				// fmt.Println("[ch2] Received value:", val)
				wg.Add(1)
				pipelineForPub <- val
			} else {
				fmt.Println("[ch2] Channel closed")
				ch2 = nil

			}
		default:
			if ch == nil && ch2 == nil {
				break loop
			}
		}
	}
	fmt.Println("🔴 wait program...")
	wg.Wait()
	fmt.Println("🔴 end program...")
}

simple-how-to-breaking.go

package main

import "fmt"

func readContent() chan int {
	c := make(chan int)

	go func() {
		defer func() {
			close(c) // close channel when read all data
		}()
		for i := 0; i < 1000; i++ {
			if i == 10 {
				fmt.Println("🌕 skip index ", i)
				continue
			}
			c <- i
		}
	}()
	return c
}

func main() {

	fmt.Println("🟢 start ")
	receiveChannelContent := readContent()
	count := 0
	for {
		select {
		case data, ok := <-receiveChannelContent:
			if !ok {
				fmt.Println("🚀 close channel")
				receiveChannelContent = nil
				break
			}
			fmt.Printf("🔥 data :: %+v\n ", data)
			count++
		}

		// ensure receiveChannelContent close channel break for {} loop
		if receiveChannelContent == nil {
			break
		}

	}
	fmt.Println("✅ summary receiveChannelContent :: ", count) // actual 999 because skip i = 10
	fmt.Println("🎉 end program")
}