Merge pull request #8 from pandachris/BAEL-2283

BAEL-2283

pom.xml

@@ -483,6 +483,7 @@
                 
                 <module>reactor-core</module>
                 <module>resteasy</module>
+								<module>rsocket</module>
                 <module>rxjava</module>
                 <module>rxjava-2</module>
                 <module>rabbitmq</module>
@@ -1036,6 +1037,7 @@
                 
                 <module>reactor-core</module>
                 <module>resteasy</module>
+								<module>rsocket</module>
                 <module>rxjava</module>
                 <module>rxjava-2</module>
                 <module>rabbitmq</module>

rsocket/src/test/java/com/baeldung/rsocket/RSocketIntegrationTest.java

@@ -1,5 +1,6 @@
 package com.baeldung.rsocket;
 
+import java.util.ArrayList;
 import java.util.List;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -29,7 +30,7 @@ public class RSocketIntegrationTest {
     }
 
     @Test
-    public void testRequestResponse() {
+    public void whenSendingAString_thenRevceiveTheSameString() {
         ReqResClient client = new ReqResClient();
         String string = "Hello RSocket";
         assertEquals(string, client.callBlocking(string));
@@ -37,20 +38,27 @@ public class RSocketIntegrationTest {
     }
 
     @Test
-    public void testFNFAndRequestStream() {
+    public void whenSendingStream_thenReceiveTheSameStream() {
         // create the client that pushes data to the server and start sending
         FireNForgetClient fnfClient = new FireNForgetClient();
-        // get the data that is used by the client
-        List<Float> data = fnfClient.getData();
-
         // create a client to read a stream from the server and subscribe to events
         ReqStreamClient streamClient = new ReqStreamClient();
+
+        // get the data that is used by the client
+        List<Float> data = fnfClient.getData();
+        // create a place to count the results
+        List<Float> dataReceived = new ArrayList<>();
+
         // assert that the data received is the same as the data sent
         Disposable subscription = streamClient.getDataStream()
-                .index()
+          .index()
-                .doOnNext(element -> assertEquals(data.get(element.getT1().intValue()), element.getT2()))
+          .subscribe(
-                .count()
+            tuple -> {
-                .subscribe(count -> assertEquals(data.size(), count.intValue()));
+                assertEquals("Wrong value", data.get(tuple.getT1().intValue()), tuple.getT2());
+                dataReceived.add(tuple.getT2());
+            },
+            err -> LOG.error(err.getMessage())
+          );
 
         // start sending the data
         fnfClient.sendData();
@@ -59,10 +67,13 @@ public class RSocketIntegrationTest {
         try { Thread.sleep(500); } catch (Exception x) {}
         subscription.dispose();
         fnfClient.dispose();
+        
+        // verify the item count
+        assertEquals("Wrong data count received", data.size(), dataReceived.size());
     }
 
     @Test
-    public void testChannel() {
+    public void whenRunningChannelGame_thenLogTheResults() {
         ChannelClient client = new ChannelClient();
         client.playGame();
         client.dispose();