Add ribbon/proxy module.

build.gradle

@@ -5,7 +5,9 @@ buildscript {
 }
 
 allprojects {
-    repositories { mavenCentral() }
+    repositories {
+        mavenCentral()
+    }
 }
 
 apply from: file('gradle/convention.gradle')
@@ -17,9 +19,21 @@ subprojects {
 
     sourceSets.test.java.srcDir 'src/main/java'
 
+    dependencies {
+        compile "io.netty:netty-common:${netty_version}"
+        compile "io.netty:netty-transport:${netty_version}"
+        compile "com.netflix.rxnetty:rx-netty:${rxnetty_version}"
+        compile "com.netflix.rxnetty:rx-netty-servo:${rxnetty_version}"
+        compile "org.slf4j:slf4j-api:${slf4j_version}"
+        runtime "org.slf4j:slf4j-simple:${slf4j_version}"
+        compile "com.netflix.numerus:numerus:1.1"
+        testCompile "junit:junit-dep:4.10"
+        testCompile "org.mockito:mockito-core:1.8.5"
+    }
+
     tasks.withType(Javadoc).each {
         it.classpath = sourceSets.main.compileClasspath
-    }
+	}
 }
 
 if (JavaVersion.current().isJava8Compatible()) {

gradle.properties

@@ -1 +1,6 @@
 version=0.1-SNAPSHOT
+
+slf4j_version=1.7.6
+netty_version=4.0.21.Final
+rxnetty_version=0.3.8
+ribbon_version=2.0-RC1

settings.gradle

@@ -4,4 +4,5 @@ include 'ws-backend-mock', \
 'ws-impls:ws-impl-utils', \
 'ws-impls:ws-java-servlet-blocking', \
 'ws-impls:ws-java-jetty', \
-'ws-impls:ws-java-rxnetty'
+'ws-impls:ws-java-rxnetty', \
+'ws-impls:ws-java-ribbon'

ws-backend-mock/build.gradle

@@ -6,13 +6,8 @@ sourceCompatibility = JavaVersion.VERSION_1_8
 targetCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
-	compile 'com.netflix.rxnetty:rx-netty:0.3.8'
     compile 'com.netflix.numerus:numerus:1.1'
-	compile 'com.netflix.rxjava:rxjava-core:0.18.2'
 	compile 'org.codehaus.jackson:jackson-core-asl:1.9.2'
-    compile 'org.slf4j:slf4j-api:1.7.0'
-	testCompile 'junit:junit-dep:4.10'
-	testCompile 'org.mockito:mockito-core:1.8.5'
 }
 
 /**

ws-backend-mock/src/main/java/perf/backend/MockResponse.java

@@ -4,7 +4,6 @@
 import io.netty.buffer.ByteBufOutputStream;
 import io.netty.buffer.Unpooled;
 import org.codehaus.jackson.JsonFactory;
-import org.codehaus.jackson.JsonGenerationException;
 import org.codehaus.jackson.JsonGenerator;
 import rx.Observable;
 import rx.Scheduler.Worker;
@@ -44,16 +43,10 @@ private MockResponse() {
     }
 
     /**
-     * 
-     * @param id
-     *            ID from client used to assert correct client/server interaction
-     * @param delay
-     *            How long to delay delivery to simulate server-side latency
-     * @param itemSize
-     *            Length of each item String.
-     * @param numItems
-     *            Number of items in response.
-     * 
+     * @param id       ID from client used to assert correct client/server interaction
+     * @param delay    How long to delay delivery to simulate server-side latency
+     * @param itemSize Length of each item String.
+     * @param numItems Number of items in response.
      * @return String json
      */
     public static Observable<ByteBuf> generateJson(long id, int delay, int itemSize, int numItems) {
@@ -62,33 +55,8 @@ public static Observable<ByteBuf> generateJson(long id, int delay, int itemSize,
             subscriber.add(worker);
             worker.schedule(() -> {
                 try {
-                    ByteBuf buffer = Unpooled.buffer();
-                    ByteBufOutputStream jsonAsBytes = new ByteBufOutputStream(buffer);
-                    JsonGenerator json = jsonFactory.createJsonGenerator(jsonAsBytes);
-
-                    json.writeStartObject();
-
-                    // manipulate the ID such that we can know the response is from the server (client will know the logic)
-                    long responseKey = getResponseKey(id);
-
-                    json.writeNumberField("responseKey", responseKey);
-
-                    json.writeNumberField("delay", delay);
-                    if (itemSize > MAX_ITEM_LENGTH) {
-                        throw new IllegalArgumentException("itemSize can not be larger than: " + MAX_ITEM_LENGTH);
-                    }
-                    json.writeNumberField("itemSize", itemSize);
-                    json.writeNumberField("numItems", numItems);
-
-                    json.writeArrayFieldStart("items");
-                    for (int i = 0; i < numItems; i++) {
-                        json.writeString(RAW_ITEM_LONG.substring(0, itemSize));
-                    }
-                    json.writeEndArray();
-                    json.writeEndObject();
-                    json.close();
-
-                    subscriber.onNext(buffer);
+                    ByteBuf byteBuf = createJsonResponse(id, delay, itemSize, numItems, false);
+                    subscriber.onNext(byteBuf);
                     subscriber.onCompleted();
                 } catch (Exception e) {
                     subscriber.onError(e);
@@ -97,7 +65,49 @@ public static Observable<ByteBuf> generateJson(long id, int delay, int itemSize,
         });
     }
 
-    /* package */static long getResponseKey(long id) {
+    public static Observable<ByteBuf> generateFallbackJson(long id, int delay, int itemSize, int numItems) {
+        try {
+            ByteBuf byteBuf = createJsonResponse(id, delay, itemSize, numItems, true);
+            return Observable.just(byteBuf);
+        } catch (IOException e) {
+            // We do not expect to get here
+            e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static ByteBuf createJsonResponse(long id, int delay, int itemSize, int numItems, boolean fallback) throws IOException {
+        ByteBuf buffer = Unpooled.buffer();
+        ByteBufOutputStream jsonAsBytes = new ByteBufOutputStream(buffer);
+        JsonGenerator json = jsonFactory.createJsonGenerator(jsonAsBytes);
+
+        json.writeStartObject();
+
+        // manipulate the ID such that we can know the response is from the server (client will know the logic)
+        long responseKey = getResponseKey(id);
+
+        json.writeNumberField("responseKey", responseKey);
+
+        json.writeNumberField("delay", delay);
+        if (itemSize > MAX_ITEM_LENGTH) {
+            throw new IllegalArgumentException("itemSize can not be larger than: " + MAX_ITEM_LENGTH);
+        }
+        json.writeNumberField("itemSize", itemSize);
+        json.writeNumberField("numItems", numItems);
+        json.writeBooleanField("fallback", fallback);
+
+        json.writeArrayFieldStart("items");
+        for (int i = 0; i < numItems; i++) {
+            json.writeString(RAW_ITEM_LONG.substring(0, itemSize));
+        }
+        json.writeEndArray();
+        json.writeEndObject();
+        json.close();
+        return jsonAsBytes.buffer();
+    }
+
+    /* package */
+    static long getResponseKey(long id) {
         return (id / 37 + 5739375) * 7;
     }
 }

ws-client/build.gradle

@@ -6,14 +6,9 @@ sourceCompatibility = JavaVersion.VERSION_1_8
 targetCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
-	compile 'com.netflix.rxnetty:rx-netty:0.3.8'
-	compile 'com.netflix.rxjava:rxjava-core:0.18.2'
-	compile 'com.netflix.numerus:numerus:1.1'
   compile 'org.codehaus.jackson:jackson-core-asl:1.9.2'
   compile 'org.codehaus.jackson:jackson-mapper-asl:1.9.2'
   compile 'commons-cli:commons-cli:1.2'
-	testCompile 'junit:junit-dep:4.10'
-	testCompile 'org.mockito:mockito-core:1.8.5'
 }
 
 

ws-impls/ws-impl-utils/build.gradle

@@ -2,15 +2,11 @@ apply plugin: 'java'
 apply plugin: 'eclipse'
 
 dependencies {
-    compile 'org.slf4j:slf4j-api:1.7.0'
     compile 'org.codehaus.jackson:jackson-core-asl:1.9.2'
-    compile 'junit:junit-dep:4.10'
     compile 'javax.servlet:javax.servlet-api:3.1.0'
-    compile 'com.netflix.rxjava:rxjava-core:0.18+'
     compile 'com.google.guava:guava:15.0'
     compile 'org.apache.commons:commons-lang3:3.1'
-    compile 'io.netty:netty-common:4.0.14.Final'
-    compile 'io.netty:netty-transport:4.0.14.Final'
+    compile 'junit:junit-dep:4.10'
 }
 
 eclipse {

ws-impls/ws-impl-utils/src/main/java/perf/test/utils/BackendResponse.java

@@ -1,21 +1,21 @@
 package perf.test.utils;
 
-import static junit.framework.Assert.assertEquals;
-
-import java.io.IOException;
-import java.io.InputStream;
-
 import org.codehaus.jackson.JsonFactory;
 import org.codehaus.jackson.JsonParser;
 import org.codehaus.jackson.JsonToken;
 import org.junit.Test;
-
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Scheduler;
 import rx.Subscriber;
 import rx.schedulers.Schedulers;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+
+import static junit.framework.Assert.*;
+
 /**
  * @author Nitesh Kant ([email protected])
  */
@@ -26,13 +26,19 @@ public class BackendResponse {
     private final int numItems;
     private final int itemSize;
     private final String[] items;
+    private final boolean fallback;
 
     public BackendResponse(long responseKey, int delay, int numItems, int itemSize, String[] items) {
+        this(responseKey, delay, numItems, itemSize, items, false);
+    }
+
+    public BackendResponse(long responseKey, int delay, int numItems, int itemSize, String[] items, boolean fallback) {
         this.responseKey = responseKey;
         this.delay = delay;
         this.numItems = numItems;
         this.itemSize = itemSize;
         this.items = items;
+        this.fallback = fallback;
     }
 
     public static BackendResponse fromJson(JsonFactory jsonFactory, byte[] content) throws Exception {
@@ -83,6 +89,7 @@ public static BackendResponse parseBackendResponse(JsonParser parser) throws IOE
             int delay = 0;
             int numItems = 0;
             int itemSize = 0;
+            boolean fallback = false;
             String[] items = null;
             JsonToken current;
 
@@ -98,6 +105,8 @@ public static BackendResponse parseBackendResponse(JsonParser parser) throws IOE
                     itemSize = parser.getIntValue();
                 } else if (fieldName.equals("numItems")) {
                     numItems = parser.getIntValue();
+                } else if(fieldName.equals("fallback")) {
+                    fallback = parser.getBooleanValue();
                 } else if (fieldName.equals("items")) {
                     // expect numItems to be populated before hitting this
                     if (numItems == 0) {
@@ -117,8 +126,8 @@ public static BackendResponse parseBackendResponse(JsonParser parser) throws IOE
 
                 }
             }
-            
-            return new BackendResponse(responseKey, delay, numItems, itemSize, items);
+
+            return new BackendResponse(responseKey, delay, numItems, itemSize, items, fallback);
         } finally {
             parser.close();
         }
@@ -144,16 +153,33 @@ public String[] getItems() {
         return items;
     }
 
+    public boolean isFallback() {
+        return fallback;
+    }
+
+    @Override
+    public String toString() {
+        return "BackendResponse{" +
+                "responseKey=" + responseKey +
+                ", delay=" + delay +
+                ", numItems=" + numItems +
+                ", itemSize=" + itemSize +
+                ", fallback=" + fallback +
+                ", items=" + Arrays.toString(items) +
+                '}';
+    }
+
     public static class UnitTest {
 
         @Test
         public void testJsonParse() throws Exception {
             JsonFactory jsonFactory = new JsonFactory();
-            BackendResponse r = BackendResponse.fromJson(jsonFactory, "{ \"responseKey\": 9999, \"delay\": 50, \"itemSize\": 128, \"numItems\": 2, \"items\": [ \"Lorem\", \"Ipsum\" ]}");
+            BackendResponse r = fromJson(jsonFactory, "{ \"responseKey\": 9999, \"delay\": 50, \"fallback\": false, \"itemSize\": 128, \"numItems\": 2, \"items\": [ \"Lorem\", \"Ipsum\" ]}");
             assertEquals(9999, r.getResponseKey());
             assertEquals(50, r.getDelay());
             assertEquals(128, r.getItemSize());
             assertEquals(2, r.getNumItems());
+            assertFalse(r.fallback);
             String[] items = r.getItems();
             assertEquals(2, items.length);
             assertEquals("Lorem", items[0]);

ws-impls/ws-java-jetty/build.gradle

@@ -3,14 +3,11 @@ apply plugin: 'eclipse'
 apply plugin: 'application'
 
 dependencies {
-    compile 'org.slf4j:slf4j-api:1.7.0'
-    runtime 'org.slf4j:slf4j-simple:1.7.0'
     compile 'org.eclipse.jetty:jetty-server:9.0.3.v20130506'
     compile 'org.eclipse.jetty:jetty-client:9.0.3.v20130506'
     compile 'org.eclipse.jetty:jetty-continuation:9.0.3.v20130506'
     compile 'org.codehaus.jackson:jackson-core-asl:1.9.2'
     compile 'com.google.guava:guava:r05'
-    provided 'junit:junit-dep:4.10'
     compile project(':ws-impls:ws-impl-utils')
 }
 

ws-impls/ws-java-ribbon/build.gradle

@@ -0,0 +1,29 @@
+apply plugin: "eclipse"
+apply plugin: "idea"
+apply plugin: "java"
+apply plugin: "application"
+
+sourceCompatibility = JavaVersion.VERSION_1_8
+targetCompatibility = JavaVersion.VERSION_1_8
+
+dependencies {
+    compile "com.netflix.ribbon:ribbon:${ribbon_version}"
+    compile project(":ws-impls:ws-impl-utils")
+    compile project(":ws-backend-mock")
+}
+
+/**
+ * ../../gradlew start
+ * ../../gradlew start -P'a=EVENTLOOPS PORT BACKEND_SERVER_LIST'
+ * ../../gradlew start -P'a=1 8888 host1:8989,host2:8989'
+ */
+task start(type:JavaExec) {
+    main = "perf.test.ribbon.StartServer"
+    classpath = sourceSets.main.runtimeClasspath
+//    jvmArgs(['-Dorg.slf4j.simpleLogger.defaultLogLevel=debug'])
+    if (project.hasProperty('a')) {
+        args(a.split(' '))
+    }
+}
+
+mainClassName = "perf.test.ribbon.StartServer"