@Test public void readFromTest() throws IOException, InterruptedException, TimeoutException, ExecutionException { final CountDownLatch readThreadIsReadyLatch = new CountDownLatch(1); final Exchanger<Example> exampleExchanger = new Exchanger<Example>(); final PipedInputStream pipedInputStream = new PipedInputStream(); final PipedOutputStream pipedOutputStream = new PipedOutputStream(pipedInputStream); ExecutorService executorService = Executors.newCachedThreadPool(); Future<Integer> readingThreadFuture = executorService.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { readThreadIsReadyLatch.countDown(); Iterable<Example> readStructuredExamples = toTest.readFrom(ExamplesIterable.class, null, null, null, null, pipedInputStream); int numExamplesRead = 0; LOGGER.trace("Starting to read examples..."); for (Example readExample : readStructuredExamples) { exampleExchanger.exchange(readExample); numExamplesRead++; } return Integer.valueOf(numExamplesRead); } }); readThreadIsReadyLatch.await(); LOGGER.trace("Writing examples..."); StructuredExample lastComputedExample = null; OutputStreamWriter outputStreamWriter = new OutputStreamWriter(pipedOutputStream, Charsets.UTF_8); JsonWriter jsonWriter = new JsonWriter(outputStreamWriter); jsonWriter.beginArray(); Iterable<StructuredExample> structuredExamplesIterable = TestUtils.getStructuredExamplesFromNerTrain(); for (StructuredExample example : structuredExamplesIterable) { if (lastComputedExample != null) { Assert.assertEquals(lastComputedExample.getVWStringRepresentation(), exampleExchanger.exchange(null, 2000, TimeUnit.MILLISECONDS).getVWStringRepresentation()); } if (example != StructuredExample.EMPTY_EXAMPLE) JsonTestUtils.writeExample(jsonWriter, example); else { jsonWriter.beginObject(); jsonWriter.endObject(); } jsonWriter.flush(); lastComputedExample = example; } jsonWriter.endArray(); jsonWriter.flush(); jsonWriter.close(); LOGGER.trace("Verifying final example..."); Assert.assertEquals(lastComputedExample.getVWStringRepresentation(), exampleExchanger.exchange(null, 2000, TimeUnit.MILLISECONDS).getVWStringRepresentation()); Assert.assertEquals(272274, readingThreadFuture.get().intValue()); }
@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 StructuredJsonExamplesProvider(-1, -1).getExamplesFromStream(entityStream)); }
StructuredJsonExamplesMessageBodyReader 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 StructuredJsonExamplesProvider(-1, -1).getExamplesFromStream(entityStream)); } }
StructuredJsonExamplesMessageBodyReader 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 StructuredJsonExamplesProvider(-1, -1).getExamplesFromStream(entityStream)); }  }
StructuredJsonExamplesMessageBodyReader 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 StructuredJsonExamplesProvider(-1, -1).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); }
StructuredJsonExamplesMessageBodyReader 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 StructuredJsonExamplesProvider(-1, -1).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);  }