add service registry events allowing to listen on service references

services-api/src/main/java/io/scalecube/services/registry/api/RegistryEvent.java

@@ -0,0 +1,44 @@
+package io.scalecube.services.registry.api;
+
+import io.scalecube.services.ServiceReference;
+import io.scalecube.transport.Address;
+
+public class RegistryEvent {
+
+  public enum Type {
+    ADDED, REMOVED;
+  }
+
+  private ServiceReference serviceReference;
+  private Type type;
+
+  public RegistryEvent(Type type, ServiceReference serviceReference) {
+    this.serviceReference = serviceReference;
+    this.type = type;
+  }
+
+  public RegistryEvent(RegistryEvent e) {
+    this.serviceReference = e.serviceReference;
+    this.type = e.type;
+  }
+
+  public ServiceReference serviceReference() {
+    return this.serviceReference;
+  }
+
+  public boolean isAdded() {
+    return Type.ADDED.equals(type);
+  }
+
+  public boolean isRemoved() {
+    return Type.REMOVED.equals(type);
+  }
+
+  public Type type() {
+    return this.type;
+  }
+
+  public Address address() {
+    return Address.create(this.serviceReference.host(), this.serviceReference.port());
+  }
+}

services-api/src/main/java/io/scalecube/services/registry/api/ServiceRegistry.java

@@ -4,6 +4,7 @@
 import io.scalecube.services.ServiceReference;
 import io.scalecube.services.api.ServiceMessage;
 import java.util.List;
+import reactor.core.publisher.Flux;
 
 /**
  * Service registry interface provides API to register/unregister services in the system and make
@@ -20,4 +21,6 @@ public interface ServiceRegistry {
   boolean registerService(ServiceEndpoint serviceEndpoint);
 
   ServiceEndpoint unregisterService(String endpointId);
+
+  Flux<RegistryEvent> listen();
 }

services/src/main/java/io/scalecube/services/registry/ServiceRegistryImpl.java

@@ -3,6 +3,8 @@
 import io.scalecube.services.ServiceEndpoint;
 import io.scalecube.services.ServiceReference;
 import io.scalecube.services.api.ServiceMessage;
+import io.scalecube.services.registry.api.RegistryEvent;
+import io.scalecube.services.registry.api.RegistryEvent.Type;
 import io.scalecube.services.registry.api.ServiceRegistry;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -13,6 +15,10 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.jctools.maps.NonBlockingHashMap;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxProcessor;
+import reactor.core.publisher.FluxSink;
 
 public class ServiceRegistryImpl implements ServiceRegistry {
 
@@ -21,6 +27,11 @@ public class ServiceRegistryImpl implements ServiceRegistry {
   private final Map<String, List<ServiceReference>> referencesByQualifier =
       new NonBlockingHashMap<>();
 
+  private final FluxProcessor<RegistryEvent, RegistryEvent> events =
+      DirectProcessor.<RegistryEvent>create();
+
+  private final FluxSink<RegistryEvent> sink = events.serialize().sink();
+
   @Override
   public List<ServiceEndpoint> listServiceEndpoints() {
     // todo how to collect tags correctly?
@@ -52,18 +63,14 @@ public boolean registerService(ServiceEndpoint serviceEndpoint) {
       serviceEndpoint
           .serviceRegistrations()
           .stream()
-          .flatMap(
-              serviceRegistration ->
-                  serviceRegistration
-                      .methods()
-                      .stream()
-                      .map(sm -> new ServiceReference(sm, serviceRegistration, serviceEndpoint)))
-          .forEach(
-              serviceReference ->
-                  referencesByQualifier
-                      .computeIfAbsent(
-                          serviceReference.qualifier(), key -> new CopyOnWriteArrayList<>())
-                      .add(serviceReference));
+          .flatMap(serviceRegistration -> serviceRegistration.methods().stream()
+              .map(sm -> new ServiceReference(sm, serviceRegistration, serviceEndpoint)))
+          .forEach(serviceReference -> {
+            referencesByQualifier
+                .computeIfAbsent(serviceReference.qualifier(), key -> new CopyOnWriteArrayList<>())
+                .add(serviceReference);
+            sink.next(new RegistryEvent(Type.ADDED, serviceReference));
+          });
     }
     return success;
   }
@@ -72,14 +79,29 @@ public boolean registerService(ServiceEndpoint serviceEndpoint) {
   public ServiceEndpoint unregisterService(String endpointId) {
     ServiceEndpoint serviceEndpoint = serviceEndpoints.remove(endpointId);
     if (serviceEndpoint != null) {
-      referencesByQualifier
-          .values()
-          .forEach(list -> list.removeIf(sr -> sr.endpointId().equals(endpointId)));
+      referencesByQualifier.values()
+          .forEach(list -> {
+            list.stream().filter(sr -> sr.endpointId().equals(endpointId))
+                .forEach(sr -> {
+                  list.remove(sr);
+                  sink.next(new RegistryEvent(Type.REMOVED, sr));
+                });
+          });
     }
+
     return serviceEndpoint;
   }
 
   Stream<ServiceReference> serviceReferenceStream() {
     return referencesByQualifier.values().stream().flatMap(Collection::stream);
   }
+
+  /**
+   * listen on service registry events.
+   */
+  public Flux<RegistryEvent> listen() {
+    return Flux.fromIterable(referencesByQualifier.values()).flatMap(Flux::fromIterable)
+        .map(sr -> new RegistryEvent(Type.ADDED, sr))
+        .concatWith(events);
+  }
 }