spark/src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using Google.Protobuf;
using System.Globalization;
using BCG;

namespace bts {
   // Writes all the udpates to a stream in a file or multiples files dependong on the rollover size (as opposed to Saver
   // which just writes one update, as a create)
   // File naming:
   // token_epoch_yyyyMMdd_HHmmss
   // File location: directory(passed)\token\
   // directory passed on, it creates a folder with the token passed
   // data:
   // epochtime, size, serialized data
   [DataStreamUserRespresentation("pers")]
   // Saves each udpate of a stream into a file. The token is the the unique logical name of the stream,
   // which maps the name of the directory where the streams are saved. The directory (3rd arg) is the 
   // parent directory where the token folder is stored. Each file goes up to rolloevr_size after which a new
   // file is created. The file names are token_epoch_datetime
   // the first entry in a file is the complete state
   public class PersistorStream : ViewStream {
      private ViewStream _source_stream;
      private string _token;
      private FileStream _out_fs;
      private IStreamIOMsgSerializer _serializer;// = new UniversalStreamIOSerializer();
      private string _file_name = String.Empty;
      private long _rollover_size;
      private readonly string _directory_path;
      private int _n_updates_in_current_file = 0;

      public PersistorStream(ViewStream src, string token, string directory, int rollover_size) {
         _source_stream = src;
         _token = token;
         _directory_path = directory + $@"\{token}\";
         _open_new_file();
         _rollover_size = rollover_size;
      }

      protected override void myOnFirstSubscriber() {
         _source_stream.onChange += _on_change;
      }

      protected override void myOnLastSubscriber() {
         _source_stream.onChange -= _on_change;
         _out_fs.Close();
      }


      public static DateTime parse_datetime_from_filename(string s) {
         // allbooks_1697054438_2023-10-11_16-00-38
         string[] toks = s.Split("_");
         bts.Log.assert(toks.Length == 4);
         string dt = $"{toks[2]}_{toks[3]}";
         return parse_datetime(dt);
      }

      // this is used to parse the string passed the datetime texts passed to cons of the stream 
      public static DateTime parse_datetime(string s) {
         // for whatever reason "yyyy-MM-dd_HH-mm-ss" is not parsable using DateTime.ParseExact (it only gets
         // the Date not the time), "yyyy-MM-dd_HH:mm:ss" is parsable
         // however a file name can not have : in its name, therefore we need to parse "yyyy-MM-dd_HH-mm-ss"
         DateTime rv = DateTime.ParseExact(s, PersistorStream.DateTime_format, CultureInfo.InvariantCulture);
         string[] toks = s.Split("_");
         string time_str = toks[1];
         string[] time_toks = time_str.Split("-");
         bts.Log.assert(time_toks.Length == 3);
         int hh = Int32.Parse(time_toks[0]);
         int mm = Int32.Parse(time_toks[1]);
         int ss = Int32.Parse(time_toks[2]);
         rv = rv.Date;
         rv = rv.AddHours(hh);
         rv = rv.AddMinutes(mm);
         rv = rv.AddSeconds(ss);
         return rv;
      }

      public static string DateTime_format {
         get {
            return "yyyy-MM-dd_HH-mm-ss";
         }
      }

      public static long dt_to_epoch(DateTime dt) {
         DateTimeOffset dtos = new DateTimeOffset(dt, TimeSpan.Zero);
         return dtos.ToUnixTimeSeconds();
      }
      public static long epoch_now() {
         return DateTimeOffset.UtcNow.ToUnixTimeSeconds();
      }

      void _on_change(object new_value) {
         DataPacket dp = new_value as DataPacket;
         byte[] serialized_update = null;
         bool first_write = _serializer == null || _n_updates_in_current_file == 0;
         _serializer = _serializer ?? new UniversalStreamIOSerializer();
         if (first_write) {
            // first file always starts with create
            serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.CreateMessage(dp.newRows)));
         } else
            serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.MultipleRemoveAddMessage(dp.deletedRows, dp.addedRows)));
         BinaryWriter bw = new BinaryWriter(_out_fs);
         long utc_epoch = epoch_now();

         //time, length of data, data
         //bw.Write(utc_epoch);
         //bw.Write(serialized_update.Length);
         bw.Write(serialized_update);
         _out_fs.Flush();
         myFireEvent(dp);
         _n_updates_in_current_file++;
         if (_out_fs.Length > _rollover_size) {
            _close_current_and_open_new();
         }
      }

      void _close_current_and_open_new() {
         _out_fs.Close();
         _open_new_file();
      }

      void _open_new_file() {
         var epoch = epoch_now();
         _n_updates_in_current_file = 0;
         // Check if the directory was successfully created
         if (!Directory.Exists(_directory_path)) {
            Directory.CreateDirectory(_directory_path);
         }
         _file_name = $"{_directory_path}{_token}_{epoch}_{DateTime.Now.ToString(DateTime_format)}";
         _out_fs = new FileStream(_file_name, FileMode.OpenOrCreate);
      }
   }
}