/**
 * Copyright (c) 2000-2013 Liferay, Inc. All rights reserved.
 *
 * The contents of this file are subject to the terms of the Liferay Enterprise
 * Subscription License ("License"). You may not use this file except in
 * compliance with the License. You can obtain a copy of the License by
 * contacting Liferay, Inc. See the License for the specific language governing
 * permissions and limitations under the License, including but not limited to
 * distribution rights of the Software.
 *
 *
 *
 */

package com.liferay.portal.search.lucene.dump;

import com.liferay.portal.kernel.io.ProtectedObjectInputStream;
import com.liferay.portal.kernel.log.Log;
import com.liferay.portal.kernel.log.LogFactoryUtil;
import com.liferay.portal.search.lucene.dump.IndexCommitMetaInfo.Segment;
import com.liferay.portal.util.PropsValues;

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

import java.util.List;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.apache.lucene.index.IndexCommit;
import org.apache.lucene.index.SegmentInfos;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.IndexOutput;

/**
 * @author Shuyang Zhou
 */
public class IndexCommitSerializationUtil {

	public static void deserializeIndex(
			InputStream inputStream, Directory directory)
		throws IOException {

		if (PropsValues.INDEX_DUMP_COMPRESSION_ENABLED) {
			inputStream = new GZIPInputStream(inputStream);
		}

		ObjectInputStream objectInputStream = null;

		try {
			objectInputStream = new ProtectedObjectInputStream(inputStream);

			IndexCommitMetaInfo indexCommitMetaInfo = null;

			try {
				indexCommitMetaInfo =
					(IndexCommitMetaInfo)objectInputStream.readObject();
			}
			catch (ClassNotFoundException cnfe) {
				throw new IOException(cnfe.getMessage());
			}

			if (_log.isDebugEnabled()) {
				_log.debug("Deserializing " + indexCommitMetaInfo);
			}

			if (indexCommitMetaInfo.isEmpty()) {
				return;
			}

			List<Segment> segments = indexCommitMetaInfo.getSegments();

			for (Segment segment : segments) {
				if (_log.isDebugEnabled()) {
					_log.debug("Deserializing segment " + segment);
				}

				_deserializeSegment(
					objectInputStream, segment.getFileSize(),
					directory.createOutput(segment.getFileName()));
			}

			_writeSegmentsGen(directory, indexCommitMetaInfo.getGeneration());
		}
		finally {
			if (objectInputStream != null) {
				objectInputStream.close();
			}
		}
	}

	public static void serializeIndex(
			IndexCommit indexCommit, OutputStream outputStream)
		throws IOException {

		if (PropsValues.INDEX_DUMP_COMPRESSION_ENABLED) {
			outputStream = new GZIPOutputStream(outputStream);
		}

		ObjectOutputStream objectOputStream = new ObjectOutputStream(
			outputStream);

		IndexCommitMetaInfo indexCommitMetaInfo = new IndexCommitMetaInfo(
			indexCommit);

		if (_log.isDebugEnabled()) {
			_log.debug("Serializing " + indexCommitMetaInfo);
		}

		objectOputStream.writeObject(indexCommitMetaInfo);

		List<Segment> segments = indexCommitMetaInfo.getSegments();

		Directory directory = indexCommit.getDirectory();

		for (Segment segment : segments) {
			if (_log.isDebugEnabled()) {
				_log.debug("Serializing segment " + segment);
			}

			_serializeSegment(
				directory.openInput(segment.getFileName()),
				segment.getFileSize(), objectOputStream);
		}

		objectOputStream.flush();

		if (PropsValues.INDEX_DUMP_COMPRESSION_ENABLED) {
			GZIPOutputStream gZipOutputStream = (GZIPOutputStream)outputStream;

			gZipOutputStream.finish();
		}
	}

	private static void _deserializeSegment(
			InputStream inputStream, long length, IndexOutput indexOutput)
		throws IOException {

		try {
			indexOutput.setLength(length);

			byte[] buffer = new byte[_BUFFER_SIZE];

			long received = 0;

			while (received < length) {
				int bufferSize = _BUFFER_SIZE;

				if ((received + _BUFFER_SIZE) > length) {
					bufferSize = (int)(length - received);
				}

				int actualSize = inputStream.read(buffer, 0, bufferSize);

				indexOutput.writeBytes(buffer, actualSize);

				received += actualSize;
			}
		}
		finally {
			indexOutput.close();
		}
	}

	private static void _serializeSegment(
			IndexInput indexInput, long length, OutputStream outputStream)
		throws IOException {

		byte[] buffer = new byte[_BUFFER_SIZE];

		int count = (int)(length / _BUFFER_SIZE);
		int tail = (int)(length - count * _BUFFER_SIZE);

		try {
			for (int i = 0; i < count; i++) {
				indexInput.readBytes(buffer, 0, _BUFFER_SIZE);
				outputStream.write(buffer);
			}

			indexInput.readBytes(buffer, 0, tail);
			outputStream.write(buffer, 0, tail);
		}
		finally {
			indexInput.close();
		}
	}

	private static void _writeSegmentsGen(Directory directory, long generation)
		throws IOException {

		if (_log.isDebugEnabled()) {
			_log.debug(
				"Writing " + _SEGMENTS_GEN_FILE_NAME + " with generation " +
					generation);
		}

		IndexOutput indexOutput = directory.createOutput(
			_SEGMENTS_GEN_FILE_NAME);

		try {
			indexOutput.writeInt(SegmentInfos.FORMAT_LOCKLESS);
			indexOutput.writeLong(generation);
			indexOutput.writeLong(generation);
		}
		finally {
			indexOutput.close();
		}
	}

	private static final int _BUFFER_SIZE = 8192;

	private static final String _SEGMENTS_GEN_FILE_NAME = "segments.gen";

	private static Log _log = LogFactoryUtil.getLog(
		IndexCommitSerializationUtil.class);

}