补充完整逻辑

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

@@ -20,6 +20,11 @@ public class Config {
 
 
     public static class DataIndex {
+
         public static String TABLE_NAME = "data_index";
+
+        public static String BASE_PATH = "";
+
+        public static String STORAGE_INDEX_FILE_NAME = "";
     }
 }

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

@@ -0,0 +1,25 @@
+package com.a.eye.skywalking.storage.config;
+
+import static com.a.eye.skywalking.storage.config.Config.DataIndex.TABLE_NAME;
+
+public class Constants {
+
+    public static class SQL {
+        public static final String CREATE_TABLE = "CREATE TABLE " + TABLE_NAME + "\n" + "(\n"
+                + "    id INT PRIMARY KEY NOT NULL IDENTITY,\n"
+                + "    trace_id VARCHAR(32) NOT NULL,\n"
+                + "    parent_level_id VARCHAR(1024) NOT NULL,\n"
+                + "    level_id INT NOT NULL,\n"
+                + "    file_name VARCHAR(10) NOT NULL,\n"
+                + "    offset BIGINT NOT NULL,\n"
+                + "    length INT NOT NULL\n" + ");\n";
+
+        public static final String CREATE_INDEX = "CREATE INDEX \"index_data_trace_id_index\" ON " + TABLE_NAME + " (trace_id);";
+
+        public static final String INSERT_INDEX = "INSERT INTO " +TABLE_NAME + "(trace_id,parent_level_id,level_id,"
+                + "file_name,offset,length) VALUES(?,?,?,?,?,?)";
+
+        public static final String QUERY_TABLES = "SELECT count(1) AS TABLE_COUNT FROM   INFORMATION_SCHEMA.TABLES "
+                + "WHERE TABLE_NAME= '" + TABLE_NAME + "';";
+    }
+}

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

@@ -15,6 +15,11 @@ public class SpanDataConsumer implements IConsumer<SpanData> {
     private IndexDBConnectorCache cache;
     private DataFileWriter        fileWriter;
 
+    public SpanDataConsumer(){
+        cache = new IndexDBConnectorCache();
+        fileWriter = new DataFileWriter();
+    }
+
     @Override
     public void consume(List<SpanData> data) {
 

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

@@ -0,0 +1,10 @@
+package com.a.eye.skywalking.storage.data.exception;
+
+/**
+ * Created by xin on 2016/11/5.
+ */
+public class ConnectorInitializeFailedException extends RuntimeException {
+    public ConnectorInitializeFailedException(String message, Exception e) {
+        super(message, e);
+    }
+}

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

@@ -0,0 +1,10 @@
+package com.a.eye.skywalking.storage.data.exception;
+
+/**
+ * Created by xin on 2016/11/5.
+ */
+public class DataFileOperatorCreateFailedException extends RuntimeException {
+    public DataFileOperatorCreateFailedException(String message, Exception e){
+        super(message, e);
+    }
+}

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

@@ -0,0 +1,7 @@
+package com.a.eye.skywalking.storage.data.exception;
+
+public class IndexMetaPersistenceFailedException extends RuntimeException {
+    public IndexMetaPersistenceFailedException(String message, Exception e){
+        super(message, e);
+    }
+}

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

@@ -0,0 +1,9 @@
+package com.a.eye.skywalking.storage.data.exception;
+
+/**
+ * Created by xin on 2016/11/4.
+ */
+public class SpanDataPersistenceFailedException extends RuntimeException {
+    public SpanDataPersistenceFailedException(Exception e) {
+    }
+}

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

@@ -1,10 +1,94 @@
 package com.a.eye.skywalking.storage.data.file;
 
+import com.a.eye.skywalking.storage.config.Config;
+import com.a.eye.skywalking.storage.data.SpanData;
+import com.a.eye.skywalking.storage.data.exception.DataFileOperatorCreateFailedException;
+import com.a.eye.skywalking.storage.data.exception.SpanDataPersistenceFailedException;
+import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+
 /**
  * 数据文件
  */
 public class DataFile {
+
+    private String           fileName;
+    private long             currentOffset;
+    private DataFileOperator operator;
+
+    public DataFile() {
+        this.fileName = System.currentTimeMillis() + "";
+        this.currentOffset = 0;
+        operator = new DataFileOperator();
+    }
+
+    public DataFile(String fileName, long offset) {
+        this.fileName = fileName;
+        this.currentOffset = offset;
+        operator = new DataFileOperator();
+    }
+
+    public DataFile(File file) {
+        this.fileName = file.getName();
+        this.currentOffset = file.length();
+        operator = new DataFileOperator();
+    }
+
     public boolean overLimitLength() {
-        return false;
+        return currentOffset >= Config.DataFile.MAX_LENGTH;
+    }
+
+    public IndexMetaInfo write(SpanData data) {
+        byte[] bytes = data.toByteArray();
+        try {
+            operator.getWriter().write(bytes);
+            IndexMetaInfo metaInfo = new IndexMetaInfo(fileName, currentOffset, bytes.length);
+            currentOffset += bytes.length;
+            return metaInfo;
+        } catch (IOException e) {
+            throw new SpanDataPersistenceFailedException(e);
+        }
+    }
+
+    public void flush() {
+        try {
+            operator.getWriter().flush();
+        } catch (IOException e) {
+            throw new SpanDataPersistenceFailedException(e);
+        }
+    }
+
+    class DataFileOperator {
+        private FileOutputStream writer;
+        private FileInputStream  reader;
+
+        public FileOutputStream getWriter() {
+
+            if (writer == null) {
+                try {
+                    writer = new FileOutputStream(new File(fileName));
+                } catch (IOException e) {
+                    throw new DataFileOperatorCreateFailedException("Failed to create datafile output stream", e);
+                }
+            }
+
+            return writer;
+        }
+
+        public FileInputStream getReader() {
+            if (reader == null) {
+                try {
+                    reader = new FileInputStream(new File(fileName));
+                } catch (IOException e) {
+                    throw new DataFileOperatorCreateFailedException("Failed to create datafile input stream", e);
+                }
+            }
+
+            return reader;
+        }
     }
 }

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

@@ -1,16 +1,24 @@
 package com.a.eye.skywalking.storage.data.file;
 
+import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
 public class DataFileLoader {
 
-    public DataFileLoader(String basePath) {
+    private String basePath;
 
+    public DataFileLoader(String basePath) {
+        this.basePath = basePath;
     }
 
     public List<DataFile> load() {
-        return new ArrayList<DataFile>();
+        File dataFileDir = new File(basePath);
+        List<DataFile> allDataFile = new ArrayList<DataFile>();
+        for (File fileEntry : dataFileDir.listFiles()) {
+            allDataFile.add(new DataFile(fileEntry));
+        }
+        return allDataFile;
     }
 
 

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

@@ -1,10 +0,0 @@
-package com.a.eye.skywalking.storage.data.file;
-
-import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
-
-
-public class DataFileOperatorFactory {
-    public static DataFileReader getDataFileReader(IndexMetaInfo info) {
-        return new DataFileReader(info);
-    }
-}

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

@@ -1,17 +0,0 @@
-package com.a.eye.skywalking.storage.data.file;
-
-import com.a.eye.skywalking.storage.data.index.IndexMetaInfo;
-
-/**
- * Created by xin on 2016/11/4.
- */
-public class DataFileReader {
-
-    public DataFileReader(IndexMetaInfo info) {
-
-    }
-
-    public byte[] read() {
-        return new byte[0];
-    }
-}

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

@@ -14,11 +14,16 @@ public class DataFileWriter {
     }
 
     public IndexMetaCollections write(List<SpanData> spanData) {
-
         if (dataFile.overLimitLength()) {
             dataFile = DataFilesManager.createNewDataFile();
         }
 
-        return null;
+        IndexMetaCollections collections = new IndexMetaCollections();
+        for (SpanData data : spanData) {
+            collections.add(dataFile.write(data));
+        }
+        dataFile.flush();
+
+        return collections;
     }
 }

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

@@ -11,10 +11,16 @@ public class UnFinishedDataFilePicker {
     private List<DataFile> dataFiles;
 
     public UnFinishedDataFilePicker(List<DataFile> dataFiles) {
-
+        this.dataFiles = dataFiles;
     }
 
     public List<DataFile> pickUp() {
-        return new ArrayList<DataFile>();
+        List<DataFile> result = new ArrayList<DataFile>();
+        for (DataFile file : dataFiles) {
+            if (file.overLimitLength()){
+                result.add(file);
+            }
+        }
+        return result;
     }
 }

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

@@ -1,30 +1,123 @@
 package com.a.eye.skywalking.storage.data.index;
 
+import com.a.eye.skywalking.storage.config.Config;
+import com.a.eye.skywalking.storage.data.exception.ConnectorInitializeFailedException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.sql.*;
+
+import static com.a.eye.skywalking.storage.config.Constants.SQL.*;
+
 /**
  * Created by xin on 2016/11/4.
  */
 public class IndexDBConnector {
 
-    private long timestamp;
+    private static final int MAX_BATCH_SIZE = 20;
+
+    private static Logger logger = LogManager.getLogger(IndexDBConnector.class);
+
+    static {
+        try {
+            Class.forName("org.hsqldb.jdbc.JDBCDriver");
+        } catch (ClassNotFoundException e) {
+            //never
+        }
+    }
+
+    private long       timestamp;
+    private Connection connection;
+    private ConnectURLGenerator generator =
+            new ConnectURLGenerator(Config.DataIndex.BASE_PATH, Config.DataIndex.STORAGE_INDEX_FILE_NAME);
 
     public IndexDBConnector(long timestamp) {
-
+        this.timestamp = timestamp;
+        createConnection();
+        createTableAndIndexIfNecessary();
     }
 
-    private void validate() {
-
+    private void createTableAndIndexIfNecessary() {
+        try {
+            if (validateTableIsExists()) {
+                createTable();
+                createIndex();
+            }
+        } catch (SQLException e) {
+            throw new ConnectorInitializeFailedException("Failed to create table and index.", e);
+        }
     }
 
-    private void createTable() {
 
+    private void createConnection() {
+        try {
+            connection = DriverManager.getConnection(generator.generate(timestamp));
+            connection.setAutoCommit(true);
+        } catch (SQLException e) {
+            throw new ConnectorInitializeFailedException("Failed to create connection.", e);
+        }
     }
 
-    private void createIndex() {
+    private boolean validateTableIsExists() throws SQLException {
+        PreparedStatement ps = connection.prepareStatement(QUERY_TABLES);
+        ResultSet rs = ps.executeQuery();
+        rs.next();
 
+        boolean exists =  rs.getInt("TABLE_COUNT") == 1;
+        rs.close();
+        ps.close();
+
+        return exists;
     }
 
+    private void createTable() throws SQLException {
+        PreparedStatement ps = connection.prepareStatement(CREATE_TABLE);
+        ps.execute();
+        ps.close();
+    }
+
+    private void createIndex() throws SQLException {
+        PreparedStatement ps = connection.prepareStatement(CREATE_INDEX);
+        ps.execute();
+        ps.close();
+    }
 
     public long getTimestamp() {
         return timestamp;
     }
+
+    public void batchUpdate(IndexMetaGroup metaGroup) throws SQLException {
+        int currentIndex = 0;
+        PreparedStatement ps = connection.prepareStatement(INSERT_INDEX);
+        for (IndexMetaInfo metaInfo : metaGroup.getMetaInfo()) {
+            ps.setString(1, metaInfo.getTraceId());
+            ps.setString(2, metaInfo.getParentLevelId());
+            ps.setInt(3, metaInfo.getLevelId());
+            ps.setString(4, metaInfo.getFileName());
+            ps.setLong(5, metaInfo.getOffset());
+            ps.setInt(6, metaInfo.getLength());
+            ps.addBatch();
+            if (++currentIndex > MAX_BATCH_SIZE) {
+                ps.executeBatch();
+            }
+        }
+        ps.executeBatch();
+        ps.close();
+    }
+
+    class ConnectURLGenerator {
+
+        private String basePath;
+        private String dbFileName;
+
+        private ConnectURLGenerator(String basePath, String dbFileName) {
+            this.basePath = basePath;
+            this.dbFileName = dbFileName;
+        }
+
+
+        public String generate(long timestamp) {
+            return "jdbc:hsqldb:file:" + basePath + "/" + timestamp + "/" + dbFileName;
+        }
+    }
 }

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

@@ -17,10 +17,15 @@ public class IndexDBConnectorCache {
     }
 
     public IndexDBConnector get(long timestamp) {
-        return cachedOperators.get(timestamp);
+        IndexDBConnector connector = cachedOperators.get(timestamp);
+        if (connector == null) {
+            connector = new IndexDBConnector(timestamp);
+            updateCache(timestamp, connector);
+        }
+        return connector;
     }
 
-    public void updateCache(long timestamp, IndexDBConnector operator) {
+    private void updateCache(long timestamp, IndexDBConnector operator) {
         cachedOperators.put(timestamp, operator);
     }
 

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

@@ -11,7 +11,12 @@ import java.util.List;
 public class IndexMetaCollections {
 
     private List<IndexMetaInfo> metaInfo;
-    private BlockFinder finder = BlockIndexEngine.newFinder();
+    private BlockFinder         finder;
+
+    public IndexMetaCollections() {
+        metaInfo = new ArrayList<>();
+        finder = BlockIndexEngine.newFinder();
+    }
 
     public Iterator<IndexMetaGroup> group() {
         List<IndexMetaGroup> indexMetaGroups = new ArrayList<IndexMetaGroup>();
@@ -35,4 +40,7 @@ public class IndexMetaCollections {
     }
 
 
+    public void add(IndexMetaInfo info) {
+        metaInfo.add(info);
+    }
 }

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

@@ -1,6 +1,7 @@
 package com.a.eye.skywalking.storage.data.index;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 
 /**

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

@@ -11,6 +11,12 @@ public class IndexMetaInfo {
 
     private long startTime;
 
+    public IndexMetaInfo(String fileName, long offset, int length) {
+        this.fileName = fileName;
+        this.offset = offset;
+        this.length = length;
+    }
+
     public String getFileName() {
         return fileName;
     }
@@ -26,4 +32,16 @@ public class IndexMetaInfo {
     public long getStartTime() {
         return startTime;
     }
+
+    public String getTraceId() {
+        return null;
+    }
+
+    public String getParentLevelId() {
+        return null;
+    }
+
+    public int getLevelId() {
+        return 0;
+    }
 }

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

@@ -1,5 +1,7 @@
 package com.a.eye.skywalking.storage.data.index;
 
+import com.a.eye.skywalking.storage.data.exception.IndexMetaPersistenceFailedException;
+
 import java.util.ArrayList;
 import java.util.List;
 
@@ -19,7 +21,11 @@ public class IndexOperator {
 
 
     public void batchUpdate(IndexMetaGroup metaGroup) {
-
+        try {
+            connector.batchUpdate(metaGroup);
+        } catch (Exception e) {
+            throw new IndexMetaPersistenceFailedException("Failed to batch save index meta", e);
+        }
     }
 
     public static IndexOperator newOperator(IndexDBConnector indexDBConnector) {

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

@@ -1,27 +0,0 @@
-package com.a.eye.skywalking.storage.data.index;
-
-import java.sql.Connection;
-
-public class IndexOperatorHelper {
-
-    public IndexOperatorHelper(Connection connection) {
-        
-    }
-
-    public boolean validateIsReady(String tableName) {
-        return false;
-    }
-
-    public void maintain() {
-        createTable();
-        createIndex();
-    }
-
-    private void createIndex() {
-
-    }
-
-    private void createTable() {
-
-    }
-}