spark/src/DataStreamSuite/DataStreamTests/TestDriverStream.cs

namespace Bts.DataStreamTests;

using BCG;
using BCG.Utilities.General;

using Bts.DataStreamCore;

using global::Spark.Core.Collections;

public class TestDriverStream : ViewStream, IWorker {
   readonly IBlueTableSchema schema;
   readonly object lockObj = new();
   readonly ViewStreamDatabase? vsd;
   BlueTable? lastPublishedTable;
   BlueTable? currentTable;

   public TestDriverStream(IBlueTableSchema schema, ViewStreamDatabase? vsd = null) {
      this.schema = schema;
      this.vsd = vsd;
   }

   bool publishOnDoWork => vsd is not null;

   public void Reset(params IBlueRow[] rows) =>
      Reset(rows.AsEnumerable());

   public void Reset(IEnumerable<IBlueRow> rows) {
      lock (lockObj) {
         currentTable = new BlueTable(schema);
         currentTable.AddRange(rows.Select(r => new KeyValuePair<int, IBlueRow>(r.Id(), r)));
         if (!publishOnDoWork) {
            syncAndFire();
         }
      }
   }

   protected override void myOnFirstSubscriber() =>
      vsd?.addWorker(this);

   protected override void myOnLastSubscriber() =>
      vsd?.removeWorker(this);

   void IWorker.doWork() {
      lock (lockObj) {
         if (!publishOnDoWork || currentTable is null || currentTable == lastPublishedTable) {
            return;
         }
         syncAndFire();
      }
   }

   void syncAndFire() {
      if (currentTable is null) {
         throw new ArgumentNullException(nameof(currentTable));
      }
      var deletedRows = new BlueTable(schema);
      var addedRows = new BlueTable(schema);
      if (lastPublishedTable is null) {
         addedRows.addRows(currentTable);
      } else {
         var comp = BlueRowEqualityComparer.Instance;
         deletedRows.AsDictionary().SetItems(lastPublishedTable.Where(oldRow =>
            !(currentTable.AsDictionary().TryGetValue(oldRow.Key, out var newRow) && comp.Equals(newRow, oldRow.Value))));
         addedRows.SetItems(currentTable.Where(newRow =>
            !(lastPublishedTable.AsDictionary().TryGetValue(newRow.Key, out var oldRow) && comp.Equals(oldRow, newRow.Value))));
      }
      lastPublishedTable = currentTable;
      if (deletedRows.Count > 0 || addedRows.Count > 0) {
         myFireEvent(new DataPacket(currentTable, addedRows, deletedRows));
      }
   }
}