package serf
|
|
import (
|
"fmt"
|
"reflect"
|
"testing"
|
"time"
|
)
|
|
// Mock EventCounter type
|
const EventCounter EventType = 9000
|
|
type counterEvent struct {
|
delta int
|
}
|
|
func (c counterEvent) EventType() EventType {
|
return EventCounter
|
}
|
func (c counterEvent) String() string {
|
return fmt.Sprintf("CounterEvent %d", c.delta)
|
}
|
|
// Mock coalescer
|
type mockCoalesce struct {
|
value int
|
}
|
|
func (c *mockCoalesce) Handle(e Event) bool {
|
return e.EventType() == EventCounter
|
}
|
|
func (c *mockCoalesce) Coalesce(e Event) {
|
c.value += e.(counterEvent).delta
|
}
|
|
func (c *mockCoalesce) Flush(outChan chan<- Event) {
|
outChan <- counterEvent{c.value}
|
c.value = 0
|
}
|
|
func testCoalescer(cPeriod, qPeriod time.Duration) (chan<- Event, <-chan Event, chan<- struct{}) {
|
in := make(chan Event, 64)
|
out := make(chan Event)
|
shutdown := make(chan struct{})
|
c := &mockCoalesce{}
|
go coalesceLoop(in, out, shutdown, cPeriod, qPeriod, c)
|
return in, out, shutdown
|
}
|
|
func TestCoalescer_basic(t *testing.T) {
|
in, out, shutdown := testCoalescer(5*time.Millisecond, time.Second)
|
defer close(shutdown)
|
|
send := []Event{
|
counterEvent{1},
|
counterEvent{39},
|
counterEvent{2},
|
}
|
for _, e := range send {
|
in <- e
|
}
|
|
select {
|
case e := <-out:
|
if e.EventType() != EventCounter {
|
t.Fatalf("expected counter, got: %d", e.EventType())
|
}
|
|
if e.(counterEvent).delta != 42 {
|
t.Fatalf("bad: %#v", e)
|
}
|
|
case <-time.After(50 * time.Millisecond):
|
t.Fatalf("timeout")
|
}
|
}
|
|
func TestCoalescer_quiescent(t *testing.T) {
|
// This tests the quiescence by creating a long coalescence period
|
// with a short quiescent period and waiting only a multiple of the
|
// quiescent period for results.
|
in, out, shutdown := testCoalescer(10*time.Second, 10*time.Millisecond)
|
defer close(shutdown)
|
|
send := []Event{
|
counterEvent{1},
|
counterEvent{39},
|
counterEvent{2},
|
}
|
for _, e := range send {
|
in <- e
|
}
|
|
select {
|
case e := <-out:
|
if e.EventType() != EventCounter {
|
t.Fatalf("expected counter, got: %d", e.EventType())
|
}
|
|
if e.(counterEvent).delta != 42 {
|
t.Fatalf("bad: %#v", e)
|
}
|
case <-time.After(50 * time.Millisecond):
|
t.Fatalf("timeout")
|
}
|
}
|
|
func TestCoalescer_passThrough(t *testing.T) {
|
in, out, shutdown := testCoalescer(time.Second, time.Second)
|
defer close(shutdown)
|
|
send := []Event{
|
UserEvent{
|
Name: "test",
|
Payload: []byte("foo"),
|
},
|
}
|
|
for _, e := range send {
|
in <- e
|
}
|
|
select {
|
case e := <-out:
|
if e.EventType() != EventUser {
|
t.Fatalf("expected user event, got: %d", e.EventType())
|
}
|
|
if e.(UserEvent).Name != "test" {
|
t.Fatalf("name should be test. %v", e)
|
}
|
|
if !reflect.DeepEqual([]byte("foo"), e.(UserEvent).Payload) {
|
t.Fatalf("bad: %#v", e.(UserEvent).Payload)
|
}
|
case <-time.After(50 * time.Millisecond):
|
t.Fatalf("timeout")
|
}
|
}
|
