| 152 | } |
| 153 | |
| 154 | func NewCompressedInput(opener func(fn string) (io.ReadCloser, int64, time.Time, *url.URL, error), sizer func(fn string) (int64, error), done chan bool) *CompressedInput { |
| 155 | s := &CompressedInput{ |
| 156 | Opener: opener, |
| 157 | Sizer: sizer, |
| 158 | Done: done, |
| 159 | stats: newInputStats(), |
| 160 | files: make(chan string, 1024), |
| 161 | stopNow: make(chan struct{}), |
| 162 | pool: sync.Pool{ |
| 163 | New: func() interface{} { |
| 164 | return &baker.Data{Bytes: make([]byte, kChunkBuffer)} |
| 165 | }, |
| 166 | }, |
| 167 | } |
| 168 | |
| 169 | // Start workers, that will read incoming files in the queue |
| 170 | // and process them. |
| 171 | var wg sync.WaitGroup |
| 172 | for i := 0; i < 4; i++ { |
| 173 | wg.Add(1) |
| 174 | go func() { |
| 175 | defer wg.Done() |
| 176 | s.worker() |
| 177 | }() |
| 178 | } |
| 179 | |
| 180 | go func() { |
| 181 | wg.Wait() |
| 182 | close(s.Done) |
| 183 | }() |
| 184 | |
| 185 | return s |
| 186 | } |
| 187 | |
| 188 | func (s *CompressedInput) worker() { |
| 189 | // Process incoming files on the s.files channel. |