diff --git a/interleaver/main.go b/interleaver/main.go index 656023e..9efbf98 100644 --- a/interleaver/main.go +++ b/interleaver/main.go @@ -49,17 +49,32 @@ func selectKeyEncoder(interleaved bool) common.KeyEncoder { } func copyProducerToConsumer(producer BlockProducer, consumer BlockConsumer) (err error) { - var block Block - i := 0 - for { - if err = producer.Next(&block); err != nil { - if err == ErrNoMoreBlocks { - err = nil - break + + blocks := make(chan *Block) + done := make(chan struct{}) + defer close(done) + + go func() { + defer close(blocks) + for { + block := new(Block) + if err = producer.Next(block); err != nil { + if err != ErrNoMoreBlocks { + log.Printf("Reading failed: %s\n", err) + } + return + } + select { + case blocks <- block: + case <-done: + return } - return } - if err = consumer.Consume(&block); err != nil { + }() + + i := 0 + for block := range blocks { + if err = consumer.Consume(block); err != nil { return } i++