()
| 237 | } |
| 238 | |
| 239 | func (om *offsetManager) mainLoop() { |
| 240 | defer om.ticker.Stop() |
| 241 | defer close(om.closed) |
| 242 | |
| 243 | for { |
| 244 | select { |
| 245 | case <-om.ticker.C: |
| 246 | om.Commit() |
| 247 | case <-om.closing: |
| 248 | return |
| 249 | } |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | func (om *offsetManager) Commit() { |
| 254 | om.flushToBroker() |