/*
 *  Copyright (c) 2014 DataTorrent, Inc. ALL Rights Reserved.
 */
package com.datatorrent.flume.storage;

import java.util.HashMap;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;

import com.datatorrent.lib.codec.KryoSerializableStreamCodec;

/**
 *
 * @author Chetan Narsude <chetan@datatorrent.com>
 */
public class EventCodec extends KryoSerializableStreamCodec<Event>
{
  public EventCodec()
  {
    super();
    kryo.addDefaultSerializer(Event.class, new EventSerializer());
  }

  public static class EventSerializer extends Serializer<Event>
  {
    @Override
    public void write(Kryo kryo, Output output, Event object)
    {
      kryo.writeObject(output, object.getHeaders());
      kryo.writeObject(output, object.getBody());
    }

    @Override
    public Event read(Kryo kryo, Input input, Class<Event> type)
    {
      @SuppressWarnings("unchecked")
      HashMap<String, String> headers = kryo.readObject(input, HashMap.class);
      byte[] body = kryo.readObject(input, byte[].class);
      return EventBuilder.withBody(body, headers);
    }

  }

}

@@ -102,16 +102,91 @@
 orage;%0A%0A
+import java.io.ByteArrayInputStream;%0Aimport java.io.ByteArrayOutputStream;%0A
 import j
@@ -247,53 +247,8 @@
 yo;%0A
-import com.esotericsoftware.kryo.Serializer;%0A
 impo
@@ -434,45 +434,66 @@
 ent.
-lib.codec.KryoSerializableStreamCodec
+api.StreamCodec;%0A%0Aimport com.datatorrent.common.util.Slice
 ;%0A%0A/
@@ -581,32 +581,19 @@
 dec 
-extend
+implement
 s 
-KryoSerializable
 Stre
@@ -616,186 +616,180 @@
 %0A  p
-ublic EventCodec()%0A  %7B%0A    super();%0A    kryo.addDefaultSerializer(Event.class, new EventSerializer());%0A  %7D%0A%0A  public static class EventSerializer extends Serializer%3CEvent%3E%0A  %7B%0A  
+rivate transient final Kryo kryo;%0A%0A  public EventCodec()%0A  %7B%0A    this.kryo = new Kryo();%0A    this.kryo.setClassLoader(Thread.currentThread().getContextClassLoader());%0A  %7D%0A%0A
   @O
@@ -788,34 +788,32 @@
  %7D%0A%0A  @Override%0A
-  
   public void wr
@@ -809,69 +809,53 @@
 lic 
-void write(Kryo kryo, Output output, Event objec
+Object fromByteArray(Slice fragmen
 t)%0A
-  
   %7B%0A    
   kr
@@ -854,201 +854,145 @@
     
-  kryo.writeObject(output, object.getHeaders());%0A      kryo.writeObject(output, object.getBody());%0A    %7D%0A%0A    @Override%0A    public Event read(Kryo kryo, Input input, Class%3CEvent%3E type)%0A    %7B%0A  
+ByteArrayInputStream is = new ByteArrayInputStream(fragment.buffer, fragment.offset, fragment.length);%0A    Input input = new Input(is);%0A%0A
     
@@ -1026,18 +1026,16 @@
 d%22)%0A    
-  
 HashMap%3C
@@ -1103,18 +1103,16 @@
 s);%0A    
-  
 byte%5B%5D b
@@ -1159,18 +1159,16 @@
 s);%0A    
-  
 return E
@@ -1206,20 +1206,436 @@
 ers);%0A  
-  
 %7D%0A
+%0A  @Override%0A  public Slice toByteArray(Event event)%0A  %7B%0A    ByteArrayOutputStream os = new ByteArrayOutputStream();%0A    Output output = new Output(os);%0A    kryo.writeObject(output, event.getHeaders());%0A    kryo.writeObject(output, event.getBody());%0A    output.flush();%0A    return new Slice(os.toByteArray(), 0, os.toByteArray().length);%0A  %7D%0A%0A  @Override%0A  public int getPartition(Event o)%0A  %7B%0A    return o.hashCode();
 %0A  %7D%0A%0A%7D%0A

Implement native EventCodec.