package com.linkedin.camus.etl.kafka.coders;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Properties;
import java.text.SimpleDateFormat;

import com.google.gson.JsonParser;
import com.google.gson.JsonObject;

import com.linkedin.camus.coders.CamusWrapper;
import com.linkedin.camus.coders.MessageDecoder;
import com.linkedin.camus.coders.MessageDecoderException;

import org.apache.log4j.Logger;


/**
 * MessageDecoder class that will convert the payload into a JSON object,
 * look for a the camus.message.timestamp.field, convert that timestamp to
 * a unix epoch long using camus.message.timestamp.format, and then set the CamusWrapper's
 * timestamp property to the record's timestamp.  If the JSON does not have
 * a timestamp or if the timestamp could not be parsed properly, then
 * System.currentTimeMillis() will be used.
 *
 * camus.message.timestamp.format will be used with SimpleDateFormat.  If your
 * camus.message.timestamp.field is stored in JSON as a long, then you may
 * set camus.message.timestamp.format to 'unix' to avoid having your timestamp
 * converted.
 *
 * This MessageDecoder returns a CamusWrapper that works with Strings payloads,
 * since JSON data is always a String.
 */
public class JsonStringMessageDecoder extends MessageDecoder<byte[], String> {
	private static org.apache.log4j.Logger log = Logger.getLogger(JsonStringMessageDecoder.class);

	// Property for format of timestamp in JSON timestamp field.
	public  static final String CAMUS_MESSAGE_TIMESTAMP_FORMAT = "camus.message.timestamp.format";
	public  static final String DEFAULT_TIMESTAMP_FORMAT       = "[dd/MMM/yyyy:HH:mm:ss Z]";

	// Property for the JSON field name of the timestamp.
	public  static final String CAMUS_MESSAGE_TIMESTAMP_FIELD  = "camus.message.timestamp.field";
	public  static final String DEFAULT_TIMESTAMP_FIELD        = "timestamp";

	private String timestampFormat;
	private String timestampField;

	@Override
	public void init(Properties props, String topicName) {
		this.props     = props;
		this.topicName = topicName;

		timestampFormat = props.getProperty(CAMUS_MESSAGE_TIMESTAMP_FORMAT, DEFAULT_TIMESTAMP_FORMAT);
		timestampField  = props.getProperty(CAMUS_MESSAGE_TIMESTAMP_FIELD,  DEFAULT_TIMESTAMP_FIELD);
	}

	@Override
	public CamusWrapper<String> decode(byte[] payload) {
		long       timestamp = 0;
		String     payloadString;
		JsonObject jsonObject;

		payloadString =  new String(payload);

		// Parse the payload into a JsonObject.
		try {
			jsonObject = new JsonParser().parse(payloadString).getAsJsonObject();
		} catch (RuntimeException e) {
			log.error("Caught exception while parsing JSON string '" + payloadString + "'.");
			throw new RuntimeException(e);
		}

		// Attempt to read and parse the timestamp element into a long.
		if (jsonObject.has(timestampField)) {
			// If timestampFormat is 'unix',
			// then the timestamp should be stored as unix epoch timestamp.
			if (timestampFormat.equals("unix")) {
				timestamp = jsonObject.get(timestampField).getAsLong();
			}
			// Otherwise parse the timestamp as a string in timestampFormat.
			else {
				String timestampString = jsonObject.get(timestampField).getAsString();
				try {
					timestamp = new SimpleDateFormat(timestampFormat).parse(timestampString).getTime();
				} catch (Exception e) {
					log.error("Could not parse timestamp '" + timestampString + "' while decoding JSON message.");
				}
			}
		}

		// If timestamp wasn't set in the above block,
		// then set it to current time.
		if (timestamp == 0) {
			log.warn("Couldn't find or parse timestamp field '" + timestampField + "' in JSON message, defaulting to current time.");
			timestamp = System.currentTimeMillis();
		}

		return new CamusWrapper<String>(payloadString, timestamp);
	}
}

@@ -3046,16 +3046,325 @@
 Long();%0A
+%09%09%09%09// if timestamp is larger than 1000000000000, it is%0A%09%09%09%09// already in milliseconds...unless the timestamp%0A%09%09%09%09// is meant to be after the year 33658 :p%0A%09%09%09%09if (timestamp %3C 1000000000000L) %7B%0A%09%09%09%09%09// This timestamp is in seconds!%0A%09%09%09%09%09// Convert it to milliseconds.%0A%09%09%09%09%09timestamp = timestamp * 1000;%0A%09%09%09%09%7D%0A
 %09%09%09%7D%0A%09%09%09

Automatically converting to millisecond timestamp if the value is large enough