fix host to 0.0.0.0 and tests for plaintext, json, db, update, query and fortune (#4168)

* Adding GreenLightning

* fix missing quote

create object for json production

* turn off telemetry for default test

* allow any external domain or ip

* Update to next version and template construction

* Revert "Update to next version and template construction"

This reverts commit da4fbc3b085f6e3b3bb50283d4d6e79efffdd3bb.

* Revert "Revert "Update to next version and template construction""

This reverts commit eb05517a192554caec7c5f690eeaaab171a549b6.

* Revert "Update to next version and template construction"

This reverts commit da4fbc3b085f6e3b3bb50283d4d6e79efffdd3bb.

* fix host ip to 0.0.0.0 to take load

added database read tests

* response large enough for multi db

remove epoll, it was not helping

added required headers

* update list of which tests we have implmented

* removed unused dependency

* remove unrequired 16G memory grab to get past travis check.

* Update to next GL version to fix overload issue past 1.5M rps

Refine template method to remove dead argument

* simplify arg parse

fixed issue with multi db under heavy load

* updat to atomic int

* Removed support for MultiTest, not stable at this time.

* added clean for safety

* remove dead code

* remove old comments

* [ci fw-only Java/greenlightning]

added comment

* [ci fw-only Java/greenlightning]

Added tests for remaining multi, update and fortunes

* [ci fw-only Java/greenlightning]

disable update test, seems to be missing some writes.

* [ci fw-only Java/greenlightning]

re-test of DBUpdate

* [ci fw-only Java/greenlightning]

disabled multi and update while tracking issue

* fixed muti paylod response JSON dups.

* [ci fw-only Java/greenlightning]

narrow building

* [ci skip] update readme

frameworks/Java/greenlightning/README.md

@@ -11,13 +11,12 @@ The tests were run with:
 * [Software](https://oci-pronghorn.gitbook.io/greenlightning/)
 * [Example](https://github.com/oci-pronghorn/GreenLightning/tree/master/slipstream)
 
-## Test URLs
-### JSON
-
-http://localhost:8080/json
-
-### PLAINTEXT
-
-http://localhost:8080/plaintext
+### Implemented benchmarks
+- [x] JSON serialization
+- [x] Single query
+- [x] Multiple queries
+- [x] Fortunes
+- [x] Data updates
+- [x] Plaintext
 
 

frameworks/Java/greenlightning/benchmark_config.json

@@ -5,14 +5,18 @@
       "default": {
         "json_url": "/json",
         "plaintext_url": "/plaintext",
+        "db_url": "/db",
+        "query_url": "/queries?queries=",
+        "update_url": "/updates?queries=",
+        "fortune_url": "/fortunes",
         "port": 8080,
         "approach": "Realistic",
         "classification": "Micro",
-        "database": "None",
+        "database": "Postgres",
         "framework": "GreenLightning",
         "language": "Java",
         "flavor": "None",
-        "orm": "None",
+        "orm": "Raw",
         "platform": "None",
         "webserver": "None",
         "os": "Linux",

frameworks/Java/greenlightning/greenlightning.dockerfile

@@ -1,8 +1,9 @@
 FROM maven:3.5.3-jdk-10-slim as maven
-WORKDIR /greenlightning
+
+WORKDIR /greenlightning    
 COPY pom.xml pom.xml
 COPY src src
-RUN mvn clean install -q -U
+RUN mvn clean install -q
 
 FROM openjdk:10-jre-slim
 WORKDIR /greenlightning

frameworks/Java/greenlightning/pom.xml

@@ -8,15 +8,13 @@
 
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-		<maven.compiler.source>1.8</maven.compiler.source>
-		<maven.compiler.target>1.8</maven.compiler.target>
 	</properties>
 
 	<dependencies>
 		<dependency>
 			<groupId>com.ociweb</groupId>
 			<artifactId>greenlightning</artifactId>
-			<version>[1.0.6,1.1)</version>
+			<version>1.0.12</version>
 		</dependency>
 		<dependency>
 			<groupId>org.slf4j</groupId>
@@ -36,12 +34,15 @@
 			<type>jar</type>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+		    <groupId>io.reactiverse</groupId>
+		    <artifactId>reactive-pg-client</artifactId>
+		    <version>0.10.6</version>
+		</dependency>
 	</dependencies>
 
 	<build>
-
 		<plugins>
-
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-compiler-plugin</artifactId>
@@ -50,6 +51,8 @@
 					<compilerArguments>
 						<profile>compact1</profile>
 					</compilerArguments>
+					<source>1.8</source>
+					<target>1.8</target>
 				</configuration>
 			</plugin>
 

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/DBRest.java

@@ -0,0 +1,274 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import com.ociweb.gl.api.GreenRuntime;
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.gl.api.PubSubMethodListener;
+import com.ociweb.gl.api.RestMethodListener;
+import com.ociweb.gl.api.TickListener;
+import com.ociweb.json.encode.JSONRenderer;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.PgPool;
+import io.reactiverse.pgclient.PgPoolOptions;
+import io.reactiverse.pgclient.Tuple;
+
+public class DBRest implements RestMethodListener, PubSubMethodListener, TickListener {
+
+	private final transient PgPoolOptions options;
+	private transient PgPool pool;
+	private final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
+	private final ObjectPipe<ResultObject> inFlight;
+		
+	public DBRest(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int maxResponseCount, int maxResponseSize) {
+		this.options = options;	
+		this.inFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		this.service = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);
+	}		
+	
+	private PgPool pool() {
+		if (null==pool) {
+			pool = PgClient.pool(options);
+		}
+		return pool;
+	}
+	
+	private int randomValue() {
+		return 1+localRandom.nextInt(10000);
+	}		
+
+	public boolean multiRestRequest(HTTPRequestReader request) { 
+
+		final int queries;
+		if (Struct.DB_MULTI_ROUTE_INT == request.getRouteAssoc() ) {		
+			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
+		} else {
+			queries = 1;
+		}
+		
+	
+		if (inFlight.hasRoomFor(queries)) {
+			
+			
+			int q = queries;
+			while (--q >= 0) {
+				
+					final ResultObject target = inFlight.headObject();
+					
+					//already released but not published yet: TODO: we have a problem here!!!
+					assert(null!=target && -1==target.getStatus()) : "found status "+target.getStatus()+" on query "+q+" of "+queries ; //must block that this has been consumed?? should head/tail rsolve.
+									
+					target.setConnectionId(request.getConnectionId());
+					target.setSequenceId(request.getSequenceCode());
+					assert(target.getStatus()==-1);//waiting for work
+					target.setStatus(-2);//out for work	
+					target.setGroupSize(queries);
+				
+					pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
+							if (r.succeeded()) {
+								
+								PgIterator resultSet = r.result().iterator();
+						        Tuple row = resultSet.next();			        
+						        
+						        target.setId(row.getInteger(0));
+						        target.setResult(row.getInteger(1));					
+								target.setStatus(200);
+								
+							} else {
+								System.out.println("fail: "+r.cause().getLocalizedMessage());
+								target.setStatus(500); 
+							}				
+						});	
+								
+					inFlight.moveHeadForward(); //always move to ensure this can be read.
+			
+			}
+				
+			return true;
+		} else {
+			return false;
+		}	
+	}
+
+	
+
+	
+	public boolean singleRestRequest(HTTPRequestReader request) { 
+
+		final ResultObject target = inFlight.headObject();
+		if (null!=target && -1==target.getStatus()) {
+			target.setConnectionId(request.getConnectionId());
+			target.setSequenceId(request.getSequenceCode());
+			assert(target.getStatus()==-1);//waiting for work
+			target.setStatus(-2);//out for work	
+			target.setGroupSize(0);//do not put in a list so mark as 0.
+		
+			pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
+					if (r.succeeded()) {
+						
+						PgIterator resultSet = r.result().iterator();
+				        Tuple row = resultSet.next();			        
+				        
+				        target.setId(row.getInteger(0));
+				        target.setResult(row.getInteger(1));					
+						target.setStatus(200);
+						
+					} else {
+						System.out.println("fail: "+r.cause().getLocalizedMessage());
+						target.setStatus(500);
+					}				
+				});
+
+			
+			inFlight.moveHeadForward(); //always move to ensure this can be read.
+			return true;
+		} else {
+			return false;//can not pick up new work now			
+		}
+	}
+
+
+	
+	////////////////////////////////////
+	////////////////////////////////////
+	
+	private final JSONRenderer<List<ResultObject>> multiTemplate = new JSONRenderer<List<ResultObject>>()
+	    	  .array((o,i) -> i<o.size()?o:null)
+		          .startObject((o, i) -> o.get(i))
+					.integer("id", o -> o.getId() )
+					.integer("randomNumber", o -> o.getResult())
+		          .endObject();
+	
+	private final JSONRenderer<ResultObject> singleTemplate = new JSONRenderer<ResultObject>()
+		   	  .startObject()
+				.integer("id", o -> o.getId() )
+				.integer("randomNumber", o -> o.getResult())
+	          .endObject();
+	
+	private boolean collectionPending = false;
+
+	//this collector is for the multi db test so we can collect all the objects until we have them all for 
+	//the request we are currently sending back
+	private final List<ResultObject> collector = new ArrayList<ResultObject>();
+	private final HTTPResponseService service;
+
+
+	@Override
+	public void tickEvent() { 
+		
+		ResultObject temp = inFlight.tailObject();
+		while (isReady(temp)) {			
+			if (consumeResultObject(temp)) {
+				temp = inFlight.tailObject();
+			} else {
+				break;
+			}
+		}	   
+		
+	}
+
+	private boolean isReady(ResultObject temp) {
+
+		if (collectionPending) {
+			//now ready to send, we have all the data	
+			if (!publishMultiResponse(collector.get(0).getConnectionId(), collector.get(0).getSequenceId() )) {				
+				return false;
+			}
+		}
+		
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private boolean consumeResultObject(final ResultObject t) {
+		boolean ok;
+						
+		///////////////////////////////
+		if (0 == t.getGroupSize()) {	
+			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200,
+				   HTTPContentTypeDefaults.JSON,
+				   w-> {
+					   singleTemplate.render(w, t);
+					   t.setStatus(-1);
+					   inFlight.moveTailForward();//only move forward when it is consumed.
+					   inFlight.publishTailPosition();
+
+				   });					
+		} else {
+			//collect all the objects
+			assert(isValidToAdd(t, collector));
+			collector.add(t);					
+			inFlight.moveTailForward();
+			if (collector.size() == t.getGroupSize()) {
+				//now ready to send, we have all the data						
+				ok =publishMultiResponse(t.getConnectionId(), t.getSequenceId());
+				
+			} else {
+				ok = true;//added to list
+			}	
+			//moved forward so we can read next but write logic will still be blocked by state not -1			 
+			
+		}
+		return ok;
+	}
+
+	private boolean isValidToAdd(ResultObject t, List<ResultObject> collector) {
+		if (collector.isEmpty()) {
+			return true;
+		}
+		if (collector.get(0).getSequenceId() != t.getSequenceId()) {
+			
+			System.out.println("show collection: "+showCollection(collector));
+			System.out.println("new result adding con "+t.getConnectionId()+" seq "+t.getSequenceId());
+			
+		};
+		
+		
+		return true;
+	}
+
+	private boolean publishMultiResponse(long conId, long seqCode) {
+		final boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
+					    				   HTTPContentTypeDefaults.JSON,
+					    				   w-> {
+					    					   multiTemplate.render(w, collector);
+					    					   
+					    					   int c = collector.size();
+					    					   assert(collector.get(0).getGroupSize()==c);
+					    					   while (--c >= 0) {
+					    						   assert(collector.get(0).getGroupSize()==collector.size());
+					    						   assert(collector.get(c).getConnectionId() == conId) : c+" expected conId "+conId+" error: "+showCollection(collector);
+					    						   assert(collector.get(c).getSequenceId() == seqCode) : c+" sequence error: "+showCollection(collector);    						   
+					    						   collector.get(c).setStatus(-1);
+					    						 
+					    					   }
+					    					   collector.clear();					    					   
+					    					   inFlight.publishTailPosition();
+					    				   });
+		collectionPending = !result;
+		return result;
+	}
+
+	private String showCollection(List<ResultObject> collector) {
+		
+		StringBuilder builder = new StringBuilder();
+		builder.append("\n");
+		int i = 0;
+		for(ResultObject ro: collector) {
+			builder.append(++i+" Con:"+ro.getConnectionId()).append(" Id:").append(ro.getId()).append(" Seq:").append(ro.getSequenceId());
+			builder.append("\n");
+		}
+		
+		
+		return builder.toString();
+	}
+	
+	
+}
+

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/DBUpdate.java

@@ -0,0 +1,200 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import com.ociweb.gl.api.GreenRuntime;
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.gl.api.RestMethodListener;
+import com.ociweb.gl.api.TickListener;
+import com.ociweb.json.encode.JSONRenderer;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.PgPool;
+import io.reactiverse.pgclient.PgPoolOptions;
+import io.reactiverse.pgclient.Tuple;
+
+public class DBUpdate implements RestMethodListener, TickListener {
+
+
+	private final transient PgPoolOptions options;
+	private transient PgPool pool;
+	
+	private final HTTPResponseService service;
+	private ObjectPipe<ResultObject> inFlight;	
+	private boolean collectionPending = false;	
+	private final List<ResultObject> collector = new ArrayList<ResultObject>();
+	
+	private static final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
+	private static final JSONRenderer<List<ResultObject>> multiTemplate = new JSONRenderer<List<ResultObject>>()
+	    	  .array((o,i) -> i<o.size()?o:null)
+		          .startObject((o, i) ->  o.get(i))  
+					.integer("id", o -> o.getId() )
+					.integer("randomNumber", o -> o.getResult())
+		          .endObject();
+	
+	public DBUpdate(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int maxResponseCount, int maxResponseSize) {
+		this.options = options;
+		this.options.setMaxSize(6);//bump up the connections since we use the pool nested twice		
+		this.service = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);
+		this.inFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+	}
+	
+	private PgPool pool() {
+		if (null==pool) {
+			pool = PgClient.pool(options);
+		}
+		return pool;
+	}
+	
+	private int randomValue() {
+		return 1+localRandom.nextInt(10000);
+	}	
+	
+	public boolean updateRestRequest(HTTPRequestReader request) {
+		int queries;
+		if (Struct.UPDATES_ROUTE_INT == request.getRouteAssoc() ) {		
+			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
+		} else {
+			queries = 1;
+		}
+		long conId = request.getConnectionId();
+		long seqCode = request.getSequenceCode();
+
+		if (inFlight.hasRoomFor(queries)) {		
+				    	
+				int q = queries;
+				while (--q >= 0) {
+				
+						final ResultObject worldObject = inFlight.headObject();
+						assert(null!=worldObject);
+											
+						worldObject.setConnectionId(conId);
+						worldObject.setSequenceId(seqCode);
+						worldObject.setStatus(-2);//out for work	
+						worldObject.setGroupSize(queries);
+						
+						worldObject.setId(randomValue());
+												
+						pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(worldObject.getId()), r -> {
+								if (r.succeeded()) {
+																		
+									PgIterator resultSet = r.result().iterator();
+							        Tuple row = resultSet.next();			        
+							        
+							        assert(worldObject.getId()==row.getInteger(0));
+							        
+							        //read the existing random value and store it in the world object
+							        worldObject.setResult(row.getInteger(1));
+							        
+							        ///////////////////////////////////
+							        //set the new random value in this object
+							        worldObject.setResult(randomValue());
+							        							       
+							        
+							        pool().preparedQuery("UPDATE world SET randomnumber=$1 WHERE id=$2", 							        		
+							        			Tuple.of(worldObject.getResult(), worldObject.getId()), ar -> {							        	
+										if (ar.succeeded()) {
+											
+								        	worldObject.setStatus(200);							
+								        	
+										} else {	
+											System.out.println("unable to update");
+											if (ar.cause()!=null) {
+												ar.cause().printStackTrace();
+											}
+											
+											worldObject.setStatus(500);
+										}	
+																													
+							        });
+								} else {	
+									System.out.println("unable to query");
+									if (r.cause()!=null) {
+										r.cause().printStackTrace();
+									}
+									
+									worldObject.setStatus(500);
+								}		
+								
+								
+							});	
+									
+						inFlight.moveHeadForward(); //always move to ensure this can be read.
+				
+				}
+				
+			return true;
+		} else {
+			return false;
+		}
+	}
+	
+	
+	@Override
+	public void tickEvent() { 
+		
+		ResultObject temp = inFlight.tailObject();
+		while (isReady(temp)) {			
+			if (consumeResultObject(temp)) {
+				temp = inFlight.tailObject();
+			} else {
+				break;
+			}
+		}	   
+		
+	}
+
+	private boolean isReady(ResultObject temp) {
+
+		if (collectionPending) {
+			//now ready to send, we have all the data	
+			if (!publishMultiResponse(collector.get(0).getConnectionId(), collector.get(0).getSequenceId() )) {
+				return false;
+			}
+		}
+		
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private boolean consumeResultObject(final ResultObject t) {
+		boolean ok;
+		//collect all the objects
+		collector.add(t);
+		inFlight.moveTailForward();//only move forward when it is consumed.
+		if (collector.size() == t.getGroupSize()) {
+			//now ready to send, we have all the data						
+			ok =publishMultiResponse(t.getConnectionId(), t.getSequenceId());
+		} else {
+			ok = true;//added to list
+		}				
+		
+		return ok;
+	}
+
+	private boolean publishMultiResponse(long conId, long seqCode) {
+		boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
+					    				   HTTPContentTypeDefaults.JSON,
+					    				   w-> {
+					    					   multiTemplate.render(w, collector);
+					    					   int c = collector.size();
+					    					   while (--c>=0) {
+					    						   assert(collector.get(c).getConnectionId() == conId);
+					    						   assert(collector.get(c).getSequenceId() == seqCode);					    						   
+					    						   collector.get(c).setStatus(-1);
+					    					   }
+					    					   collector.clear();
+					    					   inFlight.publishTailPosition();
+					    				   });
+		collectionPending = !result;
+		return result;
+	}
+	
+	
+	
+}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/Field.java

@@ -1,5 +1,6 @@
 package com.ociweb.gl.benchmark;
 
 public enum Field {
-	PAYLOAD
+	//identifiers for each field used by routes and/or structures
+	CONNECTION, SEQUENCE, QUERIES; 
 }

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FortuneObject.java

@@ -0,0 +1,26 @@
+package com.ociweb.gl.benchmark;
+
+public class FortuneObject implements Comparable<FortuneObject>{
+
+	private int id;
+	private String fortune;
+	
+	public int getId() {
+		return id;
+	}
+	public void setId(int id) {
+		this.id = id;
+	}
+	public String getFortune() {
+		return fortune;
+	}
+	public void setFortune(String fortune) {
+		this.fortune = fortune;
+	}
+	
+	@Override
+	public int compareTo(FortuneObject o) {
+		return fortune.compareTo(o.fortune);
+	}	
+	
+}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FortuneRest.java

@@ -0,0 +1,183 @@
+package com.ociweb.gl.benchmark;
+
+import com.ociweb.gl.api.GreenRuntime;
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.gl.api.RestMethodListener;
+import com.ociweb.gl.api.TickListener;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+import com.ociweb.pronghorn.util.AppendableBuilder;
+import com.ociweb.pronghorn.util.Appendables;
+import com.ociweb.pronghorn.util.template.StringTemplateBuilder;
+import com.ociweb.pronghorn.util.template.StringTemplateRenderer;
+
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.PgPool;
+import io.reactiverse.pgclient.PgPoolOptions;
+import io.reactiverse.pgclient.Row;
+
+public class FortuneRest implements RestMethodListener, TickListener {
+
+	private static final byte[] ROW_FINISH = "</td></tr>\n".getBytes();
+	private static final byte[] ROW_MIDDLE = "</td><td>".getBytes();
+	private static final byte[] ROW_START = "<tr><td>".getBytes();
+	private final HTTPResponseService service; 
+	
+	private final transient PgPoolOptions options;
+	private transient PgPool pool;
+			
+	//SQL results write to these object, these same objects are used by template
+	private transient ObjectPipe<FortunesObject> inFlight;
+	
+	private static final transient StringTemplateRenderer<FortunesObject> template =		
+			new StringTemplateBuilder<FortunesObject>()
+				   .add("<!DOCTYPE html> <html> <head><title>Fortunes</title></head> <body> <table> <tr><th>id</th><th>message</th></tr>\n")
+			       .add((t,s,i)-> {
+						if (i<s.list().size()) {													
+							t.write(ROW_START);
+							Appendables.appendValue(t, s.list().get(i).getId());
+							t.write(ROW_MIDDLE);							
+							Appendables.appendHTMLEntityEscaped(t, s.list().get(i).getFortune());							
+							t.write(ROW_FINISH);
+							return true;
+						} else {
+							return false;
+						}
+			         })		
+			       .add("</table></body></html>")
+			       .finish();
+
+	
+	public FortuneRest(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int responseCount, int maxResponseSize) {;
+	    
+		this.options = options;	
+		this.service = runtime.newCommandChannel().newHTTPResponseService(responseCount, maxResponseSize);
+		this.inFlight =  new ObjectPipe<FortunesObject>(pipelineBits, FortunesObject.class,	FortunesObject::new);
+		
+	}
+	
+	
+	private PgPool pool() {
+		if (null==pool) {
+			pool = PgClient.pool(options);
+		}
+		return pool;
+	}
+
+	public boolean restRequest(HTTPRequestReader request) {
+	
+		final FortunesObject target = inFlight.headObject(); 
+		if (null!=target) {
+			target.setConnectionId(request.getConnectionId());
+			target.setSequenceId(request.getSequenceCode());
+	
+			target.setStatus(-2);//out for work	
+			target.clear();
+		
+			pool().preparedQuery( "SELECT id, message FROM fortune", r -> {
+				    //NOTE: we want to do as little work here a s possible since
+				    //      we want this thread to get back to work on other calls.
+					if (r.succeeded()) {
+						PgIterator resultSet = r.result().iterator();						
+						while (	resultSet.hasNext() ) {
+					        Row next = resultSet.next();
+							target.addFortune(next.getInteger(0), next.getString(1));						
+						}
+						target.setStatus(200);
+					} else {
+						System.out.println("fail: "+r.cause().getLocalizedMessage());
+						target.setStatus(500);
+					}		
+					
+				});
+			
+			inFlight.moveHeadForward(); //always move to ensure this can be read.  //TODO: remove and combined with above
+			return true;
+		} else {
+			return false;//can not pick up new work now			
+		}		
+	}
+	
+
+	
+	@Override
+	public void tickEvent() { //TODO: remove tickEvent here and replace with  pub sub to take next...
+		
+		FortunesObject temp = inFlight.tailObject();
+		while (isReady(temp)) {			
+			if (consumeResultObject(temp)) {
+				temp = inFlight.tailObject();
+			} else {
+				break;
+			}
+		}		
+	}
+	
+	private boolean isReady(FortunesObject temp) {
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private int htmlPos=0;
+	private final transient AppendableBuilder htmlBuffer = new AppendableBuilder();
+	
+	
+	private boolean consumeResultObject(final FortunesObject t) {
+					
+		if (0 == htmlBuffer.byteLength()) {
+			//capture all the output text
+			t.addFortune(0, "Additional fortune added at request time.");
+			t.sort();
+			template.render(htmlBuffer, t);
+			htmlPos = 0;
+		}
+		
+		
+		int bytesRemaining = htmlBuffer.byteLength() - htmlPos;
+		int roomForWrite = service.maxVarLength();
+		boolean hasContinuation  = bytesRemaining >roomForWrite;
+		
+		//as long as htmlPos does not match the total bytes of the payload keep 
+		//sending out continuation chunks. We do not know how many rows of fortunes
+		//may be in the database.
+		boolean ok;
+		if (0 == htmlPos) {	
+			
+			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200, hasContinuation,
+						   HTTPContentTypeDefaults.HTML, 
+						   w-> {
+							   htmlPos += htmlBuffer.copyTo(w, htmlPos);								   
+							   assert(hasContinuation == (htmlPos!=htmlBuffer.byteLength())) : "internal error";
+							   
+						   });
+		} else {		
+			ok =service.publishHTTPResponseContinuation(t.getConnectionId(), t.getSequenceId(), hasContinuation,  
+							w-> {
+								htmlPos += htmlBuffer.copyTo(w,htmlPos);	
+								assert(hasContinuation == (htmlPos!=htmlBuffer.byteLength())) : "internal error";
+								
+							});
+		}
+		
+		if (ok) {
+			if (htmlPos == htmlBuffer.byteLength()) {
+				
+				t.setStatus(-1);
+				inFlight.moveTailForward();//only move forward when it is consumed.
+				inFlight.publishTailPosition();
+				t.list().clear();
+				htmlBuffer.clear();
+				return true;//do consume this since it is now fully sent
+			} else {
+				assert(htmlPos < htmlBuffer.byteLength()) : "internal error";			
+				return false;//still have more to send later
+			}	
+		} else {
+			return false;
+		}		
+		
+	}
+	
+
+}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FortunesObject.java

@@ -0,0 +1,76 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import io.reactiverse.pgclient.Tuple;
+
+public class FortunesObject {
+
+	private long connectionId; 
+	private long sequenceId;
+	private int status;
+	private List<FortuneObject> list = new ArrayList<FortuneObject>();
+	private List<FortuneObject> recycle = new ArrayList<FortuneObject>();
+
+	public long getConnectionId() {
+		return connectionId;
+	}
+
+
+	public void setConnectionId(long connectionId) {
+		this.connectionId = connectionId;
+	}
+
+
+	public long getSequenceId() {
+		return sequenceId;
+	}
+
+
+	public void setSequenceId(long sequenceId) {
+		this.sequenceId = sequenceId;
+	}
+
+
+	public int getStatus() {
+		return status;
+	}
+
+
+	public void setStatus(int status) {
+		this.status = status;
+	}
+
+
+	public void clear() {
+		recycle.addAll(list);
+		list.clear();
+	}
+
+	public void sort() {
+		Collections.sort(list);
+	}
+
+	public List<FortuneObject> list() {		
+		return list;
+	}
+	
+	public void addFortune(int id, String fortune) {
+		
+		FortuneObject obj; //This eliminates any GC by recycling the old Fortune Objects to be repopulated with new data.
+		if (recycle.isEmpty()) {
+			obj = new FortuneObject(); 
+		} else {
+			obj = recycle.remove(recycle.size()-1);			
+		}
+		
+		obj.setId(id);
+		obj.setFortune(fortune);
+		list.add(obj);
+
+	}
+
+		
+}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FrameworkTest.java

@@ -1,5 +1,7 @@
 package com.ociweb.gl.benchmark;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import com.ociweb.gl.api.GreenApp;
 /**
  * ************************************************************************
@@ -9,7 +11,15 @@ import com.ociweb.gl.api.GreenApp;
  */
 import com.ociweb.gl.api.GreenFramework;
 import com.ociweb.gl.api.GreenRuntime;
-import com.ociweb.pronghorn.stage.scheduling.GraphManager;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.pronghorn.network.ServerSocketReaderStage;
+import com.ociweb.pronghorn.network.ServerSocketWriterStage;
+import com.ociweb.pronghorn.network.http.HTTP1xRouterStage;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgPoolOptions;
+import io.reactiverse.reactivex.pgclient.PgConnection;
 
 public class FrameworkTest implements GreenApp {
 
@@ -22,40 +32,112 @@ public class FrameworkTest implements GreenApp {
     private int queueLengthOfPendingRequests;
     private int telemetryPort;//for monitoring
     private int minMemoryOfInputPipes;
-
+    private int maxResponseSize;
+	private	int maxResponseCount = 1<<8;
+    private int pipelineBits;
+	
+    private final PgPoolOptions options;
+    
+	public static int connectionsPerTrack =   2;
+	public static int connectionPort =        5432;
+	public AtomicBoolean foundDB = new AtomicBoolean(false);
+	public static String connectionHost =     "localhost";
+	public static String connectionDB =       "testdb";
+	public static String connectionUser =     "postgres";
+	public static String connectionPassword = "postgres";
+			    
     public FrameworkTest() {
+    	// use this in commit messages to narrow travis testing to just this project
+    	// [ci fw-only Java/greenlightning]
+    	
     	//this server works best with  -XX:+UseNUMA    	
-    	this(System.getProperty("host","*.*.*.*"), 
-    		 8080, 4, 16*1024, 1<<21, 
-    		 Integer.parseInt(System.getProperty("telemetry.port", "-1")));
-    }
-    
+    	this(System.getProperty("host","0.0.0.0"), 
+    		 8080,    //default port for test 
+    		 10,      //default concurrency, 10 to support 280 channels on 14 core boxes
+    		 8*1024, //default max rest requests allowed to queue in wait
+    		 1<<21,   //default network buffer per input socket connection
+    		 Integer.parseInt(System.getProperty("telemetry.port", "-1")),
+    		 "tfb-database", // jdbc:postgresql://tfb-database:5432/hello_world
+    		 "hello_world",
+    		 "benchmarkdbuser",
+    		 "benchmarkdbpass"
+    		 );    	
+    }   
+        
     public FrameworkTest(String host, int port, 
     		             int concurrentWritesPerChannel, 
     		             int queueLengthOfPendingRequests, 
     		             int minMemoryOfInputPipes,
-    		             int telemetryPort) {
+    		             int telemetryPort,
+    		             String dbHost,
+    		             String dbName,
+    		             String dbUser,
+    		             String dbPass) {
+    	
     	this.bindPort = port;
     	this.host = host;
     	this.concurrentWritesPerChannel = concurrentWritesPerChannel;
     	this.queueLengthOfPendingRequests = queueLengthOfPendingRequests;
     	this.minMemoryOfInputPipes = minMemoryOfInputPipes;
     	this.telemetryPort = telemetryPort;
+    	this.maxResponseSize = 20_000; //for 500 mult db call in JSON format
+    	this.pipelineBits = 19;
+    	
+    	if (!"127.0.0.1".equals(System.getProperty("host",null))) { 
+    		    		
+	    	if (null!=dbHost) {
+	    		this.connectionHost = dbHost;
+	    	}
+	    	if (null!=dbName) {
+	    		this.connectionDB = dbName;
+	    	}
+	    	if (null!=dbUser) {
+	    		this.connectionUser = dbUser;
+	    	}
+	    	if (null!=dbPass) {
+	    		this.connectionPassword = dbPass;
+	    	}    	
+    	}
+    	
+	    	options = new PgPoolOptions()
+	    			.setPort(connectionPort)
+	    			.setPipeliningLimit(1<<pipelineBits)
+	    			.setTcpFastOpen(true)
+	    			.setHost(connectionHost)
+	    			.setDatabase(connectionDB)
+	    			.setUser(connectionUser)
+	    			.setPassword(connectionPassword)
+	    			.setCachePreparedStatements(true)
+	    			.setMaxSize(connectionsPerTrack);	    	
+	    		
+    	try {
+	    	///early check to know if we have a database or not,
+	    	///this helps testing to know which tests should be run on different boxes.
+	    	PgClient.pool(options).getConnection(a->{
+	    		foundDB.set(a.succeeded());
+	    		a.result().close();
+	    	});
+    	} catch (Throwable t) {
+    		t.printStackTrace();
+    		System.out.println("No database in use");
+    	}
+    	
     }
 
+    
 	@Override
     public void declareConfiguration(GreenFramework framework) {
 		
-		GraphManager.showThreadIdOnTelemetry = true;
-			
+		//for 14 cores this is expected to use less than 16G
 		framework.useHTTP1xServer(bindPort, this::parallelBehavior) //standard auto-scale
     			 .setHost(host)
     			 .setConcurrentChannelsPerDecryptUnit(concurrentWritesPerChannel)
     			 .setConcurrentChannelsPerEncryptUnit(concurrentWritesPerChannel)
     			 .setMaxQueueIn(queueLengthOfPendingRequests)
     			 .setMinimumInputPipeMemory(minMemoryOfInputPipes)
+    			 .setMaxResponseSize(maxResponseSize) //big enough for large mult db response
     	         .useInsecureServer(); //turn off TLS
-        
+
 		framework.defineRoute()
 		         .path("/plaintext")
 		         .routeId(Struct.PLAINTEXT_ROUTE);
@@ -63,26 +145,73 @@ public class FrameworkTest implements GreenApp {
 		framework.defineRoute()
 		        .path("/json")
 		        .routeId(Struct.JSON_ROUTE);
-	
+		
+		framework.defineRoute()
+		        .path("/db")
+		        .routeId(Struct.DB_SINGLE_ROUTE);
+			
+		framework.defineRoute()
+		        .path("/queries?queries=#{queries}")
+		        .path("/queries")
+		        .refineInteger("queries", Field.QUERIES, 1)
+		        .routeId(Struct.DB_MULTI_ROUTE_INT);
+		
+		framework.defineRoute()
+		        .path("/queries?queries=${queries}")
+			    .routeId(Struct.DB_MULTI_ROUTE_TEXT);
+		
+		framework.defineRoute()
+		        .path("/updates?queries=#{queries}")
+		        .path("/updates")
+		        .refineInteger("queries", Field.QUERIES, 1)
+		        .routeId(Struct.UPDATES_ROUTE_INT);
+
+		framework.defineRoute()
+		        .path("/updates?queries=${queries}") 
+		        .routeId(Struct.UPDATES_ROUTE_TEXT);
+		
+		framework.defineRoute()
+		        .path("/fortunes")
+		        .routeId(Struct.FORTUNES_ROUTE);		
+		
 		if (telemetryPort>0) {
 			framework.enableTelemetry(host,telemetryPort);
 		}
 		
+		
+		ServerSocketWriterStage.lowLatency = false; //turn on high volume mode, less concerned about low latency. 
+	
     }
 
 	public void parallelBehavior(GreenRuntime runtime) {
 
-		runtime.addRestListener("PlainResponder",new PlainBehaviorInstance(runtime))
-		       .includeRoutes(Struct.PLAINTEXT_ROUTE);
+		SimpleRest restTest = new SimpleRest(runtime);
+		
+		runtime.registerListener("Simple", restTest)
+		       .includeRoutes(Struct.PLAINTEXT_ROUTE, restTest::plainRestRequest)
+		       .includeRoutes(Struct.JSON_ROUTE, restTest::jsonRestRequest);
+		 
 
-		runtime.addRestListener("JSONResponder",new JSONBehaviorInstance(runtime))
-		       .includeRoutes(Struct.JSON_ROUTE);
+		DBRest dbRestInstance = new DBRest(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
+		runtime.registerListener("DBRest", dbRestInstance)
+				.includeRoutes(Struct.DB_SINGLE_ROUTE, dbRestInstance::singleRestRequest)
+				.includeRoutes(Struct.DB_MULTI_ROUTE_TEXT, dbRestInstance::multiRestRequest)		
+		        .includeRoutes(Struct.DB_MULTI_ROUTE_INT, dbRestInstance::multiRestRequest);
 
+		
+		DBUpdate dbUpdateInstance = new DBUpdate(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
+		runtime.registerListener("DBUpdate", dbUpdateInstance)
+		        .includeRoutes(Struct.UPDATES_ROUTE_TEXT, dbUpdateInstance::updateRestRequest)
+		        .includeRoutes(Struct.UPDATES_ROUTE_INT,  dbUpdateInstance::updateRestRequest);
+		
+		FortuneRest fortuneInstance = new FortuneRest(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
+		runtime.registerListener("Fortune", fortuneInstance)
+		        .includeRoutes(Struct.FORTUNES_ROUTE, fortuneInstance::restRequest);	
+		
 	}
 	 
     @Override
-    public void declareBehavior(GreenRuntime runtime) {   
-    	
+    public void declareBehavior(GreenRuntime runtime) { 
     }
   
 }

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/GreenLightning.java

@@ -5,7 +5,15 @@ import com.ociweb.gl.api.GreenRuntime;
 public class GreenLightning {
 
 	public static void main(String[] args) {
+		//ServerSocketWriterStage.showWrites=true;
+	
+		//  [{"id":1480,"randomNumber":1784090351}
+		//  ,{"id":6038,"randomNumber":-447995528}
+		//  ,{"id":2669,"randomNumber":493033553}
+		//  ,{"id":2487,"randomNumber":511963400}]
+		
 		GreenRuntime.run(new FrameworkTest(),args);
+	
 	}
 	
 }

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/JSONBehaviorInstance.java

@@ -1,39 +0,0 @@
-package com.ociweb.gl.benchmark;
-
-import com.ociweb.gl.api.GreenRuntime;
-import com.ociweb.gl.api.HTTPRequestReader;
-import com.ociweb.gl.api.HTTPResponseService;
-import com.ociweb.gl.api.RestListener;
-import com.ociweb.json.encode.JSONRenderer;
-import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
-
-public class JSONBehaviorInstance implements RestListener {
-
-
-	private final HTTPResponseService responseService;
-
-	
-	public JSONBehaviorInstance(GreenRuntime runtime) {
-		responseService = runtime.newCommandChannel().newHTTPResponseService(1<<14, 1<<8);		
-	}
-
-
-	@Override
-	public boolean restRequest(HTTPRequestReader request) {
-	
-		//NOTE: this is only done here for the framework test
-		//      in a normal production deployment this JSONRender will only
-		//      be created once and held as a member.
-		JSONRenderer<HTTPRequestReader> renderJSON = new JSONRenderer<HTTPRequestReader>()
-				.startObject()
-					.string("message", (o,t) -> t.write(FrameworkTest.payload) )
-				.endObject();
-				
-		return responseService.publishHTTPResponse(request, 
-				                            HTTPContentTypeDefaults.JSON,
-				                            w -> renderJSON.render(w,request)
-				                            );
-		
-	}
-
-}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/PlainBehaviorInstance.java

@@ -1,37 +0,0 @@
-package com.ociweb.gl.benchmark;
-
-import com.ociweb.gl.api.GreenRuntime;
-import com.ociweb.gl.api.HTTPRequestReader;
-import com.ociweb.gl.api.HTTPResponseService;
-import com.ociweb.gl.api.RestListener;
-import com.ociweb.gl.api.Writable;
-import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
-import com.ociweb.pronghorn.pipe.ChannelWriter;
-
-public class PlainBehaviorInstance implements RestListener {
-
-	private final HTTPResponseService plainResponseService;
-	
-	//a lambda could be used if you like
-	private Writable writePayload = new Writable() {
-		@Override
-		public void write(ChannelWriter writer) {
-			writer.write(FrameworkTest.payload);
-		}		
-	};
-	
-	public PlainBehaviorInstance(GreenRuntime runtime) {
-		plainResponseService = runtime.newCommandChannel().newHTTPResponseService(1<<14, 1<<8); 
-	}
-
-	@Override
-	public boolean restRequest(HTTPRequestReader request) {
-		
-		return plainResponseService.publishHTTPResponse(request, 	
-					HTTPContentTypeDefaults.PLAIN,
-					writePayload
-				);
-		
-	}
-
-}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/ResultObject.java

@@ -2,10 +2,83 @@ package com.ociweb.gl.benchmark;
 
 public class ResultObject {
 
-	public final byte[] payload;
+	private long connectionId; 
+	private long sequenceId;
+	private int status;
+	private int id;
+	private int result;
+	private int groupSize;
 	
-	public ResultObject(byte[] payload) {
-		this.payload=payload;
+	public int getGroupSize() {
+		return groupSize;
+	}
+
+
+	public void setGroupSize(int groupSize) {
+		this.groupSize = groupSize;
+	}
+
+
+	public long getConnectionId() {
+		return connectionId;
+	}
+
+
+	public void setConnectionId(long connectionId) {
+		this.connectionId = connectionId;
+	}
+
+
+	public long getSequenceId() {
+		return sequenceId;
+	}
+
+
+	public void setSequenceId(long sequenceId) {
+		this.sequenceId = sequenceId;
+	}
+
+
+	public int getStatus() {
+		return status;
+	}
+
+
+	public void setStatus(int status) {
+		this.status = status;
+	}
+
+
+	public int getId() {
+		return id;
+	}
+
+
+	public void setId(int id) {
+		this.id = id;
+	}
+
+
+	public int getResult() {
+		return result;
+	}
+
+
+	public void setResult(int result) {
+		this.result = result;
+	}
+
+	
+	public ResultObject(int id, int result) {
+		this.id = id;
+		this.result = result;
+		this.groupSize = 0;
+		this.status = -1;
+	}
+	
+	public ResultObject() {
+		this.status = -1;
+		this.groupSize = 0;
 	}
 	
 }

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/SimpleRest.java

@@ -0,0 +1,54 @@
+package com.ociweb.gl.benchmark;
+
+import com.ociweb.gl.api.GreenRuntime;
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.gl.api.RestMethodListener;
+import com.ociweb.gl.api.Writable;
+import com.ociweb.json.encode.JSONRenderer;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ChannelWriter;
+
+public class SimpleRest implements RestMethodListener {
+
+	private static final int QUEUE_LENGTH = 1<<15;
+	private static final int MAX_MESSAGE_SIZE = 1<<9;
+
+	private final HTTPResponseService responseService;
+	
+	public SimpleRest(GreenRuntime runtime) {
+		responseService = runtime.newCommandChannel().newHTTPResponseService(QUEUE_LENGTH, MAX_MESSAGE_SIZE);		
+	}
+	
+	public boolean jsonRestRequest(HTTPRequestReader request) {
+	
+		//this check is to postpone the work if the network has become saturated
+		if (responseService.hasRoomFor(1)) {
+			//NOTE: this is only done here for the framework test
+			//      in a normal production deployment this JSONRender will only
+			//      be created once and held as a member.
+			JSONRenderer<HTTPRequestReader> renderJSON = new JSONRenderer<HTTPRequestReader>()
+					.startObject()
+					.string("message", (o,t) -> t.write(FrameworkTest.payload) )
+					.endObject();
+					
+			return responseService.publishHTTPResponse(request, 
+					                            HTTPContentTypeDefaults.JSON,
+					                            w -> renderJSON.render(w,request)
+					                            );
+		} else {
+			return false;
+		}
+	}
+	
+	
+	public boolean plainRestRequest(HTTPRequestReader request) {
+	
+		return responseService.publishHTTPResponse(request, 	
+					HTTPContentTypeDefaults.PLAIN,
+					w -> w.write(FrameworkTest.payload)
+				);
+		
+	}
+
+}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/Struct.java

@@ -1,6 +1,9 @@
 package com.ociweb.gl.benchmark;
 
 public enum Struct {
-	PLAINTEXT_ROUTE, JSON_ROUTE
-
+	PLAINTEXT_ROUTE, JSON_ROUTE, 
+	DB_SINGLE_ROUTE, 
+	DB_MULTI_ROUTE_INT, DB_MULTI_ROUTE_TEXT, 
+	FORTUNES_ROUTE,
+	UPDATES_ROUTE_INT, UPDATES_ROUTE_TEXT;
 }