svn commit: r1670176 - in /lucene/dev/trunk/solr/solrj/src: java/org/apache/solr/client/solrj/io/ParallelStream.java java/org/apache/solr/client/solrj/io/Tuple.java test/org/apache/solr/client/solrj/io/StreamingTest.java

[jb...@apache.org]

Author: jbernste
Date: Mon Mar 30 18:56:47 2015
New Revision: 1670176

URL: http://svn.apache.org/r1670176
Log:
SOLR-7082: Syntactic sugar for metric gathering

Modified:
    lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java
    lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java
    lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java

Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java?rev=1670176&r1=1670175&r2=1670176&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java Mon Mar 30 18:56:47 2015
@@ -83,7 +83,9 @@ public class ParallelStream extends Clou
     if(tuple.EOF) {
       Map m = new HashMap();
       m.put("EOF", true);
-      return new Tuple(m);
+      Tuple t = new Tuple(m);
+      t.setMetrics(this.eofTuples);
+      return t;
     }
 
     return tuple;

Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java?rev=1670176&r1=1670175&r2=1670176&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java Mon Mar 30 18:56:47 2015
@@ -81,10 +81,6 @@ public class Tuple implements Cloneable
     return (List<Double>)this.fields.get(key);
   }
 
-  public Iterator<Map.Entry> getFields() {
-    return fields.entrySet().iterator();
-  }
-
   public Map getMap() {
     return this.fields;
   }
@@ -95,9 +91,15 @@ public class Tuple implements Cloneable
 
   public void setMaps(List<Map> maps) {
     this.fields.put("_MAPS_", maps);
+  }
 
+  public Map<String,Tuple> getMetrics() {
+    return (Map<String,Tuple>)this.fields.get("_METRICS_");
   }
 
+  public void setMetrics(Map<String, Tuple> metrics) {
+    this.fields.put("_METRICS_", metrics);
+  }
 
   public Tuple clone() {
     HashMap m = new HashMap();

Modified: lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java?rev=1670176&r1=1670175&r2=1670176&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java (original)
+++ lucene/dev/trunk/solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java Mon Mar 30 18:56:47 2015
@@ -163,7 +163,7 @@ public class StreamingTest extends Abstr
 
     String zkHost = zkServer.getZkAddress();
 
-    Map paramsA = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_s asc,a_f asc", "partitionKeys", "none");
+    Map paramsA = mapParams("q", "*:*", "fl", "id,a_s,a_i,a_f", "sort", "a_s asc,a_f asc", "partitionKeys", "none");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
     ParallelStream pstream = new ParallelStream(zkHost, "collection1", stream, 2, new AscFieldComp("a_s"));
 
@@ -723,8 +723,17 @@ public class StreamingTest extends Abstr
   protected List<Tuple> getTuples(TupleStream tupleStream) throws IOException {
     tupleStream.open();
     List<Tuple> tuples = new ArrayList();
-    for(Tuple t = tupleStream.read(); !t.EOF; t = tupleStream.read()) {
-      tuples.add(t);
+    for(;;) {
+      Tuple t = tupleStream.read();
+      if(t.EOF) {
+        if(tupleStream instanceof ParallelStream) {
+          ParallelStream p = (ParallelStream) tupleStream;
+          assert(t.getMetrics() == p.getEofTuples()); // Make sure the EOF tuples are properly set on the final EOF tuple
+        }
+        break;
+      } else {
+        tuples.add(t);
+      }
     }
     tupleStream.close();
     return tuples;