@Test public void readFromTest() throws IOException, InterruptedException, TimeoutException, ExecutionException { 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); 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; } }); ExamplesIterable examplesIterable = toTest.readFrom(ExamplesIterable.class, null, null, null, null, pipedInputStream); startWritingExamplesLatch.countDown(); long numExamplesProcessed = 0; for (Example example : examplesIterable) { Assert.assertEquals(exampleExchanger.exchange("", 1, TimeUnit.SECONDS), example.getVWStringRepresentation()); if (numExamplesProcessed % 20000 == 0) LOGGER.debug("Read example: {}", example.getVWStringRepresentation()); numExamplesProcessed++; } successFuture.get(); Assert.assertEquals(272274, numExamplesProcessed); }
@Override public ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException { if (LOGGER.isDebugEnabled()) if (httpHeaders != null && httpHeaders.size() > 0) { LOGGER.debug("Rec'd HTTP headers: "); for (Entry<String, List<String>> entry : httpHeaders.entrySet()) { LOGGER.debug("{}:{}", entry.getKey(), StringUtils.join(entry.getValue(), ',')); } } return new ExamplesIterableImpl(Integer.MAX_VALUE, null, new GsonJsonExamplesProvider().getExamplesFromStream(entityStream)); }
SimpleJsonExamplesMessageBodyReader implements MessageBodyReader<ExamplesIterable> { @Override public ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException { if (LOGGER.isDebugEnabled()) if (httpHeaders != null && httpHeaders.size() > 0) { LOGGER.debug("Rec'd HTTP headers: "); for (Entry<String, List<String>> entry : httpHeaders.entrySet()) { LOGGER.debug("{}:{}", entry.getKey(), StringUtils.join(entry.getValue(), ',')); } } return new ExamplesIterableImpl(Integer.MAX_VALUE, null, new GsonJsonExamplesProvider().getExamplesFromStream(entityStream)); } }
SimpleJsonExamplesMessageBodyReader implements MessageBodyReader<ExamplesIterable> { @Override public ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException { if (LOGGER.isDebugEnabled()) if (httpHeaders != null && httpHeaders.size() > 0) { LOGGER.debug("Rec'd HTTP headers: "); for (Entry<String, List<String>> entry : httpHeaders.entrySet()) { LOGGER.debug("{}:{}", entry.getKey(), StringUtils.join(entry.getValue(), ',')); } } return new ExamplesIterableImpl(Integer.MAX_VALUE, null, new GsonJsonExamplesProvider().getExamplesFromStream(entityStream)); }  }
SimpleJsonExamplesMessageBodyReader implements MessageBodyReader<ExamplesIterable> { @Override public ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException { if (LOGGER.isDebugEnabled()) if (httpHeaders != null && httpHeaders.size() > 0) { LOGGER.debug("Rec'd HTTP headers: "); for (Entry<String, List<String>> entry : httpHeaders.entrySet()) { LOGGER.debug("{}:{}", entry.getKey(), StringUtils.join(entry.getValue(), ',')); } } return new ExamplesIterableImpl(Integer.MAX_VALUE, null, new GsonJsonExamplesProvider().getExamplesFromStream(entityStream)); }  @Override boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType); @Override ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream); }
SimpleJsonExamplesMessageBodyReader implements MessageBodyReader<ExamplesIterable> { @Override public ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException { if (LOGGER.isDebugEnabled()) if (httpHeaders != null && httpHeaders.size() > 0) { LOGGER.debug("Rec'd HTTP headers: "); for (Entry<String, List<String>> entry : httpHeaders.entrySet()) { LOGGER.debug("{}:{}", entry.getKey(), StringUtils.join(entry.getValue(), ',')); } } return new ExamplesIterableImpl(Integer.MAX_VALUE, null, new GsonJsonExamplesProvider().getExamplesFromStream(entityStream)); }  @Override boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType); @Override ExamplesIterable readFrom(Class<ExamplesIterable> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream);  }