package main

import (
	"log"
	"sync"

	"code.justin.tv/release/trace/pbmsg"
	"github.com/golang/protobuf/proto"
)

// encodeTransactions converts a channel of pbmsg.Transaction structs
// into a channel of protobuf-encoded byte slices, using nWorkers
// goroutines to do so. Blocks until the input channel closes and all
// workers have successfully passed all their messages into the output
// channel.
func encodeTransactions(in <-chan *pbmsg.Transaction, out chan<- []byte, nWorkers int) {
	var wg sync.WaitGroup
	for i := 0; i < nWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for t := range in {
				encoded, err := proto.Marshal(t)
				if err != nil {
					log.Fatalf("transaction marshal err=%q", err)
				}
				out <- encoded
			}
		}()
	}
	wg.Wait()
}
