| 126 | } |
| 127 | |
| 128 | func (q *MsgQueue) Enqueue(msg []byte) { |
| 129 | q.cond.L.Lock() |
| 130 | defer q.cond.L.Unlock() |
| 131 | |
| 132 | if q.size == BufferSize { |
| 133 | // queue is full, so we're going to drop the msg we got called with. |
| 134 | // We also need to record that messages are being dropped, which we |
| 135 | // do at the last message in the queue. This potentially makes us |
| 136 | // lose 2 messages instead of one, but it's more important at this |
| 137 | // point to warn the subscriber that they're losing messages so they |
| 138 | // can do something about it. |
| 139 | back := (q.front + BufferSize - 1) % BufferSize |
| 140 | q.q[back].msg = nil |
| 141 | q.q[back].err = ErrDroppedMessages |
| 142 | return |
| 143 | } |
| 144 | // queue is not full, insert the message |
| 145 | next := (q.front + q.size) % BufferSize |
| 146 | q.q[next].msg = msg |
| 147 | q.q[next].err = nil |
| 148 | q.size++ |
| 149 | q.cond.Broadcast() |
| 150 | } |
| 151 | |
| 152 | func (q *MsgQueue) Close() { |
| 153 | q.cond.L.Lock() |