Merge pull request #69 from iminright/linebreak

quhongwei · web-flow · commit 75a2220a90cf · 2021-09-30T11:52:11.000+08:00
Linebreak
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/common/ProtocolFileWriter.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/common/ProtocolFileWriter.java
@@ -38,6 +38,8 @@ public class ProtocolFileWriter implements RdfFileWriterSpi {
     private Map<ProcessorTypeEnum, List<RdfFileProcessorSpi>> processors;
     private RdfBufferedWriter                                 writer = null;
     private Summary                                           summary;
+    // 该writer是否已经写过数据
+    private boolean                                           hasWritten = false;
 
     @Override
     public void init(FileConfig fileConfig) {
@@ -129,7 +131,11 @@ public void writeLine(String line) {
         RdfFileUtil.assertNotNull(line, "ProtocolFileWriter.writeLine(line == null)",
             RdfErrorEnum.ILLEGAL_ARGUMENT);
         ensureOpen();
-        writer.write(line + RdfFileUtil.getLineBreak(fileConfig));
+        String lineWithLB = RdfFileUtil.processLineBreak(line, RdfFileUtil.getLineBreak(fileConfig)
+                , fileConfig.isAppendLinebreakAtLast() && fileMeta.isAppendLinebreakAtLast()
+                , !hasWritten);
+        writer.write(lineWithLB);
+        hasWritten = true;
     }
 
     @Override
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/common/RawFileWriter.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/common/RawFileWriter.java
@@ -24,6 +24,8 @@ public class RawFileWriter implements RdfFileWriterSpi {
     private RdfBufferedWriter                                 writer = null;
     private FileConfig                                        fileConfig;
     private Map<ProcessorTypeEnum, List<RdfFileProcessorSpi>> processors;
+    // 该writer是否已经写过数据
+    private boolean                                           hasWritten = false;
 
     @Override
     public void init(FileConfig fileConfig) {
@@ -52,7 +54,12 @@ public void writeTail(Object tailBean) {
     @Override
     public void writeLine(String line) {
         ensureOpen();
-        writer.write(line + RdfFileUtil.getLineBreak(fileConfig));
+        String lineWithLB = RdfFileUtil.processLineBreak(line
+                , RdfFileUtil.getLineBreak(fileConfig)
+                , fileConfig.isAppendLinebreakAtLast()
+                , !hasWritten);
+        writer.write(lineWithLB);
+        hasWritten = true;
     }
 
     @Override
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/loader/TemplateLoader.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/loader/TemplateLoader.java
@@ -204,6 +204,7 @@ public static FileMeta load(String templatePath, String templateEncoding) {
             fileMeta.setColumnSplit(templateConfig.getColumnSplit());
             fileMeta.setFileEncoding(templateConfig.getFileEncoding());
             fileMeta.setLineBreak(templateConfig.getLineBreak());
+            fileMeta.setAppendLinebreakAtLast(Boolean.valueOf(templateConfig.getIsAppendLinebreakAtLast()));
             fileMeta.setRelationReadRowCompatibility(templateConfig.getRelationReadRowCompatibility());
             if (RdfFileUtil.isNotBlank(templateConfig.getRowCodecMode())) {
                 fileMeta.setRowCodecMode(templateConfig.getRowCodecMode());
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/meta/FileMeta.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/meta/FileMeta.java
@@ -53,6 +53,8 @@ public class FileMeta {
     private Map<FileDataTypeEnum, Boolean> endWithSplit          = new HashMap<FileDataTypeEnum, Boolean>();
     /**body是否是多模板配置*/
     private boolean                        multiBody             = false;
+    /**文件尾是否需要换行*/
+    private boolean                        isAppendLinebreakAtLast = true;
     /**定义或者覆盖协议文件的字段类型*/
     private final List<FileColumnMeta>     protocolDataTypes     = new ArrayList<FileColumnMeta>();
     /**关系模式读行数据兼容模式*/
@@ -434,4 +436,12 @@ public String toString() {
         sb.append("]");
         return sb.toString();
     }
+
+    public boolean isAppendLinebreakAtLast() {
+        return isAppendLinebreakAtLast;
+    }
+
+    public void setAppendLinebreakAtLast(boolean appendLinebreakAtLast) {
+        isAppendLinebreakAtLast = appendLinebreakAtLast;
+    }
 }
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/meta/TemplateConfig.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/meta/TemplateConfig.java
@@ -22,6 +22,8 @@ public class TemplateConfig {
     private String                columnSplit          = "|";
     /** 换行符*/
     private String                lineBreak            = "\r\n";
+    /** 文件尾是否需要换行 */
+    private String                isAppendLinebreakAtLast      = "true";
     /** 文件以分隔符开始   head|body|tail*/
     private String                startWithSplit;
     /** 文件以分隔符结束   head|body|tail*/
@@ -153,6 +155,14 @@ public void setStatisticColumnPairs(List<String> statisticColumnPairs) {
         this.statisticColumnPairs = statisticColumnPairs;
     }
 
+    public String getIsAppendLinebreakAtLast() {
+        return isAppendLinebreakAtLast;
+    }
+
+    public void setIsAppendLinebreakAtLast(String isAppendLinebreakAtLast) {
+        this.isAppendLinebreakAtLast = isAppendLinebreakAtLast;
+    }
+  
     public List<String> getProtocolDataType() {
         return protocolDataType;
     }
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/model/FileConfig.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/model/FileConfig.java
@@ -62,6 +62,8 @@ public class FileConfig implements Cloneable {
     private Boolean             relationReadRowCompatibility;
     /** 写文件的时候是否在文件尾部追加*/
     private boolean             isAppend         = false;
+    /**文件尾是否需要换行*/
+    private boolean             isAppendLinebreakAtLast = true;
     /** 外部构建的输入流*/
     private InputStream         is;
     /** 透传给插件的参数*/
@@ -86,7 +88,6 @@ public FileConfig(String filePath, String templatePath, StorageConfig storageCon
 
     /**
      * 构造方法
-     *
      * @param templatePath
      * @param storageConfig
      */
@@ -99,7 +100,6 @@ public FileConfig(String templatePath, StorageConfig storageConfig) {
 
     /**
      * 读数据时， 输入流由外界构造
-     *
      * @param is
      * @param templatePath
      */
@@ -325,6 +325,14 @@ public void setCreateEmptyFile(boolean createEmptyFile) {
         this.createEmptyFile = createEmptyFile;
     }
 
+    public boolean isAppendLinebreakAtLast() {
+        return isAppendLinebreakAtLast;
+    }
+
+    public void setAppendLinebreakAtLast(boolean appendLinebreakAtLast) {
+        isAppendLinebreakAtLast = appendLinebreakAtLast;
+    }
+  
     public boolean isReadAll() {
         return isReadAll;
     }
@@ -376,6 +384,7 @@ public FileConfig clone() {
         config.setRowValidators(rowValidators);
         config.setInputStream(is);
         config.setColumnSplit(columnSplit);
+        config.setAppendLinebreakAtLast(isAppendLinebreakAtLast);
         config.params = params;
         for(String processKey : processorKeys){
             config.addProcessorKey(processKey);
@@ -428,6 +437,7 @@ public String toString() {
             sb.append(",length=").append(length);
         }
         sb.append(",isAppend=").append(isAppend);
+        sb.append(",isAppendLinebreakAtLast=").append(isAppendLinebreakAtLast);
         sb.append(",isReadAll=").append(isReadAll);
         if (null != relationReadRowCompatibility) {
             sb.append(",relationReadRowCompatibility=").append(relationReadRowCompatibility);
diff --git a/rdf-file-core/src/main/java/com/alipay/rdf/file/util/RdfFileUtil.java b/rdf-file-core/src/main/java/com/alipay/rdf/file/util/RdfFileUtil.java
@@ -823,6 +823,24 @@ public static Map<String, String> parsePathParams(String path) {
         return params;
     }
 
+    /**
+     * 为originalLine处理LineBreak的逻辑
+     * @param originalLine  写入的数据的原始版本
+     * @param linebreak  换行符
+     * @param isAppendLBAtLast  是否要在文件末尾写入LineBreak
+     * @param isFirstLine  当前要写入的行是否是该文件的第一行
+     * @return
+     */
+    public static String processLineBreak(String originalLine, String linebreak, boolean isAppendLBAtLast, boolean isFirstLine){
+        if(isAppendLBAtLast){
+            // 如果需要在文件末尾添加LB则始终将LB追加在要写入的行后
+            return originalLine + linebreak;
+        }else {
+            // 否则将LB追加在要写入的行前，除了文件的第一行
+            return isFirstLine ? originalLine : linebreak + originalLine;
+        }
+    }
+
     /**
      * 从输入流读取指定长度数据到byte[]
      *
diff --git a/rdf-file-core/src/test/java/com/alipay/rdf/file/AbstractFileTestCase.java b/rdf-file-core/src/test/java/com/alipay/rdf/file/AbstractFileTestCase.java
@@ -0,0 +1,64 @@
+/*
+ * Alipay.com Inc.
+ * Copyright (c) 2004-2021 All Rights Reserved.
+ */
+package com.alipay.rdf.file;
+
+import com.alipay.rdf.file.model.FileDefaultConfig;
+import com.alipay.rdf.file.util.TemporaryFolderUtil;
+import com.alipay.rdf.file.util.TestLog;
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+
+/**
+ * @author wanhaofan
+ * @version TestUtil.java, v 0.1 2021年09月26日 2:08 PM wanhaofan
+ */
+public class AbstractFileTestCase {
+
+    public TemporaryFolderUtil tf = new TemporaryFolderUtil();
+
+    @Before
+    public void setUp() throws IOException {
+        tf.create();
+        new FileDefaultConfig().setCommonLog(new TestLog());
+    }
+
+    public final int getLastByteOfFile(String filePath) throws IOException {
+        FileInputStream fis = null;
+
+        try {
+            fis = new FileInputStream(filePath);
+            int prevByte = -1;
+            int currentByte = fis.read();
+            while(currentByte != -1){
+                prevByte = currentByte;
+                currentByte = fis.read();
+            }
+            return prevByte;
+        } finally {
+            if(fis != null){
+                try {
+                    fis.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    public final String createLocalFile(String fileName){
+        String filePath = tf.getRoot().getAbsolutePath();
+        return new File(filePath, fileName).getAbsolutePath();
+    }
+
+    @After
+    public void after() {
+        tf.delete();
+    }
+
+}
diff --git a/rdf-file-core/src/test/java/com/alipay/rdf/file/writer/LineBreakTest.java b/rdf-file-core/src/test/java/com/alipay/rdf/file/writer/LineBreakTest.java
diff --git a/rdf-file-core/src/test/resources/writer/template/template5.json b/rdf-file-core/src/test/resources/writer/template/template5.json
diff --git a/rdf-file-core/src/test/resources/writer/template/template6.json b/rdf-file-core/src/test/resources/writer/template/template6.json
diff --git a/rdf-file-oss/pom.xml b/rdf-file-oss/pom.xml
