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

using System;
using System.Collections.Generic;
using System.Linq;
using System.Globalization;
using System.IO;


namespace BCG {
   // each stream op in the replay file is reprsented by a packet here
   public class PlaybackPacket {
      public readonly int idx;
      public readonly long timestamp;
      public readonly byte[] data;
      public long replay_timestamp = 0; // non zero it has been replayed
      public long offset_from_previous = 0;
      public PlaybackPacket(int the_idx, long the_timestamp, byte[] the_data) {
         this.idx = the_idx;
         this.timestamp = the_timestamp;
         this.data = the_data;
      }
      public bool replayed {
         get { return replay_timestamp != 0; }
      }
   }

   public enum PlaybackType {
      Streaming = 0,
      Snapshot
   }
   public class SnapshotData {
      public DateTime snapshot_time;
      public string snapshot_file;
      public SnapshotData(DateTime snapshot_time_arg, string snapshot_file_arg) {
         snapshot_time = snapshot_time_arg;
         snapshot_file = snapshot_file_arg;
      }
      public IEnumerable<PlaybackPacket> packets = new List<PlaybackPacket>();
   }
   public static class StreamPersistorUtils {
      public static string find_snapshot_file(string directory_name, DateTime snapshot_time) {
         string[] file_names = Directory.GetFiles(directory_name);
         for (int i = 0; i < file_names.Count() - 1; i++) {
            string fn = file_names[i];
            string next_fn = file_names[i + 1];
            DateTime f_dt = PersistorStream.parse_datetime_from_filename(fn);
            DateTime next_f_dt = PersistorStream.parse_datetime_from_filename(next_fn);
            if (snapshot_time >= f_dt && snapshot_time <= next_f_dt) {
               return fn;
            }
         }
         return String.Empty;
      }
      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();
      }
      public static DateTime epoch_to_dt(long epoch_seconds) {
         return DateTimeOffset.FromUnixTimeSeconds(epoch_seconds).DateTime;
      }

      public static PlaybackPacket find_packet_at_time(IEnumerable<PlaybackPacket> packets, DateTime snapshot_time) {
         log_file_ts(packets, snapshot_time);
         long snapshot_time_epoch = PersistorStream.dt_to_epoch(snapshot_time);
         for (int i = 0; i < packets.Count() - 1; i++) {
            PlaybackPacket curr_p = packets.ElementAt(i);
            PlaybackPacket next_p = packets.ElementAt(i + 1);
            if (curr_p.timestamp <= snapshot_time_epoch && next_p.timestamp >= snapshot_time_epoch) {
               return curr_p;
            }
         }
         // if for some reason it cant find it, send the last packet
         return packets.Last();
      }

      private static void log_file_ts(IEnumerable<PlaybackPacket> packets, DateTime snapshot_time) {
         string big_log_msg = String.Empty;
         foreach (var p in packets) {
            big_log_msg += $"packet idx: {p.idx} epoch_ts: {p.timestamp} ts: {epoch_to_dt(p.timestamp)} snapshot_time: {snapshot_time}";
            big_log_msg += "\n";
         }
         bts.Log.Gen.info($"Packets: {big_log_msg}");
      }

      public static IEnumerable<PlaybackPacket> load_packets_from_file(FileStream fs) {
         BinaryReader br = new BinaryReader(fs);
         List<PlaybackPacket> rv = new List<PlaybackPacket>();
         while (fs.Position < fs.Length) {
            try {
               long ts = br.ReadInt64();
               int sz = br.ReadInt32();
               byte[] data = br.ReadBytes(sz);
               var pb = new PlaybackPacket(rv.Count, ts, data);
               rv.Add(pb);
            } catch (Exception ex) {
               bts.Log.Gen.error($"ex: {ex}");
               break;
            }
         }
         // setting offset field
         for (int i = 0; i < rv.Count; i++) {
            if (i == 0)
               continue;
            rv[i].offset_from_previous = rv[i].timestamp - rv[i - 1].timestamp;
         }
         return rv;
      }
   }
}