Pass done channel to counter
This commit is contained in:
parent
d74792a591
commit
959bc56bc3
|
@ -47,9 +47,9 @@ func (h *Hub) Pub(queue string, msg []byte) bool {
|
|||
}
|
||||
|
||||
func (h *Hub) Sub(s *Subscription) {
|
||||
for _, q := range s.Queues {
|
||||
if msg, ok := h.storage.Get(q); ok {
|
||||
s.Send(Message{q, msg})
|
||||
for _, queue := range s.Queues {
|
||||
if msg, ok := h.storage.Get(queue, s.Done()); ok {
|
||||
s.Send(Message{queue, msg})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
|
|
@ -35,7 +35,7 @@ func New(path string) (s *Storage, err error) {
|
|||
return
|
||||
}
|
||||
|
||||
func (s *Storage) Get(queue string) (message []byte, ok bool) {
|
||||
func (s *Storage) Get(queue string, done <-chan struct{}) (message []byte, ok bool) {
|
||||
if _, exist := s.counters[queue]; !exist {
|
||||
return
|
||||
}
|
||||
|
@ -46,7 +46,7 @@ func (s *Storage) Get(queue string) (message []byte, ok bool) {
|
|||
var index uint
|
||||
select {
|
||||
case index = <-s.counters[queue].stream:
|
||||
default:
|
||||
case <-done:
|
||||
return
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue