| 225 | } |
| 226 | |
| 227 | func (c *closableHealthAndIngesterClient) worker(ctx context.Context) error { |
| 228 | stream, err := c.PushStream(ctx) |
| 229 | if err != nil { |
| 230 | return err |
| 231 | } |
| 232 | go func() { |
| 233 | for { |
| 234 | select { |
| 235 | case <-ctx.Done(): |
| 236 | return |
| 237 | case job, ok := <-c.streamPushChan: |
| 238 | if !ok { |
| 239 | return |
| 240 | } |
| 241 | if done := c.processJob(stream, job); done { |
| 242 | return |
| 243 | } |
| 244 | } |
| 245 | } |
| 246 | }() |
| 247 | return nil |
| 248 | } |
| 249 | |
| 250 | // processJob handles a single job and returns true if the stream should be closed. |
| 251 | func (c *closableHealthAndIngesterClient) processJob(stream Ingester_PushStreamClient, job *streamWriteJob) (done bool) { |