DocumentToKeyValuePair extends StandardStep<Document, KeyValuePair> implements KeyValuePair.Source { @Override public void process(Document document) throws IOException { ByteArrayOutputStream array = new ByteArrayOutputStream(); ObjectOutputStream output; if (compressed) { output = new ObjectOutputStream(new SnappyOutputStream(array)); } else { output = new ObjectOutputStream(array); } output.writeObject(document); output.close(); byte[] key = Utility.fromLong(document.identifier); byte[] value = array.toByteArray(); KeyValuePair pair = new KeyValuePair(key, value); processor.process(pair); } DocumentToKeyValuePair();  DocumentToKeyValuePair(TupleFlowParameters parameters); @Override void process(Document document);  }
@Test public void testProcess() throws Exception { DocumentToKeyValuePair dkvp = new DocumentToKeyValuePair(); KeyValuePairProcessor kvpProcessor = new KeyValuePairProcessor(); dkvp.setProcessor(kvpProcessor); Document document = new Document(); document.identifier = 1; document.text = "This is text."; document.name = "DOC2"; document.metadata.put("this", "that"); dkvp.process(document); KeyValuePair pair = kvpProcessor.pair; assertEquals(Utility.toLong(pair.key), 1); ByteArrayInputStream stream = new ByteArrayInputStream(pair.value); ObjectInputStream input = new ObjectInputStream(stream); Document result = (Document) input.readObject(); assertEquals(result.identifier, document.identifier); assertEquals(result.text, document.text); assertEquals(result.name, document.name); assertEquals(result.metadata.size(), document.metadata.size()); assertEquals(result.metadata.get("this"), "that"); }