	@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(); //the reading thread will read from this stream
		final PipedOutputStream pipedOutputStream = new PipedOutputStream(pipedInputStream); //the submission thread will write to this stream

		ExecutorService executorService = Executors.newCachedThreadPool();

		//-------
		//this is the thread that will read the structured examples and compare
		//them to what was submitted by the submitting thread.
		Future<Integer> readingThreadFuture = executorService.submit(new Callable<Integer>() {

			@Override
			public Integer call() throws Exception {

				readThreadIsReadyLatch.countDown(); //signal to the writing thread that this thread is ready.

				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) {
					//LOGGER.trace("Read example: {}", readExample.getVWStringRepresentation());

					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;

		}//end for

		jsonWriter.endArray();

		jsonWriter.flush();

		jsonWriter.close();

		LOGGER.trace("Verifying final example...");

		//don't forget to verify the very last example!
		Assert.assertEquals(lastComputedExample.getVWStringRepresentation(), exampleExchanger.exchange(null, 2000, TimeUnit.MILLISECONDS).getVWStringRepresentation());

		Assert.assertEquals(272274, readingThreadFuture.get().intValue()); //assert that no exceptions where thrown.

	}