ServiceMethodValidationTest.java

package io.scalecube.services;

import static org.junit.jupiter.api.Assertions.assertTrue;

import io.scalecube.services.annotations.Service;
import io.scalecube.services.annotations.ServiceMethod;
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;

public class ServiceMethodValidationTest {

  @Test
  void testServiceMethodValidation() throws InterruptedException {
    Microservices serviceProvider =
        Microservices.builder()
            .discovery("serviceProvider", ScalecubeServiceDiscovery::new)
            .transport(RSocketServiceTransport::new)
            .services(new Impl())
            .startAwait();

    Microservices serviceConsumer = Microservices.builder()
        .discovery(
            "serviceConsumer",
            serviceEndpoint ->
                new ScalecubeServiceDiscovery(serviceEndpoint)
                    .membership(
                        opt ->
                            opt.seedMembers(
                                serviceProvider.discovery("serviceProvider").address())))
        .transport(RSocketServiceTransport::new)
        .startAwait();

    Invalid api = serviceConsumer.call().api(Invalid.class);
    CountDownLatch latch = new CountDownLatch(1);
    Flux<String> input = Flux.range(0, 10).map(String::valueOf)
        .doOnSubscribe(s -> latch.countDown());
    api.invalidRequestChannel(input);
    boolean valid = latch.await(5, TimeUnit.SECONDS);
    assertTrue(valid);
  }

  @Service
  interface Invalid {

    @ServiceMethod
    void invalidRequestChannel(Flux<String> input);
  }

  static class Impl implements Invalid {

    @Override
    public void invalidRequestChannel(Flux<String> input) {
      input.subscribe(System.out::println);
    }
  }
}