localhots/burlesque
1
0
Fork 0
Code Releases Activity

Abort inside queue instead of counter

Browse Source
This commit is contained in:
Gregory Eremin 2014-07-17 01:27:30 +07:00
parent 707a61d134
commit 1c62c6839b
2 changed files with 9 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
counter.go
View File

@ -16,8 +16,8 @@ type (
// If WriteIndex is greater than ReadIndex then there are unread messages // If WriteIndex is greater than ReadIndex then there are unread messages
// If WriteIndex is less tham ReadIndex then MaxIndex was reached // If WriteIndex is less tham ReadIndex then MaxIndex was reached
Read chan uint
mutex sync.Mutex mutex sync.Mutex
stream chan uint
streaming *sync.Cond streaming *sync.Cond
} }
) )
@ -48,15 +48,6 @@ func (c *Counter) Write(proc func(i uint) bool) {
} }
} }
func (c *Counter) Read(abort chan bool) uint {
select {
case i := <-c.stream:
return i
case <-abort:
return 0
}
}
func (c *Counter) Distance() uint { func (c *Counter) Distance() uint {
d := c.WriteIndex - c.ReadIndex d := c.WriteIndex - c.ReadIndex
if d < 0 { if d < 0 {

10
queue.go
View File

@ -36,9 +36,15 @@ func (q *Queue) TryFetch(abort chan bool) (Message, bool) {
} }
func (q *Queue) Fetch(abort chan bool) (Message, bool) { func (q *Queue) Fetch(abort chan bool) (Message, bool) {
i := q.Counter.Read(abort) var i uint
key := NewKey(q.Name, i)
select {
case i = <-q.Counter.Read:
case <-abort:
return Message{}, false
}
key := NewKey(q.Name, i)
msg, err := storage.Get(key) msg, err := storage.Get(key)
if err != nil { if err != nil {
Error(err, "Failed to read record '%s'", key) Error(err, "Failed to read record '%s'", key)