vendor/github.com/go-kit/kit/sd/endpointer_test.go

   1 package sd_test
   2
   3 import (
   4         "io"
   5         "testing"
   6         "time"
   7
   8         "github.com/go-kit/kit/endpoint"
   9         "github.com/go-kit/kit/log"
  10         "github.com/go-kit/kit/sd"
  11         "github.com/go-kit/kit/sd/internal/instance"
  12 )
  13
  14 func TestDefaultEndpointer(t *testing.T) {
  15         var (
  16                 ca = make(closer)
  17                 cb = make(closer)
  18                 c  = map[string]io.Closer{"a": ca, "b": cb}
  19                 f  = func(instance string) (endpoint.Endpoint, io.Closer, error) {
  20                         return endpoint.Nop, c[instance], nil
  21                 }
  22                 instancer = &mockInstancer{instance.NewCache()}
  23         )
  24         // set initial state
  25         instancer.Update(sd.Event{Instances: []string{"a", "b"}})
  26
  27         endpointer := sd.NewEndpointer(instancer, f, log.NewNopLogger(), sd.InvalidateOnError(time.Minute))
  28
  29         var (
  30                 endpoints []endpoint.Endpoint
  31                 err       error
  32         )
  33         if !within(time.Second, func() bool {
  34                 endpoints, err = endpointer.Endpoints()
  35                 return err == nil && len(endpoints) == 2
  36         }) {
  37                 t.Errorf("wanted 2 endpoints, got %d (%v)", len(endpoints), err)
  38         }
  39
  40         instancer.Update(sd.Event{Instances: []string{}})
  41
  42         select {
  43         case <-ca:
  44                 t.Logf("endpoint a closed, good")
  45         case <-time.After(time.Millisecond):
  46                 t.Errorf("didn't close the deleted instance in time")
  47         }
  48
  49         select {
  50         case <-cb:
  51                 t.Logf("endpoint b closed, good")
  52         case <-time.After(time.Millisecond):
  53                 t.Errorf("didn't close the deleted instance in time")
  54         }
  55
  56         if endpoints, err := endpointer.Endpoints(); err != nil {
  57                 t.Errorf("unepected error %v", err)
  58         } else if want, have := 0, len(endpoints); want != have {
  59                 t.Errorf("want %d, have %d", want, have)
  60         }
  61
  62         endpointer.Close()
  63
  64         instancer.Update(sd.Event{Instances: []string{"a"}})
  65         // TODO verify that on Close the endpointer fully disconnects from the instancer.
  66         // Unfortunately, because we use instance.Cache, this test cannot be in the sd package,
  67         // and therefore does not have access to the endpointer's private members.
  68 }
  69
  70 type mockInstancer struct{ *instance.Cache }
  71
  72 type closer chan struct{}
  73
  74 func (c closer) Close() error { close(c); return nil }
  75
  76 func within(d time.Duration, f func() bool) bool {
  77         deadline := time.Now().Add(d)
  78         for time.Now().Before(deadline) {
  79                 if f() {
  80                         return true
  81                 }
  82                 time.Sleep(d / 10)
  83         }
  84         return false
  85 }