(brokerWorker *brokerConsumer)
| 261 | } |
| 262 | |
| 263 | func (c *consumer) unrefBrokerConsumer(brokerWorker *brokerConsumer) { |
| 264 | c.lock.Lock() |
| 265 | defer c.lock.Unlock() |
| 266 | |
| 267 | brokerWorker.refs-- |
| 268 | |
| 269 | if brokerWorker.refs == 0 { |
| 270 | brokerWorker.stopConsuming() |
| 271 | if c.brokerConsumers[brokerWorker.broker] == brokerWorker { |
| 272 | delete(c.brokerConsumers, brokerWorker.broker) |
| 273 | } |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | func (c *consumer) abandonBrokerConsumer(brokerWorker *brokerConsumer) { |
| 278 | c.lock.Lock() |