-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
misc fixes to address race conditions, enforce ordering of callbacks,…
… add unit tests
- Loading branch information
1 parent
e0af7d4
commit fdf68c5
Showing
11 changed files
with
454 additions
and
144 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package framework | ||
|
|
||
| import ( | ||
| "sort" | ||
| "sync" | ||
| ) | ||
|
|
||
| type MessageSortingQueue struct { | ||
| C chan []byte | ||
| lastCounter int | ||
| deferred []deferredMessage | ||
| lock sync.Mutex | ||
| closeOnce sync.Once | ||
| } | ||
|
|
||
| type deferredMessage struct { | ||
| counter int | ||
| message []byte | ||
| } | ||
|
|
||
| func NewMessageSortingQueue(channelSize int) *MessageSortingQueue { | ||
| return &MessageSortingQueue{C: make(chan []byte, channelSize)} | ||
| } | ||
|
|
||
| func (q *MessageSortingQueue) Accept(counter int, message []byte) { | ||
| q.lock.Lock() | ||
| if counter > q.lastCounter+1 { | ||
| q.deferred = append(q.deferred, deferredMessage{counter: counter, message: message}) | ||
| sort.Slice(q.deferred, func(i, j int) bool { return q.deferred[i].counter < q.deferred[j].counter }) | ||
| q.lock.Unlock() | ||
| return | ||
| } | ||
| q.lastCounter = counter | ||
| q.C <- message | ||
| for len(q.deferred) > 0 { | ||
| next := q.deferred[0] | ||
| if next.counter != q.lastCounter+1 { | ||
| break | ||
| } | ||
| q.deferred = q.deferred[1:] | ||
| q.lastCounter++ | ||
| q.C <- next.message | ||
| } | ||
| q.lock.Unlock() | ||
| } | ||
|
|
||
| func (q *MessageSortingQueue) Deferred() [][]byte { | ||
| q.lock.Lock() | ||
| ret := make([][]byte, 0, len(q.deferred)) | ||
| for _, d := range q.deferred { | ||
| ret = append(ret, d.message) | ||
| } | ||
| q.lock.Unlock() | ||
| return ret | ||
| } | ||
|
|
||
| func (q *MessageSortingQueue) Close() { | ||
| q.closeOnce.Do(func() { | ||
| close(q.C) | ||
| q.C = nil | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| package framework | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func fakeItemData(counter int) []byte { | ||
| return []byte(fmt.Sprintf("item-%d", counter)) | ||
| } | ||
|
|
||
| func acceptTestItems(q *MessageSortingQueue, counters ...int) { | ||
| for _, c := range counters { | ||
| q.Accept(c, fakeItemData(c)) | ||
| } | ||
| } | ||
|
|
||
| func expectTestItems(t *testing.T, q *MessageSortingQueue, counters ...int) { | ||
| for _, c := range counters { | ||
| select { | ||
| case item := <-q.C: | ||
| assert.Equal(t, string(fakeItemData(c)), string(item)) | ||
| case <-time.After(time.Second): | ||
| var deferredList []string | ||
| for _, d := range q.Deferred() { | ||
| deferredList = append(deferredList, string(d)) | ||
| } | ||
| require.Fail(t, "timed out waiting for item from queue", | ||
| "was waiting for item %d; deferred items were [%v]", strings.Join(deferredList, ",")) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func expectDeferredItems(t *testing.T, q *MessageSortingQueue, counters ...int) { | ||
| var expected, actual []string | ||
| for _, c := range counters { | ||
| expected = append(expected, string(fakeItemData(c))) | ||
| } | ||
| for _, d := range q.Deferred() { | ||
| actual = append(actual, string(d)) | ||
| } | ||
| assert.Equal(t, expected, actual, "did not see expected items in deferred list") | ||
| } | ||
|
|
||
| func TestMessageSortingQueueWithMessagesInOrder(t *testing.T) { | ||
| q := NewMessageSortingQueue(10) | ||
| acceptTestItems(q, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10) | ||
| expectDeferredItems(t, q) // should be empty | ||
| expectTestItems(t, q, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10) | ||
| } | ||
|
|
||
| func TestMessageSortingQueueWithMessagesOutOfOrder(t *testing.T) { | ||
| q := NewMessageSortingQueue(10) | ||
|
|
||
| acceptTestItems(q, 3) | ||
| expectDeferredItems(t, q, 3) | ||
|
|
||
| acceptTestItems(q, 2) | ||
| expectDeferredItems(t, q, 2, 3) | ||
|
|
||
| acceptTestItems(q, 6) | ||
| expectDeferredItems(t, q, 2, 3, 6) | ||
|
|
||
| acceptTestItems(q, 1) | ||
| expectTestItems(t, q, 1, 2, 3) | ||
| expectDeferredItems(t, q, 6) | ||
|
|
||
| acceptTestItems(q, 5) | ||
| expectDeferredItems(t, q, 5, 6) | ||
|
|
||
| acceptTestItems(q, 4) | ||
| expectTestItems(t, q, 4, 5, 6) | ||
| expectDeferredItems(t, q) // empty | ||
| } |
Oops, something went wrong.