package com.xqbase.baiji;

import com.xqbase.baiji.specific.SpecificRecord;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * Serializer with JSON encoding.
 *
 * @author Tony He
 */
public class JSONSerializer implements Serializer {

    @Override
    public <T extends SpecificRecord> void serialize(T obj, OutputStream stream) throws IOException {

    }

    @Override
    public <T extends SpecificRecord> T deserialize(Class<T> objClass, InputStream stream) throws IOException {
        return null;
    }
}

@@ -20,16 +20,300 @@
 baiji;%0A%0A
+import com.xqbase.baiji.io.DatumReader;%0Aimport com.xqbase.baiji.io.DatumWriter;%0Aimport com.xqbase.baiji.io.DirectBinaryDecoder;%0Aimport com.xqbase.baiji.io.DirectBinaryEncoder;%0Aimport com.xqbase.baiji.specific.SpecificDatumReader;%0Aimport com.xqbase.baiji.specific.SpecificDatumWriter;%0A
 import c
@@ -353,16 +353,16 @@
 Record;%0A
-
 %0Aimport 
@@ -438,16 +438,101 @@
 tStream;
+%0Aimport java.lang.reflect.Constructor;%0Aimport java.util.concurrent.ConcurrentHashMap;
 %0A%0A/**%0A *
@@ -638,24 +638,239 @@
 rializer %7B%0A%0A
+    private static final ConcurrentHashMap%3CClass%3C?%3E, DatumWriter%3E writerCache = new ConcurrentHashMap%3C%3E();%0A    private static final ConcurrentHashMap%3CClass%3C?%3E, DatumReader%3E readerCache = new ConcurrentHashMap%3C%3E();%0A%0A
     @Overrid
@@ -973,17 +973,124 @@
 ption %7B%0A
+        DatumWriter%3CT%3E writer = getWriter(obj);%0A        writer.write(obj, new DirectBinaryEncoder(stream));
 %0A
-
     %7D%0A%0A 
@@ -1226,19 +1226,1428 @@
     
-return null
+DatumReader%3CT%3E reader = getReader(objClass);%0A        return reader.read(null, new DirectBinaryDecoder(stream));%0A    %7D%0A%0A    @SuppressWarnings(%22unchecked%22)%0A    private %3CT extends SpecificRecord%3E DatumWriter%3CT%3E getWriter(T obj) %7B%0A        Class%3C?%3E clazz = obj.getClass();%0A        DatumWriter%3CT%3E writer = writerCache.get(clazz);%0A        if (null == writer) %7B%0A            writer = new SpecificDatumWriter(obj.getSchema());%0A            DatumWriter%3CT%3E existedWriter = writerCache.putIfAbsent(clazz, writer);%0A            if (existedWriter != null) %7B%0A                writer = existedWriter;%0A            %7D%0A        %7D%0A        return writer;%0A    %7D%0A%0A    @SuppressWarnings(%22unchecked%22)%0A    private %3CT extends SpecificRecord%3E DatumReader%3CT%3E getReader(Class%3CT%3E clazz) %7B%0A        DatumReader%3CT%3E reader = readerCache.get(clazz);%0A        if (null == reader) %7B%0A            SpecificRecord record;%0A            try %7B%0A                Constructor%3CT%3E ctor = clazz.getDeclaredConstructor();%0A                ctor.setAccessible(true);%0A                record = ctor.newInstance();%0A            %7D catch (Exception e) %7B%0A                throw new RuntimeException(e);%0A            %7D%0A            reader = new SpecificDatumReader(record.getSchema());%0A            DatumReader%3CT%3E existedReader = readerCache.putIfAbsent(clazz, reader);%0A            if (existedReader != null) %7B%0A                reader = existedReader;%0A            %7D%0A        %7D%0A        return reader
 ;%0A  

add the JSONSerializer....