完成DataFile功能

skywalking-storage-center/skywalking-storage/pom.xml

@@ -67,6 +67,11 @@
             <version>2.3.4</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.a.eye</groupId>
+            <artifactId>data-carrier</artifactId>
+            <version>1.0</version>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
@@ -159,5 +164,14 @@
         </plugins>
     </build>
 
-
+    <repositories>
+        <repository>
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+            <id>bintray-wu-sheng-DataCarrier</id>
+            <name>bintray</name>
+            <url>http://dl.bintray.com/wu-sheng/DataCarrier</url>
+        </repository>
+    </repositories>
 </project>

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java

@@ -10,4 +10,9 @@ public class Config {
 
         public static String DATA_FILE_INDEX_FILE_NAME = "data_file.index";
     }
+
+
+    public static class DataFile {
+        public static String BASE_PATH = "";
+    }
 }

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanData.java

+package com.a.eye.skywalking.storage.data;
+
+/**
+ * Created by xin on 2016/11/4.
+ */
+public interface SpanData {
+    long getTimestamp();
+
+    byte[] convertToByte();
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/DataOperator.java → skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataBuilder.java

 package com.a.eye.skywalking.storage.data;
 
 /**
- * Created by xin on 2016/11/1.
+ * Created by xin on 2016/11/4.
  */
-public class DataOperator {
-
+public class SpanDataBuilder {
 }

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/SpanDataWriterConsumer.java

+package com.a.eye.skywalking.storage.data;
+
+import com.a.eye.datacarrier.consumer.IConsumer;
+import com.a.eye.skywalking.storage.data.file.DataFileWriter;
+import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
+import com.a.eye.skywalking.storage.data.index.IndexOperator;
+import com.a.eye.skywalking.storage.data.index.IndexOperatorFactory;
+
+import java.util.List;
+
+public class SpanDataWriterConsumer implements IConsumer<SpanData> {
+
+    private DataFileWriter writer;
+
+    @Override
+    public void consume(List<SpanData> list) {
+        for (SpanData data : list) {
+            IndexOperator operator = IndexOperatorFactory.get(data.getTimestamp());
+            IndexMetaInfo metaInfo = writer.write(data.convertToByte());
+            operator.update(metaInfo);
+        }
+    }
+
+    @Override
+    public void onError(List<SpanData> list, Throwable throwable) {
+
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/DataFileNotFoundException.java

+package com.a.eye.skywalking.storage.data.exception;
+
+/**
+ * Created by xin on 2016/11/4.
+ */
+public class DataFileNotFoundException extends Exception {
+    public DataFileNotFoundException(String message, Exception e) {
+        super(message, e);
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/exception/FileReaderCreateFailedException.java

+package com.a.eye.skywalking.storage.data.exception;
+
+/**
+ * Created by xin on 2016/11/4.
+ */
+public class FileReaderCreateFailedException extends Throwable {
+    public FileReaderCreateFailedException(String message, Exception e) {
+        super(message, e);
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java

+package com.a.eye.skywalking.storage.data.file;
+
+public class DataFile {
+
+    private static final long MAX_LENGTH = 3 * 1024 * 1024 * 1024;
+
+    private String name;
+    private long   currentLength;
+
+    public boolean overLimitLength() {
+        return currentLength > MAX_LENGTH;
+    }
+
+
+    public long writeAndFlush(byte[] data) {
+
+        return 0;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void close() {
+
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileOperator.java

-package com.a.eye.skywalking.storage.data.file;
-
-/**
- * Created by xin on 2016/10/31.
- */
-public class DataFileOperator {
-
-    public int write(){
-
-        return 0;
-    }
-
-    public Object read(Object dataIndex) {
-        return null;
-    }
-}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileOperatorFactory.java

+package com.a.eye.skywalking.storage.data.file;
+
+import com.a.eye.skywalking.storage.data.exception.DataFileNotFoundException;
+import com.a.eye.skywalking.storage.data.exception.FileReaderCreateFailedException;
+import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
+
+public class DataFileOperatorFactory {
+
+    public static DataFileReader newReader(IndexMetaInfo indexMetaInfo) throws FileReaderCreateFailedException {
+        try {
+            return new DataFileReader(indexMetaInfo);
+        } catch (DataFileNotFoundException e) {
+            throw new FileReaderCreateFailedException("Cannot create DataFileReader.", e);
+        }
+    }
+
+    public static DataFileWriter newWriter() {
+        return new DataFileWriter();
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileReader.java

 package com.a.eye.skywalking.storage.data.file;
 
-/**
- * Created by xin on 2016/10/31.
- */
+import com.a.eye.skywalking.storage.config.Config;
+import com.a.eye.skywalking.storage.data.exception.DataFileNotFoundException;
+import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
 public class DataFileReader {
+
+    private static Logger logger = LogManager.getLogger(DataFileReader.class);
+
+    private long            offset;
+    private int             length;
+    private String          fileName;
+    private FileInputStream reader;
+
+    public DataFileReader(IndexMetaInfo indexMetaInfo) throws DataFileNotFoundException {
+        try {
+            reader = new FileInputStream(new File(Config.DataFile.BASE_PATH, indexMetaInfo.getFileName()));
+            this.offset = indexMetaInfo.getOffset();
+            this.length = indexMetaInfo.getLength();
+            this.fileName = indexMetaInfo.getFileName();
+        } catch (FileNotFoundException e) {
+            throw new DataFileNotFoundException(indexMetaInfo.getFileName() + " not found.", e);
+        }
+    }
+
+    public byte[] read() {
+        try {
+            reader.getChannel().position(this.offset);
+            byte[] dataByte = new byte[length];
+            reader.read(dataByte, 0, length);
+            return dataByte;
+        } catch (IOException e) {
+            logger.error("Failed to read file:{} position:{} length:{}", fileName, offset, length);
+            return null;
+        }
+    }
 }

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFileWriter.java

 package com.a.eye.skywalking.storage.data.file;
 
-/**
- * Created by xin on 2016/10/31.
- */
+import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
+
 public class DataFileWriter {
+
+    private DataFile dataFile;
+
+    public DataFileWriter() {
+
+    }
+
+    public IndexMetaInfo write(byte[] data) {
+        if (dataFile.overLimitLength()) {
+            convertDataFile();
+        }
+
+        long offset = dataFile.writeAndFlush(data);
+        return new IndexMetaInfo(dataFile.getName(), offset, data.length);
+    }
+
+    private void convertDataFile() {
+        dataFile.close();
+        dataFile = new DataFile();
+    }
+
 }

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/DataIndexFile.java

-package com.a.eye.skywalking.storage.data.index;
-
-/**
- * Created by xin on 2016/10/31.
- */
-public class DataIndexFile {
-
-    public void read(String traceId) {
-    }
-
-    public void write() {
-
-    }
-}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/DataIndexFileWriter.java

-package com.a.eye.skywalking.storage.data.index;
-
-/**
- * Created by xin on 2016/10/31.
- */
-public class DataIndexFileWriter {
-}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexMetaInfo.java

+package com.a.eye.skywalking.storage.data.index;
+
+/**
+ * Created by xin on 2016/11/3.
+ */
+public class IndexMetaInfo {
+
+    private String fileName;
+
+    private long offset;
+
+    private int length;
+
+    public IndexMetaInfo(String name, long offset, int length) {
+        this.fileName = name;
+        this.offset = offset;
+        this.length = length;
+    }
+
+
+    public String getFileName() {
+        return fileName;
+    }
+
+    public long getOffset() {
+        return offset;
+    }
+
+    public int getLength() {
+        return length;
+    }
+}

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/DataIndexFileOperator.java → skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperator.java

 package com.a.eye.skywalking.storage.data.index;
 
+import java.util.List;
+
 /**
- * Created by xin on 2016/10/31.
+ * Created by xin on 2016/11/3.
  */
-public class DataIndexFileOperator {
+public class IndexOperator {
 
-    public Object read(String traceId) {
+    public List<IndexMetaInfo> find(String traceId) {
         return null;
     }
 
-    public void write(Object o, int offset) {
+    public void update(IndexMetaInfo meta) {
 
     }
-
 }

skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/DataIndexFileReader.java → skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/index/IndexOperatorFactory.java

 package com.a.eye.skywalking.storage.data.index;
 
-/**
- * Created by xin on 2016/10/31.
- */
-public class DataIndexFileReader {
+public class IndexOperatorFactory {
+
+    public static IndexOperator get(long timestamp) {
+        return null;
+    }
 }