Initial commit of an Rx based test

build.gradle

@@ -52,6 +52,8 @@ dependencies {
   testCompile( "io.vertx:testtools:$toolsVersion" ) {
     transitive = false
   }
+  provided "org.codehaus.groovy:groovy-all:$groovyVersion"
+  testCompile "io.vertx:mod-rxjava:$rxVersion"
 }
 
 test {

gradle.properties

@@ -38,3 +38,6 @@ c3p0Version=0.9.5-pre3
 dbutilsVersion=1.5
 hsqldbVersion=2.2.9
 metricsVersion=3.0.0-BETA1
+
+rxVersion=1.0.0-beta1-SNAPSHOT
+groovyVersion=2.1.4

src/test/java/com/bloidonia/vertx/mods/integration/JavaBasedRxTest.java

@@ -0,0 +1,117 @@
+package com.bloidonia.vertx.mods.integration ;
+
+import io.vertx.rxcore.java.eventbus.RxEventBus ;
+import io.vertx.rxcore.java.eventbus.RxMessage ;
+import org.junit.Test ;
+import org.vertx.java.core.AsyncResult ;
+import org.vertx.java.core.AsyncResultHandler ;
+import org.vertx.java.core.json.JsonObject ;
+import org.vertx.java.core.json.JsonArray ;
+import java.util.ArrayList ;
+import org.vertx.testtools.TestVerticle ;
+import rx.Observable ;
+import rx.util.functions.Action1 ;
+import rx.util.functions.Func1 ;
+import rx.util.functions.Func2 ;
+
+import static org.vertx.testtools.VertxAssert.* ;
+
+public class JavaBasedRxTest extends TestVerticle {
+
+  private RxEventBus rxEventBus ;
+
+  private void appReady() {
+    super.start() ;
+  }
+
+  public void start() {
+    rxEventBus = new RxEventBus( vertx.eventBus() ) ;
+    JsonObject config = new JsonObject() {{
+      putString( "address", "test.persistor" ) ;
+      putString( "url", "jdbc:hsqldb:mem:testdb?shutdown=true" ) ;
+    }} ;
+
+    container.deployModule(System.getProperty("vertx.modulename"), config, 1, new AsyncResultHandler<String>() {
+      @Override
+      public void handle(AsyncResult<String> event1) {
+        appReady() ;
+      }
+    } ) ;
+  }
+
+
+  @Test
+  public void testCreateInsertSelectAndDrop() {
+    // Last subscription in the chain; drop the table
+    final Action1<RxMessage<JsonObject>> dropSubscription = new Action1<RxMessage<JsonObject>>() {
+      @Override
+      public void call( RxMessage<JsonObject> message ) {
+        assertEquals( message.body().getString( "status" ), "ok" ) ;
+        testComplete() ;
+      }
+    } ;
+
+    // Thirdly we select the data from the table
+    final Action1<RxMessage<JsonObject>> selectSubscription = new Action1<RxMessage<JsonObject>>() {
+      @Override
+      public void call( RxMessage<JsonObject> message ) {
+        JsonArray result = message.body().getArray( "result" ) ;
+        assertEquals( ((JsonObject)result.get( 0 )).getNumber( "AGE" ), 29 ) ;
+        assertEquals( ((JsonObject)result.get( 1 )).getNumber( "AGE" ), 42 ) ;
+        assertEquals( ((JsonObject)result.get( 2 )).getNumber( "AGE" ), 65 ) ;
+
+        final JsonObject dropMsg = new JsonObject() {{
+          putString( "action", "execute" ) ;
+          putString( "stmt", "DROP TABLE test" ) ;
+        }} ;
+        rxEventBus.send( "test.persistor", dropMsg ).subscribe( dropSubscription ) ;
+      }
+    } ;
+
+    // Secondly we insert some data
+    final Action1<RxMessage<JsonObject>> insertSubscription = new Action1<RxMessage<JsonObject>>() {
+      @Override
+      public void call( RxMessage<JsonObject> message ) {
+        JsonArray result = message.body().getArray( "result" ) ;
+        assertEquals( ((JsonObject)result.get( 0 )).getNumber( "ID" ), 1 ) ;
+        assertEquals( ((JsonObject)result.get( 1 )).getNumber( "ID" ), 2 ) ;
+        assertEquals( ((JsonObject)result.get( 2 )).getNumber( "ID" ), 3 ) ;
+
+        final JsonObject selectMsg = new JsonObject() {{
+          putString( "action", "select" ) ;
+          putString( "stmt",   "SELECT * FROM test ORDER BY age ASC" ) ;
+        }} ;
+        rxEventBus.send( "test.persistor", selectMsg ).subscribe( selectSubscription ) ;
+      }
+    } ;
+
+    // Initially, we create the table
+    final Action1<RxMessage<JsonObject>> createSubscription = new Action1<RxMessage<JsonObject>>() {
+      @Override
+      public void call( RxMessage<JsonObject> message ) {
+        assertEquals( message.body().getString( "status" ), "ok" ) ;
+
+        final JsonArray insertValues = new JsonArray( new ArrayList() {{
+          add( new ArrayList() {{ add( "tim" )  ; add( 65 ) ; }} ) ;
+          add( new ArrayList() {{ add( "dave" ) ; add( 29 ) ; }} ) ;
+          add( new ArrayList() {{ add( "mike" ) ; add( 42 ) ; }} ) ;
+        }} ) ;
+        final JsonObject insertMsg = new JsonObject() {{
+          putString( "action", "insert" ) ;
+          putString( "stmt",   "INSERT INTO test ( name, age ) VALUES ( ?, ? )" ) ;
+          putArray(  "values", insertValues ) ;
+        }} ;
+        rxEventBus.send( "test.persistor", insertMsg ).subscribe( insertSubscription ) ;
+      }
+    } ;
+
+    // Kick off the chain, and subscribe
+    final JsonObject createMsg = new JsonObject() {{
+      putString( "action", "execute" ) ;
+      putString( "stmt", "CREATE TABLE IF NOT EXISTS test ( id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 1 INCREMENT BY 1) NOT NULL, name VARCHAR(80), age INTEGER, CONSTRAINT testid PRIMARY KEY ( id ) )" ) ;
+    }} ;
+
+    final Observable<RxMessage<JsonObject>> create = rxEventBus.send( "test.persistor", createMsg ) ;
+    create.subscribe( createSubscription ) ;
+  }
+}