Merge branch 'master' into scalecubeHosts

services-api/src/main/java/io/scalecube/services/ServiceCall.java

@@ -7,7 +7,6 @@
 import io.scalecube.services.exceptions.ServiceUnavailableException;
 import io.scalecube.services.methods.MethodInfo;
 import io.scalecube.services.methods.ServiceMethodInvoker;
-import io.scalecube.services.methods.ServiceMethodRegistry;
 import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.routing.Router;
 import io.scalecube.services.routing.Routers;
@@ -34,7 +33,6 @@ public class ServiceCall {
  private static final Logger LOGGER = LoggerFactory.getLogger(ServiceCall.class);
 
  private ClientTransport transport;
- private ServiceMethodRegistry methodRegistry;
  private ServiceRegistry serviceRegistry;
  private Router router;
  private ServiceClientErrorMapper errorMapper = DefaultErrorMapper.INSTANCE;
@@ -45,7 +43,6 @@ public ServiceCall() {}
 
  private ServiceCall(ServiceCall other) {
  this.transport = other.transport;
- this.methodRegistry = other.methodRegistry;
  this.serviceRegistry = other.serviceRegistry;
  this.router = other.router;
  this.errorMapper = other.errorMapper;
@@ -77,18 +74,6 @@ public ServiceCall serviceRegistry(ServiceRegistry serviceRegistry) {
  return target;
  }
 
- /**
- * Setter for {@code methodRegistry}.
- *
- * @param methodRegistry method registry.
- * @return new {@link ServiceCall} instance.
- */
- public ServiceCall methodRegistry(ServiceMethodRegistry methodRegistry) {
- ServiceCall target = new ServiceCall(this);
- target.methodRegistry = methodRegistry;
- return target;
- }
-
  /**
  * Setter for {@code routerType}.
  *
@@ -180,8 +165,8 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType
  return Mono.defer(
  () -> {
  ServiceMethodInvoker methodInvoker;
- if (methodRegistry != null
- && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
+ if (serviceRegistry != null
+ && (methodInvoker = serviceRegistry.getInvoker(request.qualifier())) != null) {
  // local service
  return methodInvoker.invokeOne(request).map(this::throwIfError);
  } else {
@@ -219,8 +204,8 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseTyp
  return Flux.defer(
  () -> {
  ServiceMethodInvoker methodInvoker;
- if (methodRegistry != null
- && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
+ if (serviceRegistry != null
+ && (methodInvoker = serviceRegistry.getInvoker(request.qualifier())) != null) {
  // local service
  return methodInvoker.invokeMany(request).map(this::throwIfError);
  } else {
@@ -262,8 +247,8 @@ public Flux<ServiceMessage> requestBidirectional(
  if (first.hasValue()) {
  ServiceMessage request = first.get();
  ServiceMethodInvoker methodInvoker;
- if (methodRegistry != null
- && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
+ if (serviceRegistry != null
+ && (methodInvoker = serviceRegistry.getInvoker(request.qualifier())) != null) {
  // local service
  return methodInvoker.invokeBidirectional(messages).map(this::throwIfError);
  } else {

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

@@ -1,8 +1,10 @@
 package io.scalecube.services.registry.api;
 
 import io.scalecube.services.ServiceEndpoint;
+import io.scalecube.services.ServiceInfo;
 import io.scalecube.services.ServiceReference;
 import io.scalecube.services.api.ServiceMessage;
+import io.scalecube.services.methods.ServiceMethodInvoker;
 import java.util.List;
 
 /**
@@ -20,4 +22,10 @@ public interface ServiceRegistry {
  boolean registerService(ServiceEndpoint serviceEndpoint);
 
  ServiceEndpoint unregisterService(String endpointId);
+
+ void registerService(ServiceInfo serviceInfo);
+
+ List<ServiceInfo> listServices();
+
+ ServiceMethodInvoker getInvoker(String qualifier);
 }

services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java

@@ -1,6 +1,6 @@
 package io.scalecube.services.transport.api;
 
-import io.scalecube.services.methods.ServiceMethodRegistry;
+import io.scalecube.services.registry.api.ServiceRegistry;
 
 public interface ServiceTransport {
 
@@ -14,10 +14,10 @@ public interface ServiceTransport {
  /**
  * Provider for {@link ServerTransport}.
  *
- * @param methodRegistry methodRegistry
+ * @param serviceRegistry serviceRegistry
  * @return {@code ServerTransport} instance
  */
- ServerTransport serverTransport(ServiceMethodRegistry methodRegistry);
+ ServerTransport serverTransport(ServiceRegistry serviceRegistry);
 
  /**
  * Starts {@link ServiceTransport} instance.

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServerTransport.java

@@ -5,7 +5,7 @@
 import io.rsocket.transport.netty.server.CloseableChannel;
 import io.scalecube.net.Address;
 import io.scalecube.services.auth.Authenticator;
-import io.scalecube.services.methods.ServiceMethodRegistry;
+import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.transport.api.DataCodec;
 import io.scalecube.services.transport.api.HeadersCodec;
 import io.scalecube.services.transport.api.ServerTransport;
@@ -18,7 +18,7 @@ public class RSocketServerTransport implements ServerTransport {
  private static final Logger LOGGER = LoggerFactory.getLogger(RSocketServerTransport.class);
 
  private final Authenticator<Object> authenticator;
- private final ServiceMethodRegistry methodRegistry;
+ private final ServiceRegistry serviceRegistry;
  private final ConnectionSetupCodec connectionSetupCodec;
  private final HeadersCodec headersCodec;
  private final Collection<DataCodec> dataCodecs;
@@ -30,21 +30,21 @@ public class RSocketServerTransport implements ServerTransport {
  * Constructor for this server transport.
  *
  * @param authenticator authenticator
- * @param methodRegistry methodRegistry
+ * @param serviceRegistry serviceRegistry
  * @param connectionSetupCodec connectionSetupCodec
  * @param headersCodec headersCodec
  * @param dataCodecs dataCodecs
  * @param serverTransportFactory serverTransportFactory
  */
  public RSocketServerTransport(
  Authenticator<Object> authenticator,
- ServiceMethodRegistry methodRegistry,
+ ServiceRegistry serviceRegistry,
  ConnectionSetupCodec connectionSetupCodec,
  HeadersCodec headersCodec,
  Collection<DataCodec> dataCodecs,
  RSocketServerTransportFactory serverTransportFactory) {
  this.authenticator = authenticator;
- this.methodRegistry = methodRegistry;
+ this.serviceRegistry = serviceRegistry;
  this.connectionSetupCodec = connectionSetupCodec;
  this.headersCodec = headersCodec;
  this.dataCodecs = dataCodecs;
@@ -64,7 +64,7 @@ public ServerTransport bind() {
  RSocketServer.create()
  .acceptor(
  new RSocketServiceAcceptor(
- connectionSetupCodec, headersCodec, dataCodecs, authenticator, methodRegistry))
+ connectionSetupCodec, headersCodec, dataCodecs, authenticator, serviceRegistry))
  .payloadDecoder(PayloadDecoder.DEFAULT)
  .bind(serverTransportFactory.serverTransport())
  .doOnSuccess(channel -> serverChannel = channel)

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceAcceptor.java

@@ -17,7 +17,7 @@
 import io.scalecube.services.exceptions.ServiceUnavailableException;
 import io.scalecube.services.exceptions.UnauthorizedException;
 import io.scalecube.services.methods.ServiceMethodInvoker;
-import io.scalecube.services.methods.ServiceMethodRegistry;
+import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.transport.api.DataCodec;
 import io.scalecube.services.transport.api.HeadersCodec;
 import java.util.Collection;
@@ -36,7 +36,7 @@ public class RSocketServiceAcceptor implements SocketAcceptor {
  private final HeadersCodec headersCodec;
  private final Collection<DataCodec> dataCodecs;
  private final Authenticator<Object> authenticator;
- private final ServiceMethodRegistry methodRegistry;
+ private final ServiceRegistry serviceRegistry;
 
  /**
  * Constructor.
@@ -45,19 +45,19 @@ public class RSocketServiceAcceptor implements SocketAcceptor {
  * @param headersCodec headersCodec
  * @param dataCodecs dataCodecs
  * @param authenticator authenticator
- * @param methodRegistry methodRegistry
+ * @param serviceRegistry serviceRegistry
  */
  public RSocketServiceAcceptor(
  ConnectionSetupCodec connectionSetupCodec,
  HeadersCodec headersCodec,
  Collection<DataCodec> dataCodecs,
  Authenticator<Object> authenticator,
- ServiceMethodRegistry methodRegistry) {
+ ServiceRegistry serviceRegistry) {
  this.connectionSetupCodec = connectionSetupCodec;
  this.headersCodec = headersCodec;
  this.dataCodecs = dataCodecs;
  this.authenticator = authenticator;
- this.methodRegistry = methodRegistry;
+ this.serviceRegistry = serviceRegistry;
  }
 
  @Override
@@ -99,7 +99,7 @@ private Mono<Object> authenticate(RSocket rsocket, ConnectionSetup connectionSet
 
  private RSocket newRSocket(Object authData) {
  return new RSocketImpl(
- authData, new ServiceMessageCodec(headersCodec, dataCodecs), methodRegistry);
+ authData, new ServiceMessageCodec(headersCodec, dataCodecs), serviceRegistry);
  }
 
  private UnauthorizedException toUnauthorizedException(Throwable th) {
@@ -115,13 +115,13 @@ private static class RSocketImpl implements RSocket {
 
  private final Object authData;
  private final ServiceMessageCodec messageCodec;
- private final ServiceMethodRegistry methodRegistry;
+ private final ServiceRegistry serviceRegistry;
 
  private RSocketImpl(
- Object authData, ServiceMessageCodec messageCodec, ServiceMethodRegistry methodRegistry) {
+ Object authData, ServiceMessageCodec messageCodec, ServiceRegistry serviceRegistry) {
  this.authData = authData;
  this.messageCodec = messageCodec;
- this.methodRegistry = methodRegistry;
+ this.serviceRegistry = serviceRegistry;
  }
 
  @Override
@@ -130,7 +130,8 @@ public Mono<Payload> requestResponse(Payload payload) {
  .doOnNext(this::validateRequest)
  .flatMap(
  message -> {
- ServiceMethodInvoker methodInvoker = methodRegistry.getInvoker(message.qualifier());
+ ServiceMethodInvoker methodInvoker =
+ serviceRegistry.getInvoker(message.qualifier());
  validateMethodInvoker(methodInvoker, message);
  return methodInvoker
  .invokeOne(message)
@@ -147,7 +148,8 @@ public Flux<Payload> requestStream(Payload payload) {
  .doOnNext(this::validateRequest)
  .flatMapMany(
  message -> {
- ServiceMethodInvoker methodInvoker = methodRegistry.getInvoker(message.qualifier());
+ ServiceMethodInvoker methodInvoker =
+ serviceRegistry.getInvoker(message.qualifier());
  validateMethodInvoker(methodInvoker, message);
  return methodInvoker
  .invokeMany(message)
@@ -168,7 +170,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
  ServiceMessage message = first.get();
  validateRequest(message);
  ServiceMethodInvoker methodInvoker =
- methodRegistry.getInvoker(message.qualifier());
+ serviceRegistry.getInvoker(message.qualifier());
  validateMethodInvoker(methodInvoker, message);
  return methodInvoker
  .invokeBidirectional(messages)

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceTransport.java

@@ -9,7 +9,7 @@
 import io.scalecube.services.auth.Authenticator;
 import io.scalecube.services.auth.CredentialsSupplier;
 import io.scalecube.services.exceptions.ConnectionClosedException;
-import io.scalecube.services.methods.ServiceMethodRegistry;
+import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.transport.api.ClientTransport;
 import io.scalecube.services.transport.api.DataCodec;
 import io.scalecube.services.transport.api.HeadersCodec;
@@ -195,10 +195,10 @@ public ClientTransport clientTransport() {
  }
 
  @Override
- public ServerTransport serverTransport(ServiceMethodRegistry methodRegistry) {
+ public ServerTransport serverTransport(ServiceRegistry serviceRegistry) {
  return new RSocketServerTransport(
  authenticator,
- methodRegistry,
+ serviceRegistry,
  connectionSetupCodec,
  headersCodec,
  dataCodecs,

services/src/main/java/io/scalecube/services/Microservices.java

@@ -15,8 +15,6 @@
 import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.gateway.Gateway;
 import io.scalecube.services.gateway.GatewayOptions;
-import io.scalecube.services.methods.ServiceMethodRegistry;
-import io.scalecube.services.methods.ServiceMethodRegistryImpl;
 import io.scalecube.services.registry.ServiceRegistryImpl;
 import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.routing.RoundRobinServiceRouter;
@@ -123,7 +121,6 @@ public final class Microservices implements AutoCloseable {
  private final Map<String, String> tags;
  private final List<ServiceProvider> serviceProviders;
  private final ServiceRegistry serviceRegistry;
- private final ServiceMethodRegistry methodRegistry;
  private final Authenticator<Object> defaultAuthenticator;
  private final ServiceTransportBootstrap transportBootstrap;
  private final GatewayBootstrap gatewayBootstrap;
@@ -141,7 +138,6 @@ private Microservices(Builder builder) {
  this.tags = builder.tags;
  this.serviceProviders = builder.serviceProviders;
  this.serviceRegistry = builder.serviceRegistry;
- this.methodRegistry = builder.methodRegistry;
  this.defaultAuthenticator = builder.defaultAuthenticator;
  this.gatewayBootstrap = builder.gatewayBootstrap;
  this.discoveryBootstrap = builder.discoveryBootstrap;
@@ -234,7 +230,7 @@ private Mono<ServiceDiscoveryBootstrap> concludeDiscovery(
  }
 
  private void registerService(ServiceInfo serviceInfo) {
- methodRegistry.registerService(
+ serviceRegistry.registerService(
  ServiceInfo.from(serviceInfo)
  .errorMapperIfAbsent(defaultErrorMapper)
  .dataDecoderIfAbsent(defaultDataDecoder)
@@ -251,7 +247,6 @@ public ServiceCall call() {
  return new ServiceCall()
  .transport(transportBootstrap.clientTransport)
  .serviceRegistry(serviceRegistry)
- .methodRegistry(methodRegistry)
  .contentType(defaultContentType)
  .errorMapper(DefaultErrorMapper.INSTANCE)
  .router(Routers.getRouter(RoundRobinServiceRouter.class));
@@ -281,10 +276,6 @@ public ServiceRegistry serviceRegistry() {
  return serviceRegistry;
  }
 
- public ServiceMethodRegistry methodRegistry() {
- return methodRegistry;
- }
-
  public Address discoveryAddress() {
  return discoveryBootstrap.serviceDiscovery != null
  ? discoveryBootstrap.serviceDiscovery.address()
@@ -343,7 +334,7 @@ private Mono<Void> applyBeforeDestroy() {
  return Mono.defer(
  () ->
  Mono.whenDelayError(
- methodRegistry.listServices().stream()
+ serviceRegistry.listServices().stream()
  .map(ServiceInfo::serviceInstance)
  .map(s -> Mono.fromRunnable(() -> Injector.processBeforeDestroy(this, s)))
  .collect(Collectors.toList())));
@@ -363,7 +354,6 @@ public static final class Builder {
  private Map<String, String> tags = new HashMap<>();
  private final List<ServiceProvider> serviceProviders = new ArrayList<>();
  private ServiceRegistry serviceRegistry = new ServiceRegistryImpl();
- private ServiceMethodRegistry methodRegistry = new ServiceMethodRegistryImpl();
  private Authenticator<Object> defaultAuthenticator = null;
  private final ServiceDiscoveryBootstrap discoveryBootstrap = new ServiceDiscoveryBootstrap();
  private ServiceTransportBootstrap transportBootstrap = new ServiceTransportBootstrap();
@@ -428,11 +418,6 @@ public Builder serviceRegistry(ServiceRegistry serviceRegistry) {
  return this;
  }
 
- public Builder methodRegistry(ServiceMethodRegistry methodRegistry) {
- this.methodRegistry = methodRegistry;
- return this;
- }
-
  public Builder discovery(ServiceDiscoveryFactory discoveryFactory) {
  this.discoveryBootstrap.operator(options -> options.discoveryFactory(discoveryFactory));
  return this;
@@ -709,7 +694,7 @@ private ServiceTransportBootstrap start(Microservices microservices) {
  try {
  try {
  serviceTransport = serviceTransport.start();
- serverTransport = serviceTransport.serverTransport(microservices.methodRegistry).bind();
+ serverTransport = serviceTransport.serverTransport(microservices.serviceRegistry).bind();
  } catch (Exception e) {
  throw new RuntimeException(e);
  }