DateTimeSerializer.java

package com.evntcast.server.storm;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormatterBuilder;

import backtype.storm.serialization.ISerialization;

public class DateTimeSerializer implements ISerialization<DateTime> {
	private static final String DATE_FORMAT = "YYYY-MM-dd'T'HH:mm:ssZZ";
	private static final int DATE_FORMAT_LENGTH = new DateTime().toString(DATE_FORMAT).length();

	@SuppressWarnings("rawtypes")
	@Override
	public boolean accept(Class clazz) {
		return clazz.getName().equals("org.joda.time.DateTime");
	}

	@Override
	public DateTime deserialize(DataInputStream stream)
			throws IOException {
		byte[] buffer = new byte[DATE_FORMAT_LENGTH];
		stream.read(buffer, 0, buffer.length);

		return new DateTime( new String(buffer));
	}

	@Override
	public void serialize(DateTime object, DataOutputStream stream)
			throws IOException {
		stream.write(object.toString(DATE_FORMAT).getBytes());
	}
}