Merge pull request #3 from Comcast/feature/rehash

Use the new RegisteredEndpoints type to filter out hashed endpoints t…

src/glide.yaml

@@ -2,7 +2,7 @@ package: .
 homepage: https://github.com/Comcast/talaria
 import:
 - package: github.com/Comcast/webpa-common
-  version: e6275e251ccc388a3f0dacf2c26e07f2bb6f387c
+  version: 447435f44d105295b7af54bc28c22f9b3f48a137
   subpackages:
   - concurrent
   - device

src/talaria/talaria.go

@@ -75,7 +75,7 @@ func talaria(arguments []string) int {
 	// Now, initialize the service discovery infrastructure
 	//
 
-	serviceOptions, registrar, err := service.Initialize(logger, nil, v.Sub(service.DiscoveryKey))
+	serviceOptions, registrar, registeredEndpoints, err := service.Initialize(logger, nil, v.Sub(service.DiscoveryKey))
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "Unable to initialize service discovery: %s\n", err)
 		return 2
@@ -90,6 +90,21 @@ func talaria(arguments []string) int {
 			Registrar: registrar,
 			Listener: func(endpoints []string) {
 				accessor.Update(endpoints)
+				manager.DisconnectIf(func(candidate device.ID) bool {
+					hashedEndpoint, err := accessor.Get(candidate.Bytes())
+					if err != nil {
+						logger.Error("Error while attempting to rehash device id %s: %s", candidate, err)
+						return true
+					}
+
+					// disconnect if hashedEndpoint was NOT found in the endpoints that this talaria instance registered under
+					disconnect := !registeredEndpoints.Has(hashedEndpoint)
+					if disconnect {
+						logger.Info("Disconnecting %s due to service discovery rehash", candidate)
+					}
+
+					return disconnect
+				})
 			},
 		}