Паттерны многопоточности: Генератор, Стоп-Кран, Паттерн обработки ошибок

package main

import "fmt"

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

    // чтение из канала: 0, false
    read, open := <-ch

    // отправка в канал: panic
    ch <- 1

    // эта строка никогда не выполнится, так как программа будет паниковать
    fmt.Printf("Прочитанные данные: %d, Канал открыт? %t", read, open)
} 

package main

import "fmt"

func main() {
    // создаём канал
    ch := make(chan int)

    // вызываем горутину отправителя
    go sender(ch)

    // вызываем получателя
    recipient(ch)
}

// sender отправляет в канал числа от 0 до 9
func sender(ch chan int) {
    for i := 0; i < 10; i++ {
        ch <- i
    }

    // закрываем канал после отправки
    close(ch)
}

// recipient забирает из канала значения и выводит на экран, 
// когда канал закрыт, выходит из цикла и завершает функцию
func recipient(ch chan int) {
    // читаем данные из канала, пока он открыт
    for data := range ch {
        // и выводим их на экран
        fmt.Println(data)
    }
} 

package main

import "fmt"

func main() {
    // данные в слайсе, которые будем отправлять
    input := []int{1, 2, 3, 4, 5, 6}

    // получаем канал с данными из генератора
    inputCh := generator(input)

    // отправляем данные потребителю через канал inputCh
    consumer(inputCh)
}

// generator — генератор, который создает канал и сразу возвращает его
func generator(input []int) chan int {
    inputCh := make(chan int)

    // через отдельную горутину генератор отправляет данные в канал
    go func() {
        // закрываем канал по завершению горутины — это отправитель
        defer close(inputCh)

        // перебираем данные в слайсе
        for _, data := range input {
            // отправляем данные в канал inputCh
            inputCh <- data
        }
    }()

    // возвращаем канал inputCh
    return inputCh
}

// consumer — потребитель проходит через канал и одновременно обрабатывает 
// данные из него (выводит на экран)
func consumer(inputCh chan int) {
    for data := range inputCh {
        fmt.Println(data)
    }
} 

package main

import (
    "errors"
    "log"
    "time"
)

func main() {
    input := []int{1, 2, 3, 4}

    // генератор возвращает канал, который потом передаётся получателю
    inputCh := generator(input)

    // получатель, в который передаём канал
    go consumer(inputCh)

    // добавим секунду сна, чтобы выводились сообщения ошибки
    time.Sleep(time.Second)
}

// generator генерирует данные и отправляет их в канал inputCh, который потом закрывает, потому что он отправитель
func generator(input []int) chan int {
    inputCh := make(chan int)

    go func() {
        defer close(inputCh)

        for _, data := range input {
            inputCh <- data
        }
    }()
    return inputCh
}

// consumer читает данные из канала, отправляет в функцию, которая возвращает ошибку
func consumer(ch chan int) {
    // читаем данные из канала ch, пока он открыт
    for data := range ch {
        // получаем ошибку
        err := callDatabase(data)
        if err != nil {
            // не самый лучший способ обработки ошибок — вывод на экран
            log.Println(err)
        }
    }
}

// callDatabase функция обращения к базе данных, которая всегда возвращает ошибку
func callDatabase(data int) error {
    return errors.New("ошибка запроса к базе данных")
} 

package main

import (
    "fmt"
    "errors"
)

// структура, в которую добавили ошибку
type Result struct {
    data int
    err error
}

func main() {
    // ваши данные
    input := []int{1, 2, 3, 4}

    // канал с результатами работы функции consumer
    resultCh := make(chan Result)

    // получаем канал с данными из генератора
    inputCh := generator(input)

    // порождаем горутину которая отправляет результат в resultCh вместе с ошибкой
    go consumer(inputCh, resultCh)

    // читаем результаты из канала resultCh
    for res := range resultCh {
        if res.err != nil {
            // здесь обрабатываем ошибку как обычно в Go
            log.Println("разберемся с ошибкой здесь")
        }
    }
}

// consumer вызывает другую функцию, которая возвращает ошибку
func consumer(inputCh chan int, resultCh chan Result) {
    // закрваем resultCh при завершении функции consumer
    defer close(resultCh)

    // перебираем данные из канала inputCh
    for data := range inputCh {
        // получаем ошибку
        resp, err := callDatabase(data)

        // создаем структуру
        result := Result{
            data: resp,
            err: err,
        }

        // отправляем структуру в канал
        resultCh <- result
    }
}

// generator отправляет данные в канал inputCh
func generator(input []int) chan int {
    // создаём канал, куда будем отправлять данные из слайса
    inputCh := make(chan int)

    // через отдельную горутину генератор отправляет данные в канал
    go func() {
        defer close(inputCh)

        // перебираем данные из слайса
        for _, data := range input {
            // отправляем данные в канал
            inputCh <- data
        }
    }()

    // возвращаем канал с данными
    return inputCh
}


// callDatabase просто возвращает ошибку как бы из функции обращения к базе данных
func callDatabase(data int) (int, error) {
    return data, errors.New("ошибка запроса к базе данных")
} 

package main

import (
    "context"
    "errors"
    "log"

    "golang.org/x/sync/errgroup"
)

func main() {
    // создаём переменную errgroup
    g := new(errgroup.Group)

    // наши данные
    input := []int{1, 2, 3, 4}

    // генератор возвращает канал, через который он отправляет данные
    inputCh := generator(input)

    for data := range inputCh {
        // тут объявляем новую переменную внутри цикла, чтобы копировать переменную 
        // в замыкание каждой горутины, а не использовать одно общее на всех значение.
        data := data

        // потребитель должен возвращать ошибку.
        // сигнатура анонимной функции всегда такая как в примере.
        g.Go(func() error {
            // получаем ошибку
            err := callDatabase(data)
            if err != nil {
                // возвращаем ошибку
                return err
            }

            return nil
        })
    }

    // здесь ждём выполнения горутин, и если хотя бы в одной из них возникает ошибка, 
    // то присваиваем её err и обрабатываем. В этом случае просто выводим на экран.
    // Обратите внимание, что g.Wait() ждёт завершения всех запущенных горутин, даже 
    // если приозошла ошибка.
    if err := g.Wait(); err != nil {
        log.Println(err)
    }
}

// generator возвращает канал, а затем отправляет в него данные
func generator(input []int) chan int {
    // создаём канал данных
    inputCh := make(chan int)

    // вызываем горутину в которой отправляем данные в канал inputCh
    go func() {
        // по завершении горутины закрываем канал
        defer close(inputCh)

        // перебираем данные в слайсе
        for _, data := range input {
            // отправляем данные из слайса в канал
            inputCh <- data
        }
    }()

    // возвращаем канал с данными
    return inputCh
}

// callDatabase просто возвращает ошибку
func callDatabase(data int) error {
    // допустим ошибка возникнет когда data = 3
    if data == 3 {
        return errors.New("ошибка запроса к базе данных")
    }

    return nil
} 

package main

import (
    "fmt"
)

func main() {
    handler()
}

// handler получает данные из канала
func handler() {
    // слайс данных
    input := []int{1, 2, 3, 4, 5, 6}
    // получаем канал в который будут приходить данные из generator
    inputCh := generator(input)

    // перебираем данные из канала
    for data := range inputCh {
        // если среди данных из канала встретится 1, то выходим из handler
        if data == 1 {
            fmt.Println("Выходим из handler")
            return
        }
    }
}

// generator отправляет данные из слайса в канал, а потом его возвращает
func generator(input []int) chan int {
    inputCh := make(chan int)

    go func() {
        // по завершении закрываем канал inputCh
        defer close(inputCh)

        // передаём данные в канал inputCh
        for _, data := range input {
            inputCh <- data
        }
    }()

    // возвращаем канал inputCh
    return inputCh
} 

package main

import (
 "log"
 "time"
)

func main() {
    input := []int{1, 2, 3, 4, 5, 6}
    handler(input)
    time.Sleep(time.Second)
}

// handler получает данные из слайса
func handler(input []int) {
    // канал для явной отмены
    doneCh := make(chan struct{})
    // когда выходим из handler — сразу закрываем канал doneCh
    defer close(doneCh)

    // теперь передаём и канал отмены doneCh
    inputCh := generator(doneCh, input)

    // забираем данные из канала
    for data := range inputCh {
        // если в данных 3 — выходим из handler
        if data == 3 {
            log.Println("Прекращаем обработку данных из канала")
            return
        }
        log.Println(data)
    }
    log.Println("Данные во входном канале закончились")
}

// generator возвращает канал с данными
func generator(doneCh chan struct{}, input []int) chan int {
    // канал, в который будем отправлять данные из слайса
    inputCh := make(chan int)

    // горутина, в которой отправляются данные в канал inputCh
    go func() {
        // по завершении закрываем канал inputCh
        defer close(inputCh)

        // перебираем данные в слайсе input
        for _, data := range input {
            select {
            // если канал doneCh закрылся - сразу выходим из горутины
            case <-doneCh:
                log.Println("Останавливаем генератор")
                return
            // отправляем данные в канал inputCh
            case inputCh <- data:
            }
        }
    }()

    // возвращаем канал с данными
    return inputCh
} 

package main

import (
    "log"
    "context"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    input := []int{1, 2, 3, 4, 5, 6}

    go func() {
        handler(ctx, input)
        cancel()
    }

    time.Sleep(time.Seconds)
}

// передадим контекст и данные из слайса
func handler(ctx context.Context, input []int) {
    // передаём данные и контекст в генератор
    inputCh := generator(ctx, input)

    // теперь канал для отмены не нужен

    for data := range inputCh {
        if data == 3 {
            log.Println("Прекращаем обработку данных из канала")
            return
        }
        log.Println(data)
    }
    log.Println("Данные во входном канале закончились")
}

func generator(ctx context.Context, input []int) chan int {
    inputCh := make(chan int)

    go func() {
        defer close(inputCh)

        for _, data := range input {
            select {
            // вместо отменяющего канала используем Context.Done()
            case <-ctx.Done():
                log.Println("Останавливаем генератор")
                return
            case inputCh <- data:
            }
        }
    }()

    return inputCh
}