Merge pull request #46 from aosagie/feature/avro-file-output

Add Avro file output and test

Author: cowtowncoder

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/AvroGenerator.java

@@ -49,7 +49,13 @@ public enum Feature
          *
          * @since 2.7
          */
-        AVRO_BUFFERING(true)
+        AVRO_BUFFERING(true),
+
+        /**
+         * Feature that tells Avro to write data in file format (i.e. including the schema with the data)
+         * rather than the RPC format
+         */
+        AVRO_FILE_OUTPUT(false)
         ;
 
         protected final boolean _defaultState;
@@ -600,8 +606,12 @@ protected void _complete() throws IOException
         // do not want to hide the original problem...
         // First one sanity check, for a (relatively?) common case
         if (_rootContext != null) {
-            _rootContext.complete();
-            _encoder.flush();
+            if (isEnabled(Feature.AVRO_FILE_OUTPUT)) {
+                _rootContext.complete(_output);
+            } else {
+                _rootContext.complete();
+                _encoder.flush();
+            }
         }
     }
 }

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/ser/AvroWriteContext.java

@@ -1,6 +1,7 @@
 package com.fasterxml.jackson.dataformat.avro.ser;
 
 import java.io.IOException;
+import java.io.OutputStream;
 
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Type;
@@ -97,6 +98,10 @@ public void complete() throws IOException {
         throw new IllegalStateException("Can not be called on "+getClass().getName());
     }
 
+    public void complete(OutputStream outputStream) throws IOException {
+        throw new IllegalStateException("Can not be called on "+getClass().getName());
+    }
+
     @Deprecated // remove from 2.9
     public void complete(BinaryEncoder encoder) throws IOException { complete(); }
 

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/ser/RootContext.java

@@ -1,13 +1,16 @@
 package com.fasterxml.jackson.dataformat.avro.ser;
 
 import java.io.IOException;
+import java.io.OutputStream;
 
 import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.generic.*;
 import org.apache.avro.io.BinaryEncoder;
 
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.dataformat.avro.AvroGenerator;
+import org.apache.avro.io.DatumWriter;
 
 class RootContext
     extends AvroWriteContext
@@ -105,6 +108,16 @@ public void complete() throws IOException {
         _rootValue = null;
     }
 
+    @Override
+    public void complete(OutputStream outputStream) throws IOException {
+        DatumWriter<Object> datumWriter = new NonBSGenericDatumWriter<>(_schema);
+        DataFileWriter<Object> dataFileWriter = new DataFileWriter<>(datumWriter);
+
+        dataFileWriter.create(_schema, outputStream);
+        dataFileWriter.append(_rootValue);
+        dataFileWriter.close();
+    }
+
     @Override
     public void appendDesc(StringBuilder sb) {
         sb.append("/");

avro/src/test/java/com/fasterxml/jackson/dataformat/avro/SimpleGenerationTest.java

@@ -3,6 +3,11 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
+import org.apache.avro.file.DataFileReader;
+import org.apache.avro.file.SeekableByteArrayInput;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.DatumReader;
 import org.junit.Assert;
 
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
@@ -160,4 +165,31 @@ public void testIgnoringOfUnknownObject() throws Exception
         BinaryAndNumber output = mapper.reader(SCHEMA_WITH_BINARY_JSON).forType(BinaryAndNumber.class).readValue(bytes);
         assertEquals("Bob", output.name);
     }
+
+    public void testFileOutput() throws Exception
+    {
+        Employee empl = new Employee();
+        empl.name = "Bobbee";
+        empl.age = 39;
+        empl.emails = new String[] { "[email protected]", "[email protected]" };
+        empl.boss = null;
+
+        AvroFactory af = new AvroFactory();
+        ObjectMapper mapper = new ObjectMapper(af);
+
+        af.enable(AvroGenerator.Feature.AVRO_FILE_OUTPUT);
+
+        AvroSchema schema = getEmployeeSchema();
+        byte[] bytes = mapper.writer(schema).writeValueAsBytes(empl);
+
+        assertNotNull(bytes);
+        assertEquals(301, bytes.length);
+
+        DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema.getAvroSchema());
+        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(new SeekableByteArrayInput(bytes), datumReader);
+        GenericRecord output = dataFileReader.next();
+
+        assertNotNull(output);
+        assertEquals(output.get("name").toString(), empl.name);
+    }
 }