Support push model for service discovery

examples/apigateway/main.go

@@ -84,18 +84,19 @@ func main() {
 			tags        = []string{}
 			passingOnly = true
 			endpoints   = addsvc.Endpoints{}
+			instancer   = consulsd.NewInstancer(client, logger, "addsvc", tags, passingOnly)
 		)
 		{
 			factory := addsvcFactory(addsvc.MakeSumEndpoint, tracer, logger)
-			subscriber := consulsd.NewSubscriber(client, factory, logger, "addsvc", tags, passingOnly)
-			balancer := lb.NewRoundRobin(subscriber)
+			endpointer := sd.NewEndpointer(instancer, factory, logger)
+			balancer := lb.NewRoundRobin(endpointer)
 			retry := lb.Retry(*retryMax, *retryTimeout, balancer)
 			endpoints.SumEndpoint = retry
 		}
 		{
 			factory := addsvcFactory(addsvc.MakeConcatEndpoint, tracer, logger)
-			subscriber := consulsd.NewSubscriber(client, factory, logger, "addsvc", tags, passingOnly)
-			balancer := lb.NewRoundRobin(subscriber)
+			endpointer := sd.NewEndpointer(instancer, factory, logger)
+			balancer := lb.NewRoundRobin(endpointer)
 			retry := lb.Retry(*retryMax, *retryTimeout, balancer)
 			endpoints.ConcatEndpoint = retry
 		}
@@ -120,18 +121,19 @@ func main() {
 			passingOnly = true
 			uppercase   endpoint.Endpoint
 			count       endpoint.Endpoint
+			instancer   = consulsd.NewInstancer(client, logger, "stringsvc", tags, passingOnly)
 		)
 		{
 			factory := stringsvcFactory(ctx, "GET", "/uppercase")
-			subscriber := consulsd.NewSubscriber(client, factory, logger, "stringsvc", tags, passingOnly)
-			balancer := lb.NewRoundRobin(subscriber)
+			endpointer := sd.NewEndpointer(instancer, factory, logger)
+			balancer := lb.NewRoundRobin(endpointer)
 			retry := lb.Retry(*retryMax, *retryTimeout, balancer)
 			uppercase = retry
 		}
 		{
 			factory := stringsvcFactory(ctx, "GET", "/count")
-			subscriber := consulsd.NewSubscriber(client, factory, logger, "stringsvc", tags, passingOnly)
-			balancer := lb.NewRoundRobin(subscriber)
+			endpointer := sd.NewEndpointer(instancer, factory, logger)
+			balancer := lb.NewRoundRobin(endpointer)
 			retry := lb.Retry(*retryMax, *retryTimeout, balancer)
 			count = retry
 		}

examples/profilesvc/client/client.go

@@ -40,68 +40,69 @@ func New(consulAddr string, logger log.Logger) (profilesvc.Service, error) {
 
 	var (
 		sdclient  = consul.NewClient(apiclient)
+		instancer = consul.NewInstancer(sdclient, logger, consulService, consulTags, passingOnly)
 		endpoints profilesvc.Endpoints
 	)
 	{
 		factory := factoryFor(profilesvc.MakePostProfileEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.PostProfileEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakeGetProfileEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.GetProfileEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakePutProfileEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.PutProfileEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakePatchProfileEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.PatchProfileEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakeDeleteProfileEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.DeleteProfileEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakeGetAddressesEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.GetAddressesEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakeGetAddressEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.GetAddressEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakePostAddressEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.PostAddressEndpoint = retry
 	}
 	{
 		factory := factoryFor(profilesvc.MakeDeleteAddressEndpoint)
-		subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
-		balancer := lb.NewRoundRobin(subscriber)
+		endpointer := sd.NewEndpointer(instancer, factory, logger)
+		balancer := lb.NewRoundRobin(endpointer)
 		retry := lb.Retry(retryMax, retryTimeout, balancer)
 		endpoints.DeleteAddressEndpoint = retry
 	}

examples/stringsvc3/proxying.go

@@ -40,20 +40,20 @@ func proxyingMiddleware(ctx context.Context, instances string, logger log.Logger
 	// discovery system.
 	var (
 		instanceList = split(instances)
-		subscriber   sd.FixedSubscriber
+		endpointer   sd.FixedEndpointer
 	)
 	logger.Log("proxy_to", fmt.Sprint(instanceList))
 	for _, instance := range instanceList {
 		var e endpoint.Endpoint
 		e = makeUppercaseProxy(ctx, instance)
 		e = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{}))(e)
 		e = ratelimit.NewTokenBucketLimiter(jujuratelimit.NewBucketWithRate(float64(qps), int64(qps)))(e)
-		subscriber = append(subscriber, e)
+		endpointer = append(endpointer, e)
 	}
 
 	// Now, build a single, retrying, load-balancing endpoint out of all of
 	// those individual endpoints.
-	balancer := lb.NewRoundRobin(subscriber)
+	balancer := lb.NewRoundRobin(endpointer)
 	retry := lb.Retry(maxAttempts, maxTime, balancer)
 
 	// And finally, return the ServiceMiddleware, implemented by proxymw.

sd/cache/benchmark_test.go → sd/benchmark_test.go

@@ -1,4 +1,4 @@
-package cache
+package sd
 
 import (
 	"io"
@@ -14,12 +14,12 @@ func BenchmarkEndpoints(b *testing.B) {
 		cb      = make(closer)
 		cmap    = map[string]io.Closer{"a": ca, "b": cb}
 		factory = func(instance string) (endpoint.Endpoint, io.Closer, error) { return endpoint.Nop, cmap[instance], nil }
-		c       = New(factory, log.NewNopLogger())
+		c       = newEndpointCache(factory, log.NewNopLogger(), endpointerOptions{})
 	)
 
 	b.ReportAllocs()
 
-	c.Update([]string{"a", "b"})
+	c.Update(Event{Instances: []string{"a", "b"}})
 
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {

sd/consul/doc.go

@@ -1,2 +1,2 @@
-// Package consul provides subscriber and registrar implementations for Consul.
+// Package consul provides Instancer and Registrar implementations for Consul.
 package consul