KeyValuePairToDocument extends StandardStep<KeyValuePair, Document> { public void process(KeyValuePair object) throws IOException { ByteArrayInputStream stream = new ByteArrayInputStream(object.value); Document document; try { if (compressed) { ObjectInputStream input = new ObjectInputStream(new SnappyInputStream(stream)); document = (Document) input.readObject(); } else { ObjectInputStream input = new ObjectInputStream(stream); document = (Document) input.readObject(); } } catch (ClassNotFoundException ex) { throw new RuntimeException("Unable to extract document from KeyValuePair." + ex.toString()); } processor.process(document); } KeyValuePairToDocument();  KeyValuePairToDocument(TupleFlowParameters parameters); void process(KeyValuePair object);  }
@Test public void testProcess() throws Exception { KeyValuePairToDocument doc = new KeyValuePairToDocument(); DocumentProcessor processor = new DocumentProcessor(); ByteArrayOutputStream array = new ByteArrayOutputStream(); ObjectOutputStream output = new ObjectOutputStream(array); Document document = new Document(); document.identifier = 10; document.text = "This is text"; document.metadata.put("hi", "there"); document.name = "DOC1"; output.writeObject(document); byte[] key = Utility.fromLong(document.identifier); byte[] value = array.toByteArray(); KeyValuePair pair = new KeyValuePair(key, value); doc.setProcessor(processor); doc.process(pair); assertEquals(processor.document.name, document.name); assertEquals(processor.document.text, document.text); assertEquals(processor.document.metadata.size(), document.metadata.size()); assertEquals(processor.document.metadata.get("hi"), "there"); }