Twitter4JStatusCustomSerializer.java

package com.twitsprout.smartalerts.storm;

import java.nio.ByteBuffer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import twitter4j.Status;
import twitter4j.TwitterException;
import twitter4j.json.DataObjectFactory;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.serialize.CustomSerializer;

public class Twitter4JStatusCustomSerializer extends CustomSerializer {

	private final Logger logger;

	public Twitter4JStatusCustomSerializer(Kryo kryo) {
		super(kryo);
		logger = LoggerFactory.getLogger(Twitter4JStatusCustomSerializer.class);
	}
	
	@Override
	public void writeObjectData(ByteBuffer buffer, Object object) {
		String rawJson = DataObjectFactory.getRawJSON(object);
		buffer.putShort((short) rawJson.length());
		buffer.put(DataObjectFactory.getRawJSON(object).getBytes());
	}
	
	@Override
	public Status readObjectData(ByteBuffer buffer, Class type) {
		short length = buffer.getShort();
		byte[] jsonBuffer = new byte[length];
		buffer.get(jsonBuffer);
		try {
			return DataObjectFactory.createStatus(new String(jsonBuffer));
		} catch (TwitterException e) {
			logger.error("Could not deserialize", e);
			return null;
		}
	}

}