package io.teknek.nibiru.engine;

import io.teknek.nibiru.Configuration;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentSkipListMap;

public class SSTableWriter {

  public void flushToDisk(String id, Configuration conf, Memtable m) throws IOException{
    File sstableFile = new File(conf.getSstableDirectory(), id + ".ss");
    CountingBufferedOutputStream ssOutputStream = null;
    IndexWriter indexWriter = new IndexWriter(id, conf);
    try {
      ssOutputStream = new CountingBufferedOutputStream(new FileOutputStream(sstableFile));
      indexWriter.open();
      for (Entry<Token, ConcurrentSkipListMap<String, Val>> i : m.getData().entrySet()){
        long startOfRecord = ssOutputStream.getWrittenOffset();
        ssOutputStream.writeAndCount(SsTableReader.START_RECORD);
        ssOutputStream.writeAndCount(i.getKey().getToken().getBytes());
        ssOutputStream.writeAndCount(SsTableReader.END_TOKEN);
        ssOutputStream.writeAndCount(i.getKey().getRowkey().getBytes());
        ssOutputStream.writeAndCount(SsTableReader.END_ROWKEY);
        indexWriter.handleRow(startOfRecord, i.getKey().getToken());
        boolean writeJoin = false;
        for (Entry<String, Val> j : i.getValue().entrySet()){
          if (!writeJoin){
            writeJoin = true;
          } else {
            ssOutputStream.writeAndCount(SsTableReader.END_COLUMN);
          }
          ssOutputStream.writeAndCount(j.getKey().getBytes());
          ssOutputStream.writeAndCount(SsTableReader.END_COLUMN_PART);
          ssOutputStream.writeAndCount(String.valueOf(j.getValue().getCreateTime()).getBytes());
          ssOutputStream.writeAndCount(SsTableReader.END_COLUMN_PART);
          ssOutputStream.writeAndCount(String.valueOf(j.getValue().getTime()).getBytes());
          ssOutputStream.writeAndCount(SsTableReader.END_COLUMN_PART);
          ssOutputStream.writeAndCount(String.valueOf(j.getValue().getTtl()).getBytes());
          ssOutputStream.writeAndCount(SsTableReader.END_COLUMN_PART);
          ssOutputStream.writeAndCount(String.valueOf(j.getValue().getValue()).getBytes());
        }
        ssOutputStream.writeAndCount('\n');
      }
    }
    finally {
      ssOutputStream.close();
      indexWriter.close();
    }
  }
  
  
}

@@ -71,62 +71,8 @@
 n;%0A%0A
-import java.io.File;%0Aimport java.io.FileOutputStream;%0A
 impo
@@ -296,16 +296,254 @@
 tion%7B%0A  
+  SsTableStreamWriter w = new SsTableStreamWriter(id, conf);%0A    w.open();%0A    for (Entry%3CToken, ConcurrentSkipListMap%3CString, Val%3E%3E i : m.getData().entrySet())%7B%0A      w.write(i.getKey(), i.getValue());%0A    %7D%0A    w.close();%0A  %7D%0A%7D%0A  %0A  /*%0A
   File s
@@ -605,18 +605,16 @@
 %22.ss%22);%0A
-  
   Counti
@@ -661,18 +661,16 @@
 null;%0A  
- 
- 
 IndexWri
@@ -720,18 +720,14 @@
 ;%0A  
- 
- 
 try %7B%0A
-  
     
@@ -808,26 +808,24 @@
 ableFile));%0A
-  
     indexWri
@@ -836,26 +836,24 @@
 open();%0A    
-  
 for (Entry%3CT
@@ -929,18 +929,16 @@
 %7B%0A      
-  
 long sta
@@ -983,34 +983,32 @@
 Offset();%0A      
- 
- 
 ssOutputStream.w
@@ -1049,18 +1049,16 @@
 ECORD);%0A
-  
       ss
@@ -1111,34 +1111,32 @@
 ().getBytes());%0A
-  
       ssOutputSt
@@ -1178,34 +1178,32 @@
 D_TOKEN);%0A      
-  
 ssOutputStream.w
@@ -1253,26 +1253,24 @@
 s());%0A      
-  
 ssOutputStre
@@ -1315,26 +1315,24 @@
 KEY);%0A      
- 
- 
 indexWriter.
@@ -1386,18 +1386,16 @@
 ;%0A      
-  
 boolean 
@@ -1413,18 +1413,16 @@
  false;%0A
-  
       fo
@@ -1481,18 +1481,16 @@
         
-  
 if (!wri
@@ -1498,18 +1498,16 @@
 eJoin)%7B%0A
-  
         
@@ -1534,18 +1534,16 @@
         
- 
- 
 %7D else %7B
@@ -1543,18 +1543,16 @@
  else %7B%0A
-  
         
@@ -1605,26 +1605,24 @@
 ND_COLUMN);%0A
-  
         %7D%0A  
@@ -1619,34 +1619,32 @@
       %7D%0A        
-  
 ssOutputStream.w
@@ -1672,34 +1672,32 @@
 ().getBytes());%0A
-  
         ssOutput
@@ -1749,34 +1749,32 @@
 _PART);%0A        
- 
- 
 ssOutputStream.w
@@ -1844,34 +1844,32 @@
 tes());%0A        
-  
 ssOutputStream.w
@@ -1905,34 +1905,32 @@
 D_COLUMN_PART);%0A
-  
         ssOutput
@@ -2002,34 +2002,32 @@
 tes());%0A        
-  
 ssOutputStream.w
@@ -2063,34 +2063,32 @@
 D_COLUMN_PART);%0A
-  
         ssOutput
@@ -2159,34 +2159,32 @@
 tes());%0A        
- 
- 
 ssOutputStream.w
@@ -2224,26 +2224,24 @@
 LUMN_PART);%0A
-  
         ssOu
@@ -2314,26 +2314,24 @@
 etBytes());%0A
-  
       %7D%0A    
@@ -2324,34 +2324,32 @@
 ;%0A      %7D%0A      
-  
 ssOutputStream.w
@@ -2376,20 +2376,14 @@
     
-  
 %7D%0A  
- 
- 
 %7D%0A
-  
   fi
@@ -2390,18 +2390,16 @@
 nally %7B%0A
-  
     ssOu
@@ -2418,26 +2418,24 @@
 lose();%0A    
-  
 indexWriter.
@@ -2449,20 +2449,11 @@
 ;%0A  
-  %7D%0A  %7D%0A  %0A  %0A%7D
+%7D%0A%7D%0A*/
 %0A

Remove ugly code duplication