public class RawAvroHiveSerializer extends AbstractAvroEventSerializer<AvroData> {
    private final static Logger logger = LoggerFactory.getLogger(RawAvroHiveSerializer.class);
    private final OutputStream out;
    private final Schema schema;

    private DatumWriter<AvroData> writer = null;
    private DataFileWriter<AvroData> dataFileWriter = null;

    @Override
    protected OutputStream getOutputStream() {
        return out;
    }

    @Override
    protected Schema getSchema() {
        return schema;
    }

    protected RawAvroHiveSerializer(OutputStream out) {
            this.schema =  AvroData.getClassSchema();
            this.out = out;
    }


    public void configure(Context context) {
        int syncIntervalBytes = context.getInteger("syncIntervalBytes", Integer.valueOf(2048000)).intValue();
        String compressionCodec = context.getString("compressionCodec", "null");
        this.writer = new ReflectDatumWriter(this.getSchema());
        this.dataFileWriter = new DataFileWriter(this.writer);
        this.dataFileWriter.setSyncInterval(syncIntervalBytes);

        System.out.println("The datafileWriter being initialized "+ dataFileWriter.toString());

        try {
            CodecFactory e = CodecFactory.fromString(compressionCodec);
            this.dataFileWriter.setCodec(e);
            this.dataFileWriter.create(schema,out);
        } catch (AvroRuntimeException var5) {
            logger.warn("Unable to instantiate avro codec with name (" + compressionCodec + "). Compression disabled. Exception follows.", var5);
        } catch (IOException io){
            logger.warn("Could not open dataFileWriter (" + out.toString() + ". Exception follows.", io.getStackTrace());
        }

    }


    @Override
    public void write(Event event) throws IOException {
        AvroData destType = this.convert(event);
        System.out.println("The event being written is "+ event.getBody().toString());
        System.out.println("Dest type is "+ destType);
        this.dataFileWriter.append(destType);
    }


    @Override
    protected AvroData convert(Event event) {
        try {
            logger.info("event body is "+ event.getBody());
           return Utils.decodeAvroByteArray(event.getBody());
        } catch (InvalidProtocolBufferException i){
            return null;
        }
        catch(NumberFormatException e){
            logger.error("NumberFormatException thrown from decodeAvroByteArray");
            return null;
        }
        catch (IOException i){
            logger.error("IOException thrown from the decodeAvroByteArray");
            return null;
        }
    }

    public static class Container {
        //private final Map<String, String> headers;
        private final AvroData body;


        public Container(Map<String, String> headers, AvroData body) {
            super();
            //this.headers = headers;
            this.body = body;

        }
    }

    public static class Builder implements EventSerializer.Builder {

        @Override
        public EventSerializer build(Context context, OutputStream out) {
                System.out.println("Context sent to the builder is "+ context.toString());
                RawAvroHiveSerializer writer = new RawAvroHiveSerializer(out);
                writer.configure(context);
                return writer;
        }
    }


    //To adhere to the interface
    @Override
    public void afterCreate() throws IOException {
        //noop
    }

    @Override
    public void afterReopen() throws IOException {
        // impossible to initialize DataFileWriter without writing the schema?
        throw new UnsupportedOperationException("Avro API doesn't support append");
    }

    @Override
    public void flush() throws IOException {
        //noop
    }

    @Override
    public void beforeClose() throws IOException {
        //noop
    }

    @Override
    public boolean supportsReopen() {
        return true;
    }

}