/*
 * NIOUtils.java Sep 3, 2008
 *
 * Copyright 2007 Kobie Marketing, Inc. All rights reserved.
 */
package com.kobie.alchemy.util;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.lang.reflect.ParameterizedType;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.text.DateFormat;
import java.text.DecimalFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

import org.apache.log4j.Logger;

/**
 * TODO class description
 * 
 * @author Vlad Orzhekhovskiy
 */
public class NIOUtils
{
	/**
	 * This method is a wrapper around
	 * {@link FileChannel#map(MapMode, long, long)} method because of the Sun's
	 * bug# <code>6417205</code> leading to {@link OutOfMemoryError} problem.
	 * 
	 * @param chin channel to map.
	 * @param position position in the channel.
	 * @param size number of bytes to read.
	 * @return a byte buffer mapped to the channel.
	 */
	public static MappedByteBuffer mapChannel(FileChannel chin, long position,
			int size) throws IllegalStateException
	{
		// workaround for Sun's bug# 6417205
		MappedByteBuffer buffer = null;
		try
		{
			buffer = chin.map(MapMode.READ_ONLY, position, size);
		}
		catch (IOException e)
		{
			System.gc();
			System.runFinalization();
			try
			{
				buffer = chin.map(MapMode.READ_ONLY, position, size);
			}
			catch (IOException e1)
			{
				throw new IllegalStateException(e1);
			}
		}

		return buffer;
	}

	/**
	 * Collection of readers.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class ReaderArray
	{
		private final int MAX_BYTES_TO_READ = 1024 * 1024;
		private final Logger log = Logger.getLogger(getClass());

		private List<DataListener> listeners = new ArrayList<DataListener>();

		private List<DataReader<?>> readers = new ArrayList<DataReader<?>>();
		private Map<String, Integer> idxMap = new HashMap<String, Integer>();
		private List<String> names = new ArrayList<String>();

		private Object[] values = new Object[0];

		public void addField(DataReader<?> reader)
		{
			addField(Integer.toString(readers.size()), reader);
		}

		public void addField(String name, DataReader<?> reader)
		{
			names.add(name);
			idxMap.put(name, readers.size());

			readers.add(reader);
			ensureCapacity();
		}

		public void addDataListener(DataListener listener)
		{
			listeners.add(listener);
		}

		public int getFieldCount()
		{
			return readers.size();
		}

		public DataReader<?> getReader(int index)
		{
			return readers.get(index);
		}

		public String getName(int index)
		{
			return names.get(index);
		}

		public Object[] readRecord(ByteBuffer buffer)
		{
			for (int i = 0; i < readers.size(); i++)
			{
				DataReader<?> reader = readers.get(i);
				Object value = reader.read(buffer);
				values[i] = value;
			}

			return values;
		}

		public Object[] getValues()
		{
			return values;
		}

		public Object getValue(int index)
		{
			return values[index];
		}

		public Object getValue(String name)
		{
			int index = idxMap.get(name);
			return values[index];
		}

		public int getRecordSize()
		{
			int size = 0;
			for (DataReader<?> reader : readers)
				size += reader.getSize();

			return size;
		}

		public void processFile(File file) throws IOException
		{
			processFile(file, 0);
		}

		public void processFile(File file, int limit) throws IOException
		{
			FileInputStream fis = new FileInputStream(file);
			FileChannel chin = fis.getChannel();

			processFile(chin, limit);

			chin.close();
			fis.close();
		}

		public void processFile(FileChannel chin) throws IOException
		{
			processFile(chin, 0);
		}

		public void processFile(FileChannel chin, int limit) throws IOException
		{
			int recordSize = getRecordSize();
			int recordsPerChunk = MAX_BYTES_TO_READ / recordSize;

			int chunkSize = recordSize * recordsPerChunk;

			long position = chin.position();
			long channelSize = chin.size();

			for (DataListener listener : listeners)
				listener.processingStarted();

			int recordCount = 0;
			while (position < channelSize)
			{
				int size = (int) Math.min(chunkSize, channelSize - position);
				MappedByteBuffer buffer = NIOUtils.mapChannel(chin, position, size);
				if (log.isTraceEnabled())
					log.trace("Reading data chunk at offset " + position + "... ");

				int records = size / recordSize;
				for (int i = 0; i < records; i++)
				{
					Object[] values = readRecord(buffer);
					for (DataListener listener : listeners)
						listener.recordRetrieved(values, recordCount);

					if (log.isTraceEnabled())
						log.trace("Record # " + recordCount + " processed");

					recordCount++;
					if (limit > 0 && recordCount >= limit)
						break;
				}
				if (limit > 0 && recordCount >= limit)
					break;

				position += size;
			}

			for (DataListener listener : listeners)
				listener.processingComplete(recordCount);
		}

		public void dumpRecord()
		{
			dumpRecord(System.out);
		}

		public void dumpRecord(PrintStream out)
		{
			for (String name : names)
			{
				Object value = getValue(name);
				String format = getFormat(value);
				out.printf("%s: " + format + "\n", name, value);
			}
		}

		private void ensureCapacity()
		{
			if (readers.size() != values.length)
				values = new Object[readers.size()];
		}

		private static String getFormat(Object value)
		{
			if (value == null)
				return "%s";
			else if (value instanceof Date)
				return "%tF";
			else if (value instanceof Float || value instanceof Double ||
					value instanceof BigDecimal)
				return "%,f";
			else if (value instanceof Number)
				return "%,d";
			else
				return "%s";
		}
	}

	/**
	 * Data listener.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static interface DataListener
	{
		public void processingStarted();

		public void recordRetrieved(Object[] values, int recordCount);

		public void processingComplete(int recordCount);
	}

	/**
	 * Abstract implementation of {@link DataListener}.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static abstract class DataAdapter implements DataListener
	{
		public void processingComplete(int recordCount)
		{
		}

		public void processingStarted()
		{
		}

		public void recordRetrieved(Object[] values, int recordCount)
		{
		}
	}

	/**
	 * Reads data from byte buffer and converts the values to specified type.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static interface DataReader<T>
	{
		public int getSize();

		public T read(ByteBuffer buffer);
	}

	/**
	 * Abstract reader. Subclasses must implement {@link #convert(byte[])} method.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static abstract class AbstractDataReader<T> implements DataReader<T>
	{
		private final int size;

		protected AbstractDataReader(int size)
		{
			this.size = size;
		}

		public int getSize()
		{
			return size;
		}

		public abstract T read(ByteBuffer buffer);
	}

	/**
	 * Filler class. Reads the specified number of bytes, always returns
	 * <code>null</code>.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class Filler extends AbstractDataReader<Object>
	{
		private final byte[] filler;

		public Filler(int size)
		{
			super(size);
			filler = new byte[size];
		}

		@Override
		public Object read(ByteBuffer buffer)
		{
			buffer.get(filler);
			return null;
		}
	}

	/**
	 * Abstract reader based on converting the read data into string, and then -
	 * to the real data value.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static abstract class AsciiReader<T> extends AbstractDataReader<T>
	{
		private final ThreadLocal<byte[]> tl;

		protected AsciiReader(int size)
		{
			super(size);
			this.tl = new ThreadLocal<byte[]>();
		}

		public T read(ByteBuffer buffer)
		{
			byte[] data = internalByteArray();
			buffer.get(data);

			String str = new String(data);
			if (str == null)
				return null;

			str = str.trim();
			if (str.length() == 0 || str.startsWith("NULL"))
				return null;

			return convert(str);
		}

		private byte[] internalByteArray()
		{
			byte[] data = tl.get();
			if (data == null)
			{
				data = new byte[getSize()];
				tl.set(data);
			}

			return data;
		}

		protected abstract T convert(String str);
	}

	/**
	 * Reads numeric binary values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public abstract static class BinaryReader<T> extends AbstractDataReader<T>
	{
		private final ByteOrder byteOrder;

		public BinaryReader(int size)
		{
			this(size, null);
		}

		public BinaryReader(int size, ByteOrder byteOrder)
		{
			super(size);
			this.byteOrder = byteOrder;
		}

		@Override
		public T read(ByteBuffer buffer)
		{
			if (byteOrder != null)
			{
				ByteOrder bo = buffer.order();
				buffer.order(byteOrder);
				T value = fromBuffer(buffer);
				buffer.order(bo);

				return value;
			}
			else
				return fromBuffer(buffer);
		}

		protected abstract T fromBuffer(ByteBuffer buffer);
	}

	/**
	 * Reads literal values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class TextReader extends AsciiReader<String>
	{
		private final boolean intern;

		public TextReader(int size)
		{
			this(size, false);
		}

		public TextReader(int size, boolean intern)
		{
			super(size);
			this.intern = intern;
		}

		@Override
		protected String convert(String str)
		{
			return intern ? str.intern() : str;
		}
	}

	/**
	 * Reads single characters.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class CharReader extends AbstractDataReader<Character>
	{
		public CharReader()
		{
			super(1);
		}

		@Override
		public Character read(ByteBuffer buffer)
		{
			return (char) buffer.get();
		}
	}

	/**
	 * Reads integer values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class IntReader extends AsciiReader<Integer>
	{
		private final DecimalFormat fmt;

		public IntReader(int size)
		{
			super(size);
			this.fmt = null;
		}

		public IntReader(String format)
		{
			super(format.length());
			fmt = new DecimalFormat(format);
		}

		@Override
		protected Integer convert(String str)
		{
			try
			{
				return fmt != null ? fmt.parse(str).intValue() : Integer
						.valueOf(Integer.parseInt(str));
			}
			catch (Exception e)
			{
				throw new RuntimeException("Invalid integer: " + str);
			}
		}
	}

	/**
	 * Reads 32-bit binary integer values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryIntReader extends BinaryReader<Integer>
	{
		public BinaryIntReader()
		{
			this(null);
		}

		public BinaryIntReader(ByteOrder byteOrder)
		{
			super(Integer.SIZE / 8, byteOrder);
		}

		@Override
		protected Integer fromBuffer(ByteBuffer buffer)
		{
			return buffer.getInt();
		}
	}

	/**
	 * Reads short values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class ShortReader extends AsciiReader<Short>
	{
		public ShortReader(int size)
		{
			super(size);
		}

		@Override
		protected Short convert(String str)
		{
			return Short.valueOf(Short.parseShort(str));
		}
	}

	/**
	 * Reads 16-bit binary short values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryShortReader extends BinaryReader<Short>
	{
		public BinaryShortReader()
		{
			this(null);
		}

		public BinaryShortReader(ByteOrder byteOrder)
		{
			super(Short.SIZE / 8, byteOrder);
		}

		@Override
		protected Short fromBuffer(ByteBuffer buffer)
		{
			return buffer.getShort();
		}
	}

	/**
	 * Reads long values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class LongReader extends AsciiReader<Long>
	{
		public LongReader(int size)
		{
			super(size);
		}

		@Override
		protected Long convert(String str)
		{
			return Long.valueOf(Long.parseLong(str));
		}
	}

	/**
	 * Reads 64-bit binary long values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryLongReader extends BinaryReader<Long>
	{
		public BinaryLongReader()
		{
			this(null);
		}

		public BinaryLongReader(ByteOrder byteOrder)
		{
			super(Long.SIZE / 8, byteOrder);
		}

		@Override
		protected Long fromBuffer(ByteBuffer buffer)
		{
			return buffer.getLong();
		}
	}

	/**
	 * Reads byte values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class ByteReader extends AsciiReader<Byte>
	{
		public ByteReader(int size)
		{
			super(size);
		}

		@Override
		protected Byte convert(String str)
		{
			return Byte.valueOf(Byte.parseByte(str));
		}
	}

	/**
	 * Reads 1-byte values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryByteReader extends BinaryReader<Byte>
	{
		public BinaryByteReader()
		{
			this(null);
		}

		public BinaryByteReader(ByteOrder byteOrder)
		{
			super(Byte.SIZE / 8, byteOrder);
		}

		@Override
		protected Byte fromBuffer(ByteBuffer buffer)
		{
			return buffer.get();
		}
	}

	/**
	 * Reads float values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class FloatReader extends AsciiReader<Float>
	{
		public FloatReader(int size)
		{
			super(size);
		}

		@Override
		protected Float convert(String str)
		{
			return Float.valueOf(str);
		}
	}

	/**
	 * Reads 32-bit binary float values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryFloatReader extends BinaryReader<Float>
	{
		public BinaryFloatReader()
		{
			this(null);
		}

		public BinaryFloatReader(ByteOrder byteOrder)
		{
			super(Float.SIZE / 8, byteOrder);
		}

		@Override
		protected Float fromBuffer(ByteBuffer buffer)
		{
			return buffer.getFloat();
		}
	}

	/**
	 * Reads double values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class DoubleReader extends AsciiReader<Double>
	{
		public DoubleReader(int size)
		{
			super(size);
		}

		@Override
		protected Double convert(String str)
		{
			return Double.valueOf(str);
		}
	}

	/**
	 * Reads 64-bit binary double values.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BinaryDoubleReader extends BinaryReader<Double>
	{
		public BinaryDoubleReader()
		{
			this(null);
		}

		public BinaryDoubleReader(ByteOrder byteOrder)
		{
			super(Double.SIZE / 8, byteOrder);
		}

		@Override
		protected Double fromBuffer(ByteBuffer buffer)
		{
			return buffer.getDouble();
		}
	}

	/**
	 * Reads decimal values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class DecimalReader extends AsciiReader<BigDecimal>
	{
		private final DecimalFormat fmt;

		public DecimalReader(int size)
		{
			super(size);
			this.fmt = null;
		}

		public DecimalReader(String format)
		{
			super(format.length());
			this.fmt = new DecimalFormat(format);
			this.fmt.setParseBigDecimal(true);
		}

		@Override
		protected BigDecimal convert(String str)
		{
			try
			{
				return fmt != null ? (BigDecimal) fmt.parse(str) : new BigDecimal(str);
			}
			catch (Exception e)
			{
				throw new RuntimeException("Invalid decimal: " + str);
			}
		}
	}

	/**
	 * Reads big integer values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class BigIntReader extends AsciiReader<BigInteger>
	{
		public BigIntReader(int size)
		{
			super(size);
		}

		@Override
		protected BigInteger convert(String str)
		{
			return new BigInteger(str);
		}
	}

	/**
	 * Reads date values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class DateReader extends AsciiReader<Date>
	{
		private final DateFormat df;

		public DateReader(String format)
		{
			this(format.length(), new SimpleDateFormat(format));
		}

		public DateReader(int size, DateFormat df)
		{
			super(size);
			this.df = df;
		}

		@Override
		protected Date convert(String str)
		{
			try
			{
				Date date = df.parse(str);
				return date.getTime() > 0 ? date : null;
			}
			catch (ParseException e)
			{
				throw new RuntimeException("Invalid date: " + str);
			}
		}
	}

	/**
	 * Reads enumerated values in ASCII format.
	 * 
	 * @author Vlad Orzhekhovskiy
	 */
	public static class EnumReader<T extends Enum<T>> extends AsciiReader<T>
	{
		private final Class<T> enumType;

		@SuppressWarnings("unchecked")
		public EnumReader(int size)
		{
			super(size);
			ParameterizedType type = (ParameterizedType) getClass()
					.getGenericSuperclass();
			this.enumType = (Class<T>) type.getActualTypeArguments()[0];
		}

		@Override
		protected T convert(String str)
		{
			return Enum.valueOf(enumType, str);
		}
	}
}