signer

package main

import (
    "fmt"
    "sort"
    "strconv"
    "strings"
    "sync"
)

type Item struct {
    Index int
    Part  int
    Val   string
}

const numberCount = 7

func ExecutePipeline(jobs ...job) {

    in := make(chan interface{})
    var wg sync.WaitGroup
    for i := range jobs {
        jobFunc := jobs[i]
        out := make(chan interface{})
        wg.Add(1)
        go func(in, out chan interface{}) {
            defer wg.Done()
            jobFunc(in, out)
            close(out)
        }(in, out)
        in = out
    }
    wg.Wait()
}

func SingleHash(in, out chan interface{}) {
    workerIn := make(chan Item)
    workerOut := make(chan Item)

    var workerWg sync.WaitGroup
    for i := 0; i < numberCount*2; i++ {
        workerWg.Add(1)
        go func() {
            defer workerWg.Done()
            startWorker(workerIn, workerOut, DataSignerCrc32)
        }()
    }

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        result := make(map[int][]string)
        for it := range workerOut {
            parts, ok := result[it.Index]
            if !ok {
                parts = make([]string, 2)
                parts[it.Part] = it.Val
                result[it.Index] = parts
                continue
            }
            parts[it.Part] = it.Val
            out <- strings.Join(parts, "~")
        }
    }()

    var ind int
    for raw := range in {

        data := fmt.Sprintf("%v", raw)

        md5 := DataSignerMd5(data)

        workerIn <- Item{Index: ind, Part: 0, Val: data}
        workerIn <- Item{Index: ind, Part: 1, Val: md5}
        ind++
    }
    close(workerIn)
    workerWg.Wait()
    close(workerOut)
    wg.Wait()
}

func MultiHash(in, out chan interface{}) {

    workerIn := make(chan Item)
    workerOut := make(chan Item)
    var workerWg sync.WaitGroup
    for i := 0; i < numberCount*6; i++ {
        workerWg.Add(1)
        go func() {
            defer workerWg.Done()
            startWorker(workerIn, workerOut, DataSignerCrc32)
        }()
    }

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        result := make(map[int][]Item)

        for it := range workerOut {
            parts, ok := result[it.Index]
            if !ok {
                parts = make([]Item, 0, 6)
            }
            parts = append(parts, it)
            result[it.Index] = parts

            if len(parts) == 6 {
                sort.Slice(parts, func(i, j int) bool { return parts[i].Part < parts[j].Part })
                sb := strings.Builder{}
                for _, s := range parts {
                    sb.WriteString(s.Val)
                }
                out <- sb.String()
            }
        }
    }()

    var ind int
    for raw := range in {

        data := fmt.Sprintf("%v", raw)
        for th := 0; th <= 5; th++ {
            workerIn <- Item{Index: ind, Part: th, Val: strconv.Itoa(th) + data}
        }
        ind++
    }

    close(workerIn)
    workerWg.Wait()
    close(workerOut)
    wg.Wait()

}

func CombineResults(in, out chan interface{}) {
    ss := make([]string, 0, 100)
    for raw := range in {
        ss = append(ss, fmt.Sprintf("%v", raw))
    }
    sort.StringSlice(ss).Sort()
    out <- strings.Join(ss, "_")
}

func startWorker(in <-chan Item, out chan<- Item, workFunc func(s string) string) {
    for it := range in {
        it.Val = workFunc(it.Val)
        out <- it
    }
}