Producer.test

import akka.stream.javadsl.Source;
import akka.stream.testkit.TestSubscriber;
import akka.stream.testkit.javadsl.TestSink;
import com.lightbend.lagom.javadsl.persistence.PersistentEntityRegistry;
import com.lightbend.lagom.javadsl.testkit.ServiceTest;
import org.junit.BeforeClass;
import org.junit.Test;
import org.knoldus.hello.api.GreetingMessage;
import org.knoldus.hello.api.HelloEvent;
import org.knoldus.hello.api.HelloService;
import scala.concurrent.duration.FiniteDuration;

import java.util.concurrent.TimeUnit;

import static com.lightbend.lagom.javadsl.testkit.ServiceTest.defaultSetup;
import static com.lightbend.lagom.javadsl.testkit.ServiceTest.startServer;
import static org.junit.Assert.assertEquals;

public class ProducerTest {
    
    private static ServiceTest.TestServer testServer;
    
    private static HelloService service;
    
    private static PersistentEntityRegistry persistentEntityRegistry;
    
    @BeforeClass
    public static void setUp() {
        testServer = startServer(defaultSetup().withCassandra(true));
        service = testServer.client(HelloService.class);
        persistentEntityRegistry = testServer.injector().instanceOf(PersistentEntityRegistry.class);
    }
    
    @Test
    public void shouldEmitGreetingsMessageWhenHelloEntityEmitsEnEvent() {
        
        service.useGreeting("user_id").invoke(new GreetingMessage("hi")).toCompletableFuture().join();
        Source<HelloEvent, ?> source = service.helloEvents().subscribe().atMostOnceSource();
        
        TestSubscriber.Probe<HelloEvent> probe =
                source.runWith(TestSink.probe(testServer.system()), testServer.materializer());
        
        HelloEvent helloEvent = probe.request(1).expectNext(FiniteDuration.apply(25, TimeUnit.SECONDS));
        
        assertEquals("user_id", helloEvent.getName());
    }
}