| package main |
| |
| import ( |
| "context" |
| "regexp" |
| "testing" |
| "time" |
| ) |
| |
| func makeDut() (*dispatcher, context.CancelFunc, context.Context) { |
| dut := newDispatcher() |
| |
| ctx := context.Background() |
| ctxC, cancelCtx := context.WithCancel(ctx) |
| go dut.run(ctxC) |
| |
| return dut, cancelCtx, ctx |
| } |
| |
| func expectReceived(t *testing.T, s *sms, data chan *sms) { |
| ticker := time.NewTicker(100 * time.Millisecond) |
| defer ticker.Stop() |
| select { |
| case d := <-data: |
| if d.from != s.from { |
| t.Errorf("Received SMS from %q, wanted %q", d.from, s.from) |
| } |
| if d.body != s.body { |
| t.Errorf("Received SMS body %q, wanted %q", d.body, s.body) |
| } |
| if d.timestamp != s.timestamp { |
| t.Errorf("Received SMS timestamp %v, wanted %v", d.timestamp, s.timestamp) |
| } |
| case <-ticker.C: |
| t.Fatalf("Timed out waiting for message") |
| } |
| } |
| |
| func expectEmpty(t *testing.T, data chan *sms) { |
| ticker := time.NewTicker(1 * time.Millisecond) |
| defer ticker.Stop() |
| select { |
| case <-data: |
| t.Fatalf("Received unwanted message") |
| case <-ticker.C: |
| } |
| } |
| |
| func TestDispatcher(t *testing.T) { |
| dut, cancelDut, _ := makeDut() |
| defer cancelDut() |
| |
| data := make(chan *sms) |
| cancel := make(chan struct{}) |
| |
| dut.subscribe(&subscriber{ |
| re: regexp.MustCompile(".*"), |
| data: data, |
| cancel: cancel, |
| }) |
| |
| in := &sms{ |
| from: "+4821372137", |
| body: "foo", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| |
| // Make sure we ge the message. |
| expectReceived(t, in, data) |
| |
| // Make sure we don't receive the message again. |
| expectEmpty(t, data) |
| |
| // Publish a new message, but this time close our subscriber. |
| close(cancel) |
| // Hack: yield. |
| time.Sleep(1 * time.Millisecond) |
| |
| dut.publish(in) |
| expectEmpty(t, data) |
| } |
| |
| type testSubscriber struct { |
| re *regexp.Regexp |
| data chan *sms |
| cancel chan struct{} |
| } |
| |
| func TestDispatcherFilters(t *testing.T) { |
| dut, cancelDut, _ := makeDut() |
| defer cancelDut() |
| |
| subscribers := []*testSubscriber{ |
| {re: regexp.MustCompile(".*")}, |
| {re: regexp.MustCompile("foo")}, |
| {re: regexp.MustCompile("bar")}, |
| } |
| |
| for _, s := range subscribers { |
| s.data = make(chan *sms) |
| s.cancel = make(chan struct{}) |
| dut.subscribe(&subscriber{ |
| re: s.re, |
| data: s.data, |
| cancel: s.cancel, |
| }) |
| defer func(c chan struct{}) { |
| close(c) |
| }(s.cancel) |
| } |
| |
| in := &sms{ |
| from: "+4821372137", |
| body: "foo", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| expectReceived(t, in, subscribers[0].data) |
| expectReceived(t, in, subscribers[1].data) |
| expectEmpty(t, subscribers[2].data) |
| |
| in = &sms{ |
| from: "+4821372137", |
| body: "bar", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| expectReceived(t, in, subscribers[0].data) |
| expectEmpty(t, subscribers[1].data) |
| expectReceived(t, in, subscribers[2].data) |
| |
| in = &sms{ |
| from: "+4821372137", |
| body: "foobar", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| expectReceived(t, in, subscribers[0].data) |
| expectReceived(t, in, subscribers[1].data) |
| expectReceived(t, in, subscribers[2].data) |
| } |
| |
| func TestDispatcherMany(t *testing.T) { |
| dut, cancelDut, _ := makeDut() |
| defer cancelDut() |
| |
| subscribers := make([]*testSubscriber, 10000) |
| |
| for i, _ := range subscribers { |
| s := &testSubscriber{ |
| re: regexp.MustCompile(".*"), |
| data: make(chan *sms), |
| cancel: make(chan struct{}), |
| } |
| subscribers[i] = s |
| dut.subscribe(&subscriber{ |
| re: s.re, |
| data: s.data, |
| cancel: s.cancel, |
| }) |
| defer func(c chan struct{}) { |
| close(c) |
| }(s.cancel) |
| } |
| |
| in := &sms{ |
| from: "+4821372137", |
| body: "foo", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| |
| for _, s := range subscribers { |
| expectReceived(t, in, s.data) |
| } |
| } |
| |
| func TestDispatcherHammer(t *testing.T) { |
| dut, cancelDut, _ := makeDut() |
| defer cancelDut() |
| |
| for i := 0; i < 1000000; i += 1 { |
| s := &testSubscriber{ |
| re: regexp.MustCompile(".*"), |
| data: make(chan *sms), |
| cancel: make(chan struct{}), |
| } |
| |
| dut.subscribe(&subscriber{ |
| re: s.re, |
| data: s.data, |
| cancel: s.cancel, |
| }) |
| |
| in := &sms{ |
| from: "+4821372137", |
| body: "foo", |
| timestamp: time.Now(), |
| } |
| dut.publish(in) |
| expectReceived(t, in, s.data) |
| |
| close(s.cancel) |
| } |
| } |