/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2016 All Rights Reserved.
 */
package com.alipay.rdf.file.split;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;

import com.alipay.rdf.file.exception.RdfErrorEnum;
import com.alipay.rdf.file.exception.RdfFileException;
import com.alipay.rdf.file.interfaces.FileStorage;
import com.alipay.rdf.file.loader.ProtocolLoader;
import com.alipay.rdf.file.loader.TemplateLoader;
import com.alipay.rdf.file.meta.FileMeta;
import com.alipay.rdf.file.model.FileConfig;
import com.alipay.rdf.file.model.FileDataTypeEnum;
import com.alipay.rdf.file.model.FileSlice;
import com.alipay.rdf.file.spi.RdfFileSplitterSpi;
import com.alipay.rdf.file.storage.FileNasStorage.BoundedInputStream;
import com.alipay.rdf.file.util.RdfFileLogUtil;

/**
 * Copyright (C) 2013-2018 Ant Financial Services Group
 * 
 * nas 存储实现
 * 
 * @author hongwei.quhw
 * @version $Id: NasFileSliceSplitter.java, v 0.1 2017年7月31日 下午4:07:39 hongwei.quhw Exp $
 */
public class NasFileSliceSplitter implements RdfFileSplitterSpi {

    @Override
    public void init(FileStorage fileStorage) {
    }

    /** 
     * @see com.alipay.rdf.file.FileSplitter#split(java.lang.String, int)
     */
    @Override
    public List<FileSlice> split(String path, int sliceSize) {
        File file = new File(path);
        if (!file.exists()) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.split(path=" + path
                                       + ", sliceSize=" + sliceSize + ") 文件不存在",
                RdfErrorEnum.NOT_EXSIT);
        }

        return split(file, new FileSlice(path, FileDataTypeEnum.UNKOWN, 0, file.length()),
            sliceSize);
    }

    private List<FileSlice> split(File file, FileSlice fileSlice, int sliceSize) {
        if (fileSlice.getLength() < 0 || sliceSize <= 0 || fileSlice.getEnd() > file.length()) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.split(fileSlice=" + fileSlice
                                       + ", sliceSize=" + sliceSize
                                       + ") fileSlice.getLength()必须大于等于零，sliceSize必须大于零， fileSlice.getEnd() 不能大于 file.length()",
                RdfErrorEnum.ILLEGAL_ARGUMENT);
        }

        List<FileSlice> fileSlices = new ArrayList<FileSlice>();
        String filePath = fileSlice.getFilePath();

        long length = file.length();
        if (length == 0) {
            RdfFileLogUtil.common
                .info("rdf-file#NasFileSliceSplitter.split path=" + filePath + ", start=0, end=0");
            FileSlice slice = new FileSlice(filePath, fileSlice.getFileDataType(), 0, 0);
            fileSlices.add(slice);
            return fileSlices;
        } else if (RdfFileLogUtil.common.isDebug()) {
            RdfFileLogUtil.common.debug("rdf-file#NasFileSliceSplitter.split fileSize=" + length);
        }

        //文件体大小小于分片大小
        if (fileSlice.getLength() <= sliceSize) {
            fileSlices.add(fileSlice);
            if (RdfFileLogUtil.common.isDebug()) {
                RdfFileLogUtil.common
                    .debug("rdf-file#NasFileSliceSplitter.split fileSlice.getLength()="
                           + fileSlice.getLength() + " <= sliceSize=" + sliceSize);
            }
            return fileSlices;
        }

        BoundedInputStream raf = null;
        try {
            long rangeStart = fileSlice.getStart();
            boolean eof = false;
            raf = new BoundedInputStream(file, rangeStart, fileSlice.getLength());

            while (!eof) {
                raf.skip(sliceSize);
                boolean eol = false;
                while (!eol && !eof) {
                    switch (raf.read()) {
                        case -1:
                            eof = true;
                            break;
                        case '\n':
                            eol = true;
                            break;
                        case '\r':
                            eol = true;
                            long cur = raf.getFilePointer();
                            if ((raf.read()) != '\n') {
                                raf.seek(cur);
                            }
                            break;
                        default:
                            break;
                    }
                }
                if (rangeStart != raf.getFilePointer()) {
                    FileSlice slice = new FileSlice(filePath, fileSlice.getFileDataType(),
                        rangeStart, raf.getFilePointer());
                    fileSlices.add(slice);
                    if (RdfFileLogUtil.common.isDebug()) {
                        RdfFileLogUtil.common
                            .debug("rdf-file#NasFileSliceSplitter.split add slice slice=" + slice);
                    }
                    rangeStart = raf.getFilePointer();
                }
            }
        } catch (FileNotFoundException e) {
            throw new RdfFileException(
                "rdf-file#NasFileSliceSplitter.split file=" + file.getAbsolutePath() + " 文件不存在", e,
                RdfErrorEnum.NOT_EXSIT);
        } catch (IOException e) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.split io 异常", e,
                RdfErrorEnum.NOT_EXSIT);
        } finally {
            if (raf != null) {
                raf.close();
            }
        }

        return fileSlices;
    }

    /** 
     * @see com.alipay.rdf.file.interfaces.FileSplitter#getHeadSlice(com.alipay.rdf.file.model.FileConfig)
     */
    @Override
    public FileSlice getHeadSlice(FileConfig fileConfig) {
        String filePath = fileConfig.getFilePath();
        RdfFileLogUtil.common.info(
            "rdf-file#NasFileSliceSplitter.getHeadSlice(fileConfig.filePath=" + filePath + ")");

        File file = new File(filePath);
        if (!file.exists()) {
            throw new RdfFileException("rdf-file#filePath=" + filePath + ", 不存在文件",
                RdfErrorEnum.NOT_EXSIT);
        }

        long length = file.length();
        if (0 == length) {
            RdfFileLogUtil.common.warn(
                "rdf-file#NasFileSliceSplitter.getHeadSlice filePath=" + filePath + ", size=0");
            return new FileSlice(filePath, FileDataTypeEnum.HEAD, 0, 0);
        } else {
            RdfFileLogUtil.common
                .debug("rdf-file#NasFileSliceSplitter.getHeadSlice fileSize=" + length);
        }

        if (!fileConfig.isPartial() && (fileConfig.getFileDataType() == FileDataTypeEnum.BODY
                                        || fileConfig.getFileDataType() == FileDataTypeEnum.TAIL)) {
            return new FileSlice(fileConfig.getFilePath(), FileDataTypeEnum.HEAD, 0, 0);
        }

        FileMeta fileMeta = TemplateLoader.load(fileConfig.getTemplatePath(),
            fileConfig.getTemplateEncoding());
        if (!fileMeta.hasHead()) {
            throw new RdfFileException(
                "rdf-file#文件模板template=" + fileConfig.getTemplatePath() + ", 没有定义头",
                RdfErrorEnum.HEAD_NOT_DEFINED);
        }

        int headRowsAffected = ProtocolLoader.getRowsAfftected(fileConfig, FileDataTypeEnum.HEAD);
        if (headRowsAffected == 0) {
            RdfFileLogUtil.common
                .info("rdf-file#NasFileSliceSplitter.getHeadSlice file headRowsAfftected="
                      + headRowsAffected);
            return new FileSlice(filePath, FileDataTypeEnum.HEAD, 0, 0);
        } else {
            RdfFileLogUtil.common
                .debug("rdf-file#NasFileSliceSplitter.getHeadSlice file headRowsAfftected="
                       + headRowsAffected);
        }

        RandomAccessFile raf = null;
        int headLineCount = 0;
        int count = 0;
        boolean next = true;
        try {
            raf = new RandomAccessFile(file, "r");
            raf.seek(0);

            while (next) {
                count++;
                int b = raf.read();
                switch (b) {
                    case '\n':
                    case '\r':
                        long cur = raf.getFilePointer();
                        int pre = raf.read();
                        if (pre == -1 || pre != '\n') {
                            headLineCount++;
                            if (RdfFileLogUtil.common.isDebug()) {
                                RdfFileLogUtil.common.debug(
                                    "rdf-file#NasFileSliceSplitter.getHeadSlice headLineCount++, linebreak="
                                                            + (b == '\r' ? "\\r" : "\\n"));
                            }
                        }

                        if (headLineCount == headRowsAffected) {
                            next = false;
                            break;
                        }

                        raf.seek(cur);

                        break;
                    case -1:
                        count--;
                        next = false;
                        if (RdfFileLogUtil.common.isWarn()) {
                            RdfFileLogUtil.common
                                .warn("rdf-file#NasFileSliceSplitter.getHeadSlice char count="
                                      + count + ", read() == -1");
                        }
                        break;
                    default:
                        break;
                }
            }
        } catch (FileNotFoundException e) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.getHeadSlice file="
                                       + file.getAbsolutePath() + " 文件不存在",
                e, RdfErrorEnum.NOT_EXSIT);
        } catch (IOException e) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.getHeadSlice io 异常", e,
                RdfErrorEnum.NOT_EXSIT);
        } finally {
            if (null != raf) {
                try {
                    raf.close();
                } catch (IOException e) {
                    if (RdfFileLogUtil.common.isWarn()) {
                        RdfFileLogUtil.common
                            .warn("rdf-file#NasFileSliceSplitter.getHeadSlice close 异常", e);
                    }
                }
            }
        }

        if (RdfFileLogUtil.common.isInfo()) {
            RdfFileLogUtil.common.info("rdf-file#NasFileSliceSplitter.getHeadSlice filePath="
                                       + filePath + " ,char count=" + count);
        }

        return new FileSlice(filePath, FileDataTypeEnum.HEAD, 0, count);
    }

    /** 
     * @see com.alipay.rdf.file.interfaces.FileSplitter#getBodySlice(com.alipay.rdf.file.model.FileConfig)
     */
    @Override
    public FileSlice getBodySlice(FileConfig fileConfig) {
        String filePath = fileConfig.getFilePath();
        RdfFileLogUtil.common
            .info("rdf-file#NasFileSliceSplitter.getBodySlice filePath=" + filePath);
        File file = new File(filePath);
        if (!file.exists()) {
            throw new RdfFileException(
                "rdf-file#NasFileSliceSplitter.getBodySlice filePath=" + filePath + ", 不存在文件",
                RdfErrorEnum.NOT_EXSIT);
        }

        long length = file.length();
        if (length == 0) {
            RdfFileLogUtil.common.info("rdf-file#NasFileSliceSplitter.getBodySlice filePath="
                                       + filePath + ", start=0, end=0");
            return new FileSlice(filePath, FileDataTypeEnum.BODY, 0, 0);
        } else if (RdfFileLogUtil.common.isDebug()) {
            RdfFileLogUtil.common
                .debug("rdf-file#NasFileSliceSplitter.getBodySlice fileSize=" + length);
        }

        long start = 0;
        long end = length;

        try {
            FileSlice headSlice = getHeadSlice(fileConfig);
            start = headSlice.getEnd();
        } catch (RdfFileException e) {
            if (!RdfErrorEnum.HEAD_NOT_DEFINED.equals(e.getErrorEnum())) {
                throw e;
            }
        }

        try {
            FileSlice tailSlice = getTailSlice(fileConfig);
            end = tailSlice.getStart();
        } catch (RdfFileException e) {
            if (!RdfErrorEnum.TAIL_NOT_DEFINED.equals(e.getErrorEnum())) {
                throw e;
            }
        }
        if (RdfFileLogUtil.common.isDebug()) {
            RdfFileLogUtil.common.debug(
                "rdf-file#NasFileSliceSplitter.getBodySlice start=" + start + ", end=" + end);
        }

        return new FileSlice(filePath, FileDataTypeEnum.BODY, start, end);
    }

    /** 
     * @see com.alipay.rdf.file.interfaces.FileSplitter#getBodySlices(com.alipay.rdf.file.model.FileConfig, int)
     */
    @Override
    public List<FileSlice> getBodySlices(FileConfig fileConfig, int sliceSize) {
        String filePath = fileConfig.getFilePath();
        RdfFileLogUtil.common.info("rdf-file#NasFileSliceSplitter.getBodySlices(filePath="
                                   + filePath + ", sliceSize=" + sliceSize + ")");

        File file = new File(filePath);
        if (!file.exists()) {
            throw new RdfFileException(
                "rdf-file#NasFileSliceSplitter.getBodySlice filePath=" + filePath + ", 不存在文件",
                RdfErrorEnum.NOT_EXSIT);
        }

        FileSlice slice = null;
        if (fileConfig.isPartial()) {
            slice = new FileSlice(filePath, fileConfig.getFileDataType(), fileConfig.getOffset(),
                fileConfig.getOffset() + fileConfig.getLength());
        } else {
            slice = getBodySlice(fileConfig);
        }

        return split(file, slice, sliceSize);
    }

    /** 
     * @see com.alipay.rdf.file.interfaces.FileSplitter#getTailSlice(com.alipay.rdf.file.model.FileConfig)
     */
    @Override
    public FileSlice getTailSlice(FileConfig fileConfig) {
        String filePath = fileConfig.getFilePath();
        if (RdfFileLogUtil.common.isInfo()) {
            RdfFileLogUtil.common
                .info("rdf-file#NasFileSliceSplitter.getTailSlice filePath=" + filePath);
        }

        File file = new File(filePath);
        if (!file.exists()) {
            throw new RdfFileException(
                "rdf-file#NasFileSliceSplitter.getTailSlice  path=" + filePath + " 不存在文件",
                RdfErrorEnum.NOT_EXSIT);
        }

        long length = file.length();
        //文件内容为空
        if (length == 0) {
            if (RdfFileLogUtil.common.isInfo()) {
                RdfFileLogUtil.common.info(
                    "rdf-file#NasFileSliceSplitter.getTailSlice filePath=" + filePath + ", size=0");
            }
            return new FileSlice(filePath, FileDataTypeEnum.TAIL, 0, 0);
        } else if (RdfFileLogUtil.common.isDebug()) {
            RdfFileLogUtil.common
                .debug("rdf-file#NasFileSliceSplitter.getTailSlice file size=" + length);
        }

        if (!fileConfig.isPartial() && (fileConfig.getFileDataType() == FileDataTypeEnum.BODY
                                        || fileConfig.getFileDataType() == FileDataTypeEnum.HEAD)) {
            return new FileSlice(fileConfig.getFilePath(), FileDataTypeEnum.TAIL, length, length);
        }

        FileMeta fileMeta = TemplateLoader.load(fileConfig.getTemplatePath(),
            fileConfig.getTemplateEncoding());
        if (!fileMeta.hasTail()) {
            throw new RdfFileException(
                "rdf-file#文件模板template=" + fileConfig.getTemplatePath() + ", 没有定义尾",
                RdfErrorEnum.TAIL_NOT_DEFINED);
        }

        int tailRowsAffected = ProtocolLoader.getRowsAfftected(fileConfig, FileDataTypeEnum.TAIL);
        if (tailRowsAffected == 0) {
            if (RdfFileLogUtil.common.isInfo()) {
                RdfFileLogUtil.common
                    .info("rdf-file#NasFileSliceSplitter.getTailSlice tailRowsAfftected="

                          + tailRowsAffected);
            }
            return new FileSlice(filePath, FileDataTypeEnum.TAIL, length, length);
        } else if (RdfFileLogUtil.common.isDebug()) {
            RdfFileLogUtil.common.debug(
                "rdf-file#NasFileSliceSplitter.getTailSlice tailRowsAfftected=" + tailRowsAffected);
        }

        RandomAccessFile raf = null;
        int tailLineCount = 0;
        int count = 0;
        long p = length - 1;
        boolean next = true;

        try {
            raf = new RandomAccessFile(file, "r");

            //文件尾没有以换行符结束
            raf.seek(p);
            int last = raf.read();
            if (last != '\n' && last != '\r') {
                tailLineCount++;
                if (RdfFileLogUtil.common.isDebug()) {
                    RdfFileLogUtil.common.debug(
                        "rdf-file#NasFileSliceSplitter.getTailSlice file is not end with \\n or \\r");
                }
            }

            while (next) {
                count++;
                raf.seek(p--);
                int b = raf.read();
                switch (b) {
                    case '\n':
                    case '\r':
                        if (tailLineCount == tailRowsAffected) {
                            count--;
                            next = false;
                            break;
                        }

                        if (p < 0) {
                            next = false;
                            if (RdfFileLogUtil.common.isDebug()) {
                                RdfFileLogUtil.common.debug(
                                    "rdf-file#NasFileSliceSplitter.getTailSlice p < 0, linebreak="
                                                            + (b == '\r' ? "\\r" : "\\n"));
                            }
                            break;
                        }

                        raf.seek(p);
                        int pre = raf.read();
                        if (pre != '\r' && pre != '\n') {
                            tailLineCount++;
                            if (RdfFileLogUtil.common.isDebug()) {
                                RdfFileLogUtil.common.debug(
                                    "rdf-file#NasFileSliceSplitter.getTailSlice tailLineCount++, linebreak="
                                                            + (b == '\r' ? "\\r" : "\\n"));
                            }
                        }

                        break;
                    case -1:
                        count--;
                        next = false;
                        if (RdfFileLogUtil.common.isWarn()) {
                            RdfFileLogUtil.common
                                .warn("rdf-file#NasFileSliceSplitter.getTailSlice filePath="
                                      + filePath + " ,char count=" + count + ", read() == -1");
                        }
                        break;
                    default:
                        break;
                }

                if (p < 0) {
                    next = false;
                    if (RdfFileLogUtil.common.isDebug()) {
                        RdfFileLogUtil.common
                            .debug("rdf-file#NasFileSliceSplitter.getTailSlice normal p < 0");
                    }
                }
            }
        } catch (FileNotFoundException e) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.getTailSlice file="
                                       + file.getAbsolutePath() + " 文件不存在",
                e, RdfErrorEnum.NOT_EXSIT);
        } catch (IOException e) {
            throw new RdfFileException("rdf-file#NasFileSliceSplitter.getTailSlice io 异常", e,
                RdfErrorEnum.NOT_EXSIT);
        } finally {
            if (null != raf) {
                try {
                    raf.close();
                } catch (IOException e) {
                    if (RdfFileLogUtil.common.isWarn()) {
                        RdfFileLogUtil.common
                            .warn("rdf-file#NasFileSliceSplitter.getTailSlice close 异常", e);
                    }
                }
            }
        }

        long start = length - count;
        if (RdfFileLogUtil.common.isInfo()) {
            RdfFileLogUtil.common
                .info("rdf-file#NasFileSliceSplitter.getTailSlice filePath=" + filePath
                      + " ,char count=" + count + ", start=" + start + ", length=" + length);
        }

        return new FileSlice(filePath, FileDataTypeEnum.TAIL, start, length);
    }
}
