Merge pull request #772 from scalecube/expose-service-registry

artem-v · web-flow · commit dd30d655b03b · 2020-06-18T21:01:34.000+03:00
Fixed discovery contexts
diff --git a/services-api/src/main/java/io/scalecube/services/discovery/api/ServiceDiscovery.java b/services-api/src/main/java/io/scalecube/services/discovery/api/ServiceDiscovery.java
@@ -6,9 +6,9 @@
 public interface ServiceDiscovery {
 
   /**
-   * Function to subscribe and listen on {@code ServiceDiscoveryEvent} events.
+   * Function to subscribe and listen on stream of {@code ServiceDiscoveryEvent}\s.
    *
-   * @return stream of {@code ServiceDiscoveryEvent} events
+   * @return stream of {@code ServiceDiscoveryEvent}\s
    */
   Flux<ServiceDiscoveryEvent> listen();
 
diff --git a/services-api/src/main/java/io/scalecube/services/discovery/api/ServiceDiscoveryContext.java b/services-api/src/main/java/io/scalecube/services/discovery/api/ServiceDiscoveryContext.java
@@ -1,36 +1,84 @@
 package io.scalecube.services.discovery.api;
 
 import io.scalecube.net.Address;
+import io.scalecube.services.discovery.api.ServiceDiscoveryEvent.Type;
+import io.scalecube.services.registry.api.ServiceRegistry;
 import java.util.Objects;
 import java.util.StringJoiner;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
 
 public final class ServiceDiscoveryContext {
 
   private final String id;
   private final Address address;
   private final ServiceDiscovery discovery;
+  private final ServiceRegistry serviceRegistry;
+  private final Scheduler scheduler;
 
   private ServiceDiscoveryContext(Builder builder) {
-    this.id = Objects.requireNonNull(builder.id, "discoveryContext.id");
-    this.address = Objects.requireNonNull(builder.address, "discoveryContext.address");
-    this.discovery = Objects.requireNonNull(builder.discovery, "discoveryContext.discovery");
+    this.id = Objects.requireNonNull(builder.id, "id");
+    this.address = Objects.requireNonNull(builder.address, "address");
+    this.discovery = Objects.requireNonNull(builder.discovery, "discovery");
+    this.serviceRegistry = Objects.requireNonNull(builder.serviceRegistry, "serviceRegistry");
+    this.scheduler = Objects.requireNonNull(builder.scheduler, "scheduler");
   }
 
   public static Builder builder() {
     return new Builder();
   }
 
+  /**
+   * Returns copy builder.
+   *
+   * @param other instance to copy from
+   * @return new builder instance
+   */
+  public static Builder from(ServiceDiscoveryContext other) {
+    return new Builder()
+        .id(other.id)
+        .address(other.address)
+        .discovery(other.discovery)
+        .serviceRegistry(other.serviceRegistry)
+        .scheduler(other.scheduler);
+  }
+
+  /**
+   * Returns service discovery id.
+   *
+   * @return service discovery id
+   */
   public String id() {
     return id;
   }
 
+  /**
+   * Returns service discovery address. If service discovery instance is not yet started, then this
+   * method will return {@link Address#NULL_ADDRESS}.
+   *
+   * @see ServiceDiscovery#start()
+   * @see ServiceDiscovery#listen()
+   * @return service discovery address
+   */
   public Address address() {
     return address;
   }
 
+  /**
+   * Returns stream of service discovery events. Can be called before or after {@link
+   * ServiceDiscovery#start()}. If it's called before then new events will be streamed, if it's
+   * called after then {@link ServiceRegistry#listServiceEndpoints()} will be turned to service
+   * discovery events of type {@link Type#ENDPOINT_ADDED}, and concateneted with a stream of live
+   * events.
+   *
+   * @return stream of service discovery events
+   */
   public Flux<ServiceDiscoveryEvent> listen() {
-    return discovery.listen();
+    return Flux.fromStream(serviceRegistry.listServiceEndpoints().stream())
+        .map(ServiceDiscoveryEvent::newEndpointAdded)
+        .concatWith(discovery.listen())
+        .subscribeOn(scheduler)
+        .publishOn(scheduler);
   }
 
   @Override
@@ -39,6 +87,8 @@ public String toString() {
         .add("id='" + id + "'")
         .add("address=" + address)
         .add("discovery=" + discovery)
+        .add("serviceRegistry=" + serviceRegistry)
+        .add("scheduler=" + scheduler)
         .toString();
   }
 
@@ -47,6 +97,8 @@ public static class Builder {
     private String id;
     private Address address;
     private ServiceDiscovery discovery;
+    private ServiceRegistry serviceRegistry;
+    private Scheduler scheduler;
 
     private Builder() {}
 
@@ -65,6 +117,16 @@ public Builder discovery(ServiceDiscovery discovery) {
       return this;
     }
 
+    public Builder serviceRegistry(ServiceRegistry serviceRegistry) {
+      this.serviceRegistry = serviceRegistry;
+      return this;
+    }
+
+    public Builder scheduler(Scheduler scheduler) {
+      this.scheduler = scheduler;
+      return this;
+    }
+
     public ServiceDiscoveryContext build() {
       return new ServiceDiscoveryContext(this);
     }
diff --git a/services-discovery/src/main/java/io/scalecube/services/discovery/ScalecubeServiceDiscovery.java b/services-discovery/src/main/java/io/scalecube/services/discovery/ScalecubeServiceDiscovery.java
@@ -59,7 +59,6 @@ public ScalecubeServiceDiscovery(ServiceEndpoint serviceEndpoint) {
     this.clusterConfig =
         ClusterConfig.defaultLanConfig()
             .metadata(serviceEndpoint)
-            .memberIdGenerator(serviceEndpoint::id)
             .transport(opts -> opts.transportFactory(new WebsocketTransportFactory()));
   }
 
diff --git a/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/discovery/CompositeDiscoveryExample.java b/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/discovery/CompositeDiscoveryExample.java
@@ -2,9 +2,11 @@
 
 import io.scalecube.net.Address;
 import io.scalecube.services.Microservices;
+import io.scalecube.services.annotations.AfterConstruct;
 import io.scalecube.services.annotations.Service;
 import io.scalecube.services.annotations.ServiceMethod;
 import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
+import io.scalecube.services.discovery.api.ServiceDiscoveryContext;
 import io.scalecube.services.examples.helloworld.service.api.Greeting;
 import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
 import reactor.core.publisher.Mono;
@@ -16,16 +18,24 @@ public class CompositeDiscoveryExample {
    *
    * @param args arguments
    */
-  public static void main(String[] args) throws InterruptedException {
+  public static void main(String[] args) {
     Microservices seed1 =
         Microservices.builder()
-            .discovery("seed1", ScalecubeServiceDiscovery::new)
+            .discovery(
+                "seed1",
+                serviceEndpoint ->
+                    new ScalecubeServiceDiscovery(serviceEndpoint)
+                        .options(opts -> opts.memberAlias("seed1")))
             .transport(RSocketServiceTransport::new)
             .startAwait();
 
     Microservices seed2 =
         Microservices.builder()
-            .discovery("seed2", ScalecubeServiceDiscovery::new)
+            .discovery(
+                "seed2",
+                serviceEndpoint ->
+                    new ScalecubeServiceDiscovery(serviceEndpoint)
+                        .options(opts -> opts.memberAlias("seed2")))
             .transport(RSocketServiceTransport::new)
             .startAwait();
 
@@ -38,6 +48,7 @@ public static void main(String[] args) throws InterruptedException {
                 "ms1",
                 endpoint ->
                     new ScalecubeServiceDiscovery(endpoint)
+                        .options(opts -> opts.memberAlias("ms1"))
                         .membership(cfg -> cfg.seedMembers(seed1Address)))
             .transport(RSocketServiceTransport::new)
             .services(new GreetingServiceImpl1())
@@ -49,6 +60,7 @@ public static void main(String[] args) throws InterruptedException {
                 "ms2",
                 endpoint ->
                     new ScalecubeServiceDiscovery(endpoint)
+                        .options(opts -> opts.memberAlias("ms2"))
                         .membership(cfg -> cfg.seedMembers(seed2Address)))
             .transport(RSocketServiceTransport::new)
             .services(new GreetingServiceImpl2())
@@ -60,13 +72,13 @@ public static void main(String[] args) throws InterruptedException {
                 "domain1",
                 endpoint ->
                     new ScalecubeServiceDiscovery(endpoint)
-                        .options(cfg -> cfg.memberIdGenerator(endpoint::id))
+                        .options(opts -> opts.memberAlias("domain1"))
                         .membership(cfg -> cfg.seedMembers(seed1Address)))
             .discovery(
                 "domain2",
                 endpoint ->
                     new ScalecubeServiceDiscovery(endpoint)
-                        .options(cfg -> cfg.memberIdGenerator(endpoint::id))
+                        .options(opts -> opts.memberAlias("domain2"))
                         .membership(cfg -> cfg.seedMembers(seed2Address)))
             .transport(RSocketServiceTransport::new)
             .startAwait();
@@ -78,8 +90,6 @@ public static void main(String[] args) throws InterruptedException {
     Greeting greeting2 =
         compositeMs.call().api(GreetingsService2.class).sayHello("hello two").block();
     System.err.println("This is response from GreetingsService2: " + greeting2.message());
-
-    Thread.currentThread().join();
   }
 
   @Service
@@ -98,6 +108,16 @@ public interface GreetingsService2 {
 
   public static class GreetingServiceImpl1 implements GreetingsService1 {
 
+    @AfterConstruct
+    void init(Microservices ms) {
+      ServiceDiscoveryContext discoveryContext = ms.discovery("ms1");
+      System.err.println("discovery(\"ms1\"): " + discoveryContext);
+      discoveryContext
+          .listen()
+          .subscribe(
+              discoveryEvent -> System.err.println("discovery(\"ms1\") event: " + discoveryEvent));
+    }
+
     @Override
     public Mono<Greeting> sayHello(String name) {
       return Mono.just(
@@ -110,6 +130,16 @@ public Mono<Greeting> sayHello(String name) {
 
   public static class GreetingServiceImpl2 implements GreetingsService2 {
 
+    @AfterConstruct
+    void init(Microservices ms) {
+      ServiceDiscoveryContext discoveryContext = ms.discovery("ms2");
+      System.err.println("discovery(\"ms2\"): " + discoveryContext);
+      discoveryContext
+          .listen()
+          .subscribe(
+              discoveryEvent -> System.err.println("discovery(\"ms2\") event: " + discoveryEvent));
+    }
+
     @Override
     public Mono<Greeting> sayHello(String name) {
       return Mono.just(
diff --git a/services/src/main/java/io/scalecube/services/Microservices.java b/services/src/main/java/io/scalecube/services/Microservices.java

Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,6 @@ public ScalecubeServiceDiscovery(ServiceEndpoint serviceEndpoint) {`
`59`	`59`	`this.clusterConfig =`
`60`	`60`	`ClusterConfig.defaultLanConfig()`
`61`	`61`	`.metadata(serviceEndpoint)`
`62`		`- .memberIdGenerator(serviceEndpoint::id)`
`63`	`62`	`.transport(opts -> opts.transportFactory(new WebsocketTransportFactory()));`
`64`	`63`	`}`
`65`	`64`