@Test public void test() throws IOException, InterruptedException, ExecutionException, TimeoutException { GZIPInputStream gzipInputStream = new GZIPInputStream(this.getClass().getClassLoader().getResourceAsStream("ner.train.gz")); final BufferedReader testReader = new BufferedReader(new InputStreamReader(gzipInputStream, Charsets.UTF_8)); final PipedOutputStream pipedOutputStream = new PipedOutputStream(); PipedInputStream pipedInputStream = new PipedInputStream(pipedOutputStream); GsonJsonExamplesProvider toTest = new GsonJsonExamplesProvider(); final CountDownLatch startWritingExamplesLatch = new CountDownLatch(1); final Exchanger<String> exampleExchanger = new Exchanger<String>(); final JsonWriter jsonWriter = new JsonWriter(new OutputStreamWriter(pipedOutputStream)); Future<Void> successFuture = Executors.newCachedThreadPool().submit(new Callable<Void>() { @Override public Void call() throws Exception { startWritingExamplesLatch.await(); String readExample = null; jsonWriter.beginArray(); while ((readExample = testReader.readLine()) != null) { jsonWriter.beginObject(); jsonWriter.name("example"); jsonWriter.value(readExample); jsonWriter.endObject(); jsonWriter.flush(); pipedOutputStream.flush(); exampleExchanger.exchange(readExample, 1, TimeUnit.SECONDS); } jsonWriter.endArray(); jsonWriter.flush(); jsonWriter.close(); pipedOutputStream.flush(); pipedOutputStream.close(); return null; } }); Iterator<Example> examplesIterable = toTest.getExamplesFromStream(pipedInputStream); startWritingExamplesLatch.countDown(); long numExamplesProcessed = 0; while (examplesIterable.hasNext()) { Example theExample = examplesIterable.next(); Assert.assertEquals(exampleExchanger.exchange("", 1, TimeUnit.SECONDS), theExample.getVWStringRepresentation()); if (numExamplesProcessed % 20000 == 0) LOGGER.debug("Read example: {}", theExample.getVWStringRepresentation()); numExamplesProcessed++; } successFuture.get(); Assert.assertEquals(272274, numExamplesProcessed); }
@Override public Iterator<Example> getExamplesFromStream(InputStream inputStream) throws ExampleReadException { final JsonReader jsonReader = new JsonReader(new InputStreamReader(inputStream)); AbstractIterator<Example> theIterator = new AbstractIterator<Example>() { private boolean readStartOfArray = false; @Override public Example computeNext() { try { if (!readStartOfArray) { jsonReader.beginArray(); readStartOfArray = true; } if (jsonReader.hasNext()) { return readIndividualJsonExample(jsonReader); } else { jsonReader.endArray(); return endOfData(); } } catch (Exception e) { throw new ExampleReadException(e); } } }; return theIterator; }
GsonJsonExamplesProvider implements JsonExamplesProvider { @Override public Iterator<Example> getExamplesFromStream(InputStream inputStream) throws ExampleReadException { final JsonReader jsonReader = new JsonReader(new InputStreamReader(inputStream)); AbstractIterator<Example> theIterator = new AbstractIterator<Example>() { private boolean readStartOfArray = false; @Override public Example computeNext() { try { if (!readStartOfArray) { jsonReader.beginArray(); readStartOfArray = true; } if (jsonReader.hasNext()) { return readIndividualJsonExample(jsonReader); } else { jsonReader.endArray(); return endOfData(); } } catch (Exception e) { throw new ExampleReadException(e); } } }; return theIterator; } }
GsonJsonExamplesProvider implements JsonExamplesProvider { @Override public Iterator<Example> getExamplesFromStream(InputStream inputStream) throws ExampleReadException { final JsonReader jsonReader = new JsonReader(new InputStreamReader(inputStream)); AbstractIterator<Example> theIterator = new AbstractIterator<Example>() { private boolean readStartOfArray = false; @Override public Example computeNext() { try { if (!readStartOfArray) { jsonReader.beginArray(); readStartOfArray = true; } if (jsonReader.hasNext()) { return readIndividualJsonExample(jsonReader); } else { jsonReader.endArray(); return endOfData(); } } catch (Exception e) { throw new ExampleReadException(e); } } }; return theIterator; }  }
GsonJsonExamplesProvider implements JsonExamplesProvider { @Override public Iterator<Example> getExamplesFromStream(InputStream inputStream) throws ExampleReadException { final JsonReader jsonReader = new JsonReader(new InputStreamReader(inputStream)); AbstractIterator<Example> theIterator = new AbstractIterator<Example>() { private boolean readStartOfArray = false; @Override public Example computeNext() { try { if (!readStartOfArray) { jsonReader.beginArray(); readStartOfArray = true; } if (jsonReader.hasNext()) { return readIndividualJsonExample(jsonReader); } else { jsonReader.endArray(); return endOfData(); } } catch (Exception e) { throw new ExampleReadException(e); } } }; return theIterator; }  @Override Iterator<Example> getExamplesFromStream(InputStream inputStream); }
GsonJsonExamplesProvider implements JsonExamplesProvider { @Override public Iterator<Example> getExamplesFromStream(InputStream inputStream) throws ExampleReadException { final JsonReader jsonReader = new JsonReader(new InputStreamReader(inputStream)); AbstractIterator<Example> theIterator = new AbstractIterator<Example>() { private boolean readStartOfArray = false; @Override public Example computeNext() { try { if (!readStartOfArray) { jsonReader.beginArray(); readStartOfArray = true; } if (jsonReader.hasNext()) { return readIndividualJsonExample(jsonReader); } else { jsonReader.endArray(); return endOfData(); } } catch (Exception e) { throw new ExampleReadException(e); } } }; return theIterator; }  @Override Iterator<Example> getExamplesFromStream(InputStream inputStream);  }