WellKnownColumns.cs

Also change snake_case fn names to PascalCase

src/DataStreamSuite/DataStream/BlueOrb/TimeSeriesStream.cs

@@ -47,6 +47,8 @@ namespace bts.BlueOrb {
       private void _onstream_change(object o) {
          var dp = ExceptionAssert.Argument.Is<IDataPacket>(o);
          DateTime now = DateTime.Now;
+         var addedRowsBd = ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, IBlueRow>());
+         var deletedRowsBd = ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, IBlueRow>());
          using var _ = gate.BeginTransaction();
          if (_table == null) {
             var colInfos = dp.newRows.schema.columnInfo;
@@ -58,21 +60,19 @@ namespace bts.BlueOrb {
             }
             _table = new CircularBufferTable(new BlueTableSchema(colInfos, dp.newRows.schema.tableName), _size);
          }
-         var addedRowsBd = BlueTableImmutable.CreateBuilder(_table.schema);
-         var deletedRowsBd = BlueTableImmutable.CreateBuilder(_table.schema);
          foreach (IBlueRow r in dp.addedRows) {
             MutableBlueRow ar = new MutableBlueRow(_table.schema, r);
             ar.setValue(timeColumn, now);
             int id = _id++;
             ar.setValue("id", id);
             _table = _table.Add(ar, out var oldRow);
-            addedRowsBd.AddOrUpdate(ar);
+            addedRowsBd.Add(new(id, ar));
             if (oldRow is not null) {
-               deletedRowsBd.AddOrUpdate(oldRow);
+               deletedRowsBd.Add(new(oldRow.Id(), oldRow));
             }
          }
-         var addedRows = addedRowsBd.ToImmutable();
-         var deletedRows = deletedRowsBd.ToImmutable();
+         var addedRows = new BlueTableImmutable(_table.schema, addedRowsBd.ToImmutable());
+         var deletedRows = new BlueTableImmutable(_table.schema, deletedRowsBd.ToImmutable());
          var dp_out = new DataPacket(_table, addedRows, deletedRows);
          myFireEvent(dp_out);
       }

src/DataStreamSuite/DataStream/DataStreamExtensions.cs

@@ -8,6 +8,7 @@ using bts.BlueOrb;
 
 using Bts.DataStreamCore;
 using Bts.PubSub;
+using Bts.SharedStreams.StreamFactories;
 
 using mbts.utils;
 
@@ -138,7 +139,7 @@ public static class DataStreamExtensions {
             var resDp = bd.ToDataPacket(out var newRows);
             return (newRows, resDp);
          })
-         .Where(t => t.Dp is not null && (t.Dp.addedRows.Count > 0 || t.Dp.deletedRows.Count > 0))
+         .Where(t => t.Dp is not null && (t.Dp.addedRows.Any() || t.Dp.deletedRows.Any()))
          .Select(t => t.Dp!);
 
    public static IObservable<IDataPacket> AsDataPackets<T>(this IObservable<TableUpdate<T, int>> source, IBlueTableSchema schema) where T : IBlueRow =>
@@ -155,7 +156,7 @@ public static class DataStreamExtensions {
             var resDp = bd.ToDataPacket(out var newRows);
             return (newRows, resDp);
          })
-         .Where(t => t.Dp.addedRows.Count > 0 || t.Dp.deletedRows.Count > 0)
+         .Where(t => t.Dp.addedRows.Any() || t.Dp.deletedRows.Any())
          .Select(t => t.Dp);
 
    public static IObservable<IDataPacket> AsDataPackets<T, TKey>(this IObservable<TableUpdate<T, TKey>> source, BlueTableSchemaAdapter<T> schema) where TKey : IEquatable<TKey> {
@@ -170,8 +171,9 @@ public static class DataStreamExtensions {
             (BlueTableImmutableAdapter<T> Adapter, IDataPacket DataPacket) state, TableUpdate<T, TKey> tableUpdate) {
             var bd = state.Adapter.ToDataPacketBuilder();
             bd.Update(tableUpdate);
-            var newRows = bd.ToImmutable(out var dp);
-            return (newRows, dp);
+            var dp = bd.ToDataPacket();
+            var tableAdapter = bd.ToImmutable();
+            return (bd.ToImmutable(), dp);
          }
       }
    }
@@ -195,46 +197,6 @@ public static class DataStreamExtensions {
    public static ViewStream AsViewStream<T>(this IObservable<TableUpdate<T, int>> source, IBlueTableSchema schema) where T : IBlueRow =>
       source.AsDataPackets(schema).AsViewStream();
 
-   public static IObservable<IDataPacket> OnWorkScheduler(this IObservable<IDataPacket> source, IWorkScheduler scheduler) =>
-      Observable.Create<IDataPacket>(o => {
-         var inputGate = new AtomicGate();
-         var outputGate = new AtomicGate();
-         BlueTableImmutable? lastTable = null;
-         BlueTableImmutable.DataPacketBuilder? bd = null;
-         IDataPacket? pendingDpOut = null;
-         return Disposable.Create(source.Subscribe(OnNext), scheduler.Subscribe(DoWork));
-
-         void OnNext(IDataPacket value) {
-            using (inputGate.BeginTransaction()) {
-               lastTable ??= new(value.newRows.schema);
-               bd ??= lastTable.ToDataPacketBuilder();
-               if (pendingDpOut is null) {
-                  // The last outgoing datapacket was just fired. Don't merge this update with the
-                  // previous one (otherwise its adds/removes would be fired twice).
-                  bd.Reset();
-               }
-               foreach (IBlueRow r in value.deletedRows) {
-                  bd.Remove(r.Id());
-               }
-               foreach (IBlueRow r in value.addedRows) {
-                  bd.AddOrUpdate(r);
-               }
-               pendingDpOut = bd.ToDataPacket(out lastTable);
-            }
-         }
-
-         void DoWork() {
-            IDataPacket? dpOut = null;
-            if (outputGate.TryEnter()) {
-               using var exit = Disposable.Create(outputGate.Exit);
-               (dpOut, pendingDpOut) = (pendingDpOut, dpOut);
-               if (dpOut is not null) {
-                  o.OnNext(dpOut);
-               }
-            }
-         }
-      });
-
    public static ViewStream GroupByApply(this ViewStream viewStream, string[] columns, Func<IBlueRowCollection, IBlueRowCollection> apply) {
       var groupByTables = GroupByApplyStream.CreateGroupByTable(viewStream, columns);
       var applyTables = GroupByApplyStream.CreateApplyTable(groupByTables, FunctionEvaluator.Create(apply));

src/DataStreamSuite/DataStream/Primitives/Union/RowBasedAppendStream.cs

@@ -1,62 +0,0 @@
-namespace bts.BlueOrb {
-   using BCG;
-
-   using System;
-   using System.Collections;
-   using System.Collections.Generic;
-   using System.Runtime.Intrinsics.Arm;
-
-   // given a parent stream, creates a stream for each row in the parent stream and appends them all together
-   public class RowBasedAppendStream : ViewStream {
-      private Dictionary<string, ViewStream> _stream_map = new Dictionary<string, ViewStream>();
-      private MutableAppendStream _final_stream = new MutableAppendStream();
-      private ViewStream _parent_stream;
-      private string[] _key_cols;
-      private Func<IBlueRow, ViewStream> _make_stream;
-      public RowBasedAppendStream(ViewStream parent_stream,
-                                 Func<IBlueRow, ViewStream> make_stream,
-                                 params string[] key_cols) {
-         _parent_stream = parent_stream;
-         _make_stream = make_stream;
-         _key_cols = key_cols;
-      }
-
-      protected override void myOnFirstSubscriber() {
-         _parent_stream.onChange += _on_parent_stream_change;
-         _final_stream.onChange += _on_final_stream_change;
-      }
-
-      protected override void myOnLastSubscriber() {
-         _parent_stream.onChange -= _on_parent_stream_change;
-         _final_stream.onChange -= _on_final_stream_change;
-      }
-
-
-      private void _on_parent_stream_change(object dp_obj) {
-         DataPacket dp = dp_obj as DataPacket;
-         foreach (IBlueRow r in dp.deletedRows) {
-            string hash = r.get_row_cols_as_string(_key_cols);
-            ViewStream hash_str = null;
-            if (_stream_map.TryGetValue(hash, out hash_str)) {
-               _stream_map.Remove(hash);
-               _final_stream.removeStream(hash_str);
-            } else {
-               Log.assert(false);
-            }
-         }
-         foreach (IBlueRow r in dp.addedRows) {
-            string hash = r.get_row_cols_as_string(_key_cols);
-            ViewStream hash_str = null;
-            if (!_stream_map.TryGetValue(hash, out hash_str)) {
-               hash_str = _make_stream(r);
-               _stream_map.Add(hash, hash_str);
-            }
-         }
-
-      }
-
-      private void _on_final_stream_change(object dp_obj) {
-         myFireEvent(dp_obj);
-      }
-   }
-}

src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs

@@ -517,7 +517,7 @@ namespace BCG {
             rv.Add(r.get<T>(col));
          return rv;
       }
-      public static IEnumerable<T> get_colvalues<T>(IEnumerable<IBlueRow> rows, int start_idx, int n_items, string col) {
+      public static IEnumerable<T> get_colvalues<T>(IEnumerable<IBlueRow> rows, int start_idx, int n_items,string col) {
          List<T> rv = new List<T>();
          int end_idx = Math.Min(start_idx + n_items, rows.Count());
          for (int i = start_idx; i < end_idx; i++) {
@@ -538,16 +538,6 @@ namespace BCG {
          return rv;
       }
 
-      public static string get_row_cols_as_string(this IBlueRow r, params string[] cols) {
-         string rv = String.Empty;
-         foreach (string c in cols) {
-            object o = r[c];
-            o = o ?? DBNull.Value;
-            rv += o.ToString() + "_";
-         }
-         return rv;
-      }
-
       public static IDictionary<T, IEnumerable<IBlueRow>> group_by<T>(IBlueRowCollection rows, string gp_col) {
          Dictionary<T, List<IBlueRow>> rv = new Dictionary<T, List<IBlueRow>>();
          foreach (IBlueRow r in rows) {
@@ -702,7 +692,7 @@ namespace BCG {
 
       #region json
 
-      public static void write_to_json_file(this IBlueRowCollection table, string file_path) {
+      public static void write_to_json_file(this IBlueRowCollection table, string file_path) { 
          write_dict_to_json_file(file_path, table.to_namevalue_pairs());
       }
 
@@ -793,7 +783,7 @@ namespace BCG {
          return rv;
       }
 
-      public static List<Dictionary<string, string>> to_namevalue_pairs(this IBlueRowCollection table) {
+      public static List<Dictionary<string,string>> to_namevalue_pairs(this IBlueRowCollection table) {
          List<Dictionary<string, string>> rv = new List<Dictionary<string, string>>();
          foreach (IBlueRow r in table) {
             rv.Add(r.to_dict());

src/DataStreamSuite/DataStreamCore/BlueData/BlueRowAdapter.cs

@@ -3,8 +3,6 @@ namespace Bts.DataStreamCore;
 
 using BCG;
 
-using Bts.DataStream;
-
 public class BlueRowAdapter<T> : IBlueRow {
    readonly BlueTableSchemaAdapter<T> adapter;
    readonly T data;

src/DataStreamSuite/DataStreamCore/BlueData/BlueTableImmutable.cs

@@ -21,7 +21,7 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
    public BlueTableImmutable(IBlueTableSchema schema)
       : this(schema, ImmutableSortedSet.Create(KeyValuePairComparer.KeyComparer<int, IBlueRow>())) { }
 
-   BlueTableImmutable(IBlueTableSchema schema, ImmutableSortedSet<KeyValuePair<int, IBlueRow>> rows) {
+   public BlueTableImmutable(IBlueTableSchema schema, ImmutableSortedSet<KeyValuePair<int, IBlueRow>> rows) {
       this.schema = schema;
       this.rows = rows;
    }
@@ -43,12 +43,6 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
       set => throw new NotSupportedException();
    }
 
-   public static DataPacketBuilder CreateBuilder(IBlueTableSchema schema) =>
-      new(new(schema));
-
-   public static BlueTableImmutable CreateRange(IBlueTableSchema schema, IEnumerable<KeyValuePair<int, IBlueRow>> rows) =>
-      new(schema, ImmutableSortedSet.CreateRange(KeyValuePairComparer.KeyComparer<int, IBlueRow>(), rows));
-
    public IBlueRow GetRow(int index) =>
       rows[index].Value;
 
@@ -197,12 +191,12 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 }
 
 // public class BlueTableImmutable<T> : IBlueRowCollection<T> where T : IBlueRow {
-//    readonly ImmutableDictionary<int, T> rows;
+//    readonly ImmutableSortedSet<KeyValuePair<int, T>> rows;
 
 //    public BlueTableImmutable(IBlueTableSchema schema)
-//       : this(schema, ImmutableDictionary<int, T>.Empty) { }
+//       : this(schema, ImmutableSortedSet<KeyValuePair<int, T>>.Empty) { }
 
-//    public BlueTableImmutable(IBlueTableSchema schema, ImmutableDictionary<int, T> rows) {
+//    public BlueTableImmutable(IBlueTableSchema schema, ImmutableSortedSet<KeyValuePair<int, T>> rows) {
 //       this.schema = schema;
 //       this.rows = rows;
 //    }
@@ -212,19 +206,32 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 //    public IBlueTableSchema schema { get; }
 //    public object SyncRoot => this;
 
-//    public T this[int id] => rows[id];
+//    public T this[int id] =>
+// #nullable disable
+//       rows.TryGetValue(new(id, default), out var kv)
+// #nullable enable
+//          ? kv.Value
+//          : throw new KeyNotFoundException();
 
 //    public bool containsID(int id) =>
-//       rows.ContainsKey(id);
+// #nullable disable
+//       rows.Contains(new(id, default));
+// #nullable enable
+
+//    public IList<object?> GetColumn(int index) =>
+//       rows.WrapSelector(kv => kv.Value[index], null);
 
 //    public void CopyTo(Array array, int index) {
-//       foreach (var r in rows.Values) {
-//          array.SetValue(r, index++);
+//       foreach (var (k, v) in rows) {
+//          array.SetValue(v, index++);
 //       }
 //    }
 
-//    public IEnumerator<T> GetEnumerator() =>
-//       rows.Values.GetEnumerator();
+//    public IEnumerator<T> GetEnumerator() {
+//       foreach (var kv in rows) {
+//          yield return kv.Value;
+//       }
+//    }
 
 //    IEnumerator IEnumerable.GetEnumerator() =>
 //       GetEnumerator();
@@ -234,11 +241,11 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 
 //    public class DataPacketBuilder {
 //       readonly IBlueTableSchema schema;
-//       readonly ImmutableDictionary<int, T>.Builder table;
-//       readonly ImmutableDictionary<int, T>.Builder added =
-//          ImmutableDictionary<int, T>.Empty.ToBuilder();
-//       readonly ImmutableDictionary<int, T>.Builder removed =
-//          ImmutableDictionary<int, T>.Empty.ToBuilder();
+//       readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder table;
+//       readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder added =
+//          ImmutableSortedSet<KeyValuePair<int, T>>.Empty.ToBuilder();
+//       readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder removed =
+//          ImmutableSortedSet<KeyValuePair<int, T>>.Empty.ToBuilder();
 
 //       public DataPacketBuilder(BlueTableImmutable<T> table) {
 //          schema = table.schema;
@@ -247,30 +254,37 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 
 //       public void AddOrUpdate(T row, IEqualityComparer<IBlueRow>? comparer = null) {
 //          var id = row.Id();
-//          if (table.TryGetValue(id, out var orig)) {
-//             if (comparer?.Equals(row, orig) ?? false) {
+// #nullable disable
+//          if (table.TryGetValue(new(id, default), out var orig)) {
+// #nullable enable
+//             if (comparer?.Equals(row, orig.Value) ?? false) {
 //                // No op: This row was already in the table (according to 'comparer').
 //                // Remove it from 'added' and 'removed'.
-//                added.Remove(id);
-//                removed.Remove(id);
+//                added.Remove(new(id, row));
+//                removed.Remove(new(id, orig.Value));
 //                return;
 //             }
 //             // Change op: A row with this id was already in the table,
 //             // but its values have changed (according to 'comparer').
-//             removed[id] = orig;
+//             removed.Remove(orig);
+//             removed.Add(orig);
 //          }
-//          added[id] = row;
+//          added.Remove(new(id, row));
+//          added.Add(new(id, row));
 //       }
 
 //       public void Remove(int id) {
-//          added.Remove(id);
-//          if (table.TryGetValue(id, out var orig)) {
-//             removed[id] = orig;
+// #nullable disable
+//          added.Remove(new(id, default));
+//          if (table.TryGetValue(new(id, default), out var orig)) {
+// #nullable enable
+//             removed.Remove(orig);
+//             removed.Remove(orig);
 //          }
 //       }
 
 //       public DataPacket ToDataPacket(bool reset = true) {
-//          table.RemoveRange(removed.Keys);
+//          table.RemoveRange(removed);
 //          table.AddRange(added);
 //          var newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
 //          var addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
@@ -283,7 +297,7 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 //       }
 
 //       public IDataPacket<T> ToDataPacketT(bool reset = true) {
-//          table.RemoveRange(removed.Keys);
+//          table.RemoveRange(removed);
 //          table.AddRange(added);
 //          var newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
 //          var addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
@@ -301,7 +315,7 @@ public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
 //          out BlueTableImmutable<T> removedRows,
 //          bool reset = true
 //       ) {
-//          table.RemoveRange(removed.Keys);
+//          table.RemoveRange(removed);
 //          table.AddRange(added);
 //          newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
 //          addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
@@ -325,7 +339,7 @@ public class BlueTableImmutableAdapter<T> : IBlueRowCollection {
    public BlueTableImmutableAdapter(BlueTableSchemaAdapter<T> schemaAdapter)
       : this(schemaAdapter, ImmutableSortedSet.Create(KeyValuePairComparer.KeyComparer<int, T>())) { }
 
-   BlueTableImmutableAdapter(BlueTableSchemaAdapter<T> schemaAdapter, ImmutableSortedSet<KeyValuePair<int, T>> rows) {
+   public BlueTableImmutableAdapter(BlueTableSchemaAdapter<T> schemaAdapter, ImmutableSortedSet<KeyValuePair<int, T>> rows) {
       this.schemaAdapter = schemaAdapter;
       this.rows = rows;
    }
@@ -467,19 +481,23 @@ public class BlueTableImmutableAdapter<T> : IBlueRowCollection {
             AddOrUpdate(r);
          }
       }
+      public IDataPacket ToDataPacket(bool reset = true) =>
+         ToDataPacket(out _, reset);
 
-      public BlueTableImmutableAdapter<T> ToImmutable(out IDataPacket dp, bool reset = true) {
+      public IDataPacket ToDataPacket(out IBlueRowCollection newRows, bool reset = true) {
          table.RemoveRange(removed);
          table.AddRange(added);
-         var newRows = new BlueTableImmutableAdapter<T>(schemaAdapter, table.ToImmutable());
+         newRows = new BlueTableImmutableAdapter<T>(schemaAdapter, table.ToImmutable());
          var addedRows = new BlueTableImmutableAdapter<T>(schemaAdapter, added.ToImmutable());
          var removedRows = new BlueTableImmutableAdapter<T>(schemaAdapter, removed.ToImmutable());
          if (reset) {
             added.Clear();
             removed.Clear();
          }
-         dp = new DataPacket(newRows, addedRows, removedRows);
-         return newRows;
+         return new DataPacket(newRows, addedRows, removedRows);
       }
+
+      public BlueTableImmutableAdapter<T> ToImmutable() =>
+         new(schemaAdapter, table.ToImmutable());
    }
 }

src/DataStreamSuite/DataStreamCore/DynamicLoading/LiveMarketDataStreamAttribute.cs

@@ -1,5 +0,0 @@
-namespace Bts.DataStreamCore;
-
-using System;
-
-public class LiveMarketDataStreamAttribute : Attribute { }

src/DataStreamSuite/DataStreamCore/DynamicLoading/MarketDataClientParamAttribute.cs

@@ -1,5 +0,0 @@
-namespace Bts.DataStreamCore;
-
-using System;
-
-public class MarketDataClientParamAttribute : Attribute { }

src/DataStreamSuite/DataStreamCore/DynamicLoading/MarketDataStreamFactory.cs

@@ -1,49 +0,0 @@
-#nullable enable
-namespace Bts.DataStreamCore;
-
-using BCG;
-
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Reflection;
-
-/// <summary>
-/// This class is a temp hack to pass a shared md client to stream constructors.
-/// </summary>
-public class MarketDataStreamFactory {
-   readonly object mdClient;
-
-   public MarketDataStreamFactory(object mdClient) =>
-      this.mdClient = mdClient;
-
-   public ViewStream? Create(Type type, object?[]? args) {
-      var ctrs = type.GetConstructors();
-      int nArgs = args?.Length ?? 0;
-      // Since this class is a temp hack, it isn't super robust. Find a constructor with nArgs + 1
-      // or 2 params (for mdClient and optional viewStreamDatabase), inject the special args and
-      // instantiate the stream. If no such constructor exists, just use the original args.
-      foreach (var ctr in ctrs) {
-         var ctrParams = ctr.GetParameters();
-         if (!ctrParams.TryGetFirst(p => p!.GetCustomAttribute<MarketDataClientParamAttribute>() is not null, out var mdClientParam, out int mdClientIdx)) {
-            continue;
-         }
-         var paramInjections = new List<(int Index, object Value)>(2) {
-            (mdClientIdx, mdClient)
-         };
-         if (ctrParams.TryGetFirst(p => p.ParameterType == typeof(ViewStreamDatabase), out var vsdParam, out int vsdIdx)) {
-            paramInjections.Add((vsdIdx, ViewStreamDatabase.instance));
-         }
-         if (nArgs != ctrParams.Length - paramInjections.Count) {
-            continue;
-         }
-         paramInjections.Sort(Comparer<(int, object)>.Create((a, b) => a.Item1.CompareTo(b.Item1)));
-         var argList = args?.ToList() ?? new();
-         foreach (var (idx, val) in paramInjections) {
-            argList.Insert(idx, val);
-         }
-         return Activator.CreateInstance(type, argList.ToArray()) as ViewStream;
-      }
-      return Activator.CreateInstance(type, args) as ViewStream;
-   }
-}

src/DataStreamSuite/DataStreamCore/DynamicLoading/StreamFactoryLogs.cs

@@ -8,15 +8,9 @@ public static partial class StreamFactoryLogs {
    public static void LogCreatingStream(this ILogger logger, string streamType) =>
       logger.LogCreatingStream(streamType, DateTime.Now);
 
-   public static void LogCreatingStreamError(this ILogger logger, string streamType, string error) =>
-      logger.LogCreatingStreamError(streamType, error, DateTime.Now);
-
    [LoggerMessage(EventId = 1, Level = LogLevel.Warning, Message = """[crt_stream.json]{{ "typ": {StreamType}, "ts": {Timestamp} }}""")]
    public static partial void LogCreatingStream(this ILogger logger, string streamType, DateTime timestamp);
 
-   [LoggerMessage(EventId = 1, Level = LogLevel.Warning, Message = """[crt_stream_err.json]{{ "typ": {StreamType}, "err": {Error}, "ts": {Timestamp} }}""")]
-   public static partial void LogCreatingStreamError(this ILogger logger, string streamType, string error, DateTime timestamp);
-
    [LoggerMessage(EventId = 1, Level = LogLevel.Warning, Message = "Unable to create '{StreamType}' stream. Missing required parameter '{Parameter}'.")]
    public static partial void LogMissingParam(this ILogger logger, string streamType, string parameter);
 

src/DataStreamSuite/DataStreamCore/DynamicLoading/WellKnownStreamArgumentType.cs

@@ -10,7 +10,7 @@ public static class WellKnownStreamArgumentAliases {
    /// <summary>
    /// Aliases for a Type argument.
    /// </summary>
-   public const string Asset = "symbol,sym,product,asset";
+   public const string Symbol = "symbol,sym,product,asset";
 
    /// <summary>
    /// Aliases for a Type argument.
@@ -29,19 +29,21 @@ public static class WellKnownStreamArgumentAliases {
    /// </summary>
    public const string SecurityName = "name,sec,secName,security";
 
-   public const string QualifiedSecurityType = "secType,secTyp,securityType,securityTyp";
-   public const string QualifiedSecurityName = "sec,secName,security";
+   /// <summary>
+   /// Aliases for a SecurityType argument, excluding a few potentially ambiguous ones from
+   /// <see cref="SecurityType"/>.
+   /// </summary>
+   public const string QualifiedSecurityType = "secType,secTyp";
 
    public const string VenueRegex = "^(venue|exchange|exch)$";
-   public const string AssetRegex = "^(symbol|sym|asset|product)$";
+   public const string AssetRegex = "^(symbol|sym|product|asset)$";
    public const string TypeRegex = "^(type|typ)$";
    public const string SecurityTypeRegex = "^(secType|secTyp|type|typ)$";
    public const string SecurityNameRegex = "^(sec|security|name|secName|securityName)$";
 
    public const string ShowVenueRegex = "^(show|include)(Exch|Venue)$";
-   public const string ShowAssetRegex = "^(show|include)(Symbol|Sym|Asset|Product)$";
    public const string ShowSecurityTypeRegex = "^(show|include)(Sec|Security)?(Type|Typ)$";
-   public const string ShowSecurityNameRegex = "^(show|include)(Sec|Security|Name|SecName|SecurityName)$";
+   public const string ShowAssetRegex = "^(show|include)(Symbol|Sym|Asset|Product)$";
    public const string ShowExpirationRegex = "^(show|include)(Exp(iration|Date)?)$";
    public const string ShowCashPerPointRegex = "^(show|include)(CashPerPoint|Cpp)$";
 
@@ -53,6 +55,7 @@ public static class WellKnownStreamArgumentAliases {
    public const string ShowLegSecurityTypeRegex = "^(show|include)Leg(Sec|Security)?(Type|Typ)$";
    public const string ShowLegAssetRegex = "^(show|include)Leg(Symbol|Sym|Asset|Product)$";
    public const string ShowLegExpirationRegex = "^(show|include)Leg(Exp(iration|Date)?)$";
+   public const string ShowLegSymbolRegex = "^(show|include)Leg(Symbol|Sym|Asset|Product)$";
    public const string ShowLegNameRegex = "^(show|include)Leg(Name|Desc)$";
    public const string ShowLegCashPerPointRegex = "^(show|include)Leg(CashPerPoint|Cpp)$";
 

src/DataStreamSuite/DataStreamCore/ViewStreamDatabase.cs

@@ -3,7 +3,6 @@ using BCG.Utilities.General;
 using System.Collections.Generic;
 using System.Threading;
 using System.Diagnostics;
-using Bts.PubSub;
 using bts;
 using Bts.PubSub;
 using bts.collections;
@@ -117,8 +116,6 @@ namespace BCG {
          else
             _workers.Add((w, interval, DateTime.MinValue));
       }
-      void IWorkScheduler.Add(IWorker worker, TimeSpan interval) =>
-         addWorker(worker, interval > TimeSpan.Zero ? interval : null);
 
       public void removeWorker(IWorker w) {
          verifyAccess();
@@ -127,8 +124,6 @@ namespace BCG {
          else
             _workers.RemoveFirst(tup => tup.Worker == w);
       }
-      void IWorkScheduler.Remove(IWorker worker) =>
-         removeWorker(worker);
 
       internal static void SetGlobalInstance(ViewStreamDatabase vsd) {
          lock (myInstanceLockObject) {
@@ -137,6 +132,12 @@ namespace BCG {
          }
       }
 
+      void IWorkScheduler.Add(IWorker worker, TimeSpan interval) =>
+         addWorker(worker, interval);
+
+      void IWorkScheduler.Remove(IWorker worker) =>
+         removeWorker(worker);
+
       private List<PendingAddRemoveWorker> _pending_worker_ops = new List<PendingAddRemoveWorker>();
       private bool _is_worker_call_in_progress = false;
       class PendingAddRemoveWorker {

src/DataStreamSuite/DataStreamPublisher/Services/DataStreamPublisherService.cs

@@ -25,7 +25,7 @@ using System.Threading.Tasks;
 /// An <see cref="IHostedService"/> that publishes <see cref="ViewStream"/>s to the PubSub.
 /// It can be added to the application's services via <see cref="DataStreamPublisherServices.AddDataStreamPublisher"/>
 /// </summary>
-public sealed class DataStreamPublisherService : IHostedService, IDisposable {
+public class DataStreamPublisherService : IHostedService, IDisposable {
    readonly DataStreamPublisherConfig config;
    readonly IPubTableBroker pubTableBroker;
    readonly IWorkScheduler workScheduler;

src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json

@@ -6,7 +6,7 @@
     "Interval": "00:00:00.1"
   },
   "Spark": {
-    "ConfigServer": "http://127.0.0.1:8000/config"
+    "ConfigServer": "http://localhost:8000/config"
   },
   "SparkApi": {
     "Credentials": {

src/DataStreamSuite/DataStreamTests/DataPacketBuilder/DataPacketBuilderTests.cs

@@ -186,7 +186,8 @@ public class DataPacketBuilderTests {
    }
 
    static BlueTableImmutable getTable(params (int Id, double Value)[] rows) =>
-      BlueTableImmutable.CreateRange(schema, rows.Select(tup => new KeyValuePair<int, IBlueRow>(tup.Id, getBlueRow(tup))));
+      new(schema, ImmutableSortedSet.CreateRange(
+         rows.Select(tup => new KeyValuePair<int, IBlueRow>(tup.Id, getBlueRow(tup)))));
 
    static MutableBlueRow getBlueRow((int id, double value) tup) =>
       getBlueRow(tup.id, tup.value);
@@ -204,10 +205,11 @@ public class DataPacketBuilderTests {
       (int Id, double Value)[]? removedRows = null) {
       addedRows ??= Array.Empty<(int, double)>();
       removedRows ??= Array.Empty<(int, double)>();
+      var kvpComp = new ComparerBuilder<KeyValuePair<int, IBlueRow>>() { kv => kv.Key };
       return new DataPacket(
-         BlueTableImmutable.CreateRange(schema, GetRowKvps(newRows)),
-         BlueTableImmutable.CreateRange(schema, GetRowKvps(addedRows)),
-         BlueTableImmutable.CreateRange(schema, GetRowKvps(removedRows)));
+         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(newRows))),
+         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(addedRows))),
+         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(removedRows))));
 
       static IEnumerable<KeyValuePair<int, IBlueRow>> GetRowKvps((int Id, double Value)[] rows) =>
          rows.Select(tup => new KeyValuePair<int, IBlueRow>(tup.Id, getBlueRow(tup)));

src/DataStreamSuite/PubSub/Misc/WorkScheduler.cs

@@ -84,4 +84,3 @@ public class WorkScheduler : IWorkScheduler, IDisposable {
       }
    }
 }
-

src/DataStreamSuite/PubSub/Misc/WorkSchedulerExtensions.cs

@@ -35,10 +35,4 @@ public static class WorkSchedulerExtensions {
 
    public static IDisposable Subscribe(this IWorkScheduler scheduler, Action work, TimeSpan? interval = null) =>
       scheduler.Subscribe(new AnonymousWorker(work), interval);
-
-   public static void Add(this IWorkScheduler scheduler, Action work) =>
-      scheduler.Add(new AnonymousWorker(work), TimeSpan.Zero);
-
-   public static void Add(this IWorkScheduler scheduler, Action work, TimeSpan interval) =>
-      scheduler.Add(new AnonymousWorker(work), interval);
 }

src/DataStreamSuite/SharedStreams/BlueOrb/MarketData/BookStreamParams.cs

@@ -1,81 +0,0 @@
-namespace Bts.SharedStreams.MarketData;
-
-using bts.api;
-
-using Bts.DataStreamCore;
-
-using System.Text.RegularExpressions;
-
-public class BookStreamParams {
-   public Venue? Venue { get; set; }
-   public string? Asset { get; set; }
-   public SecurityType? SecurityType { get; set; }
-   public string? Name { get; set; }
-   public string? NameOrAsset { get; set; }
-   public string? SecurityName { get; set; }
-   public int NumLevels { get; set; } = 5;
-   public bool ShowVenue { get; set; } = true;
-   public bool ShowAsset { get; set; } = true;
-   public bool ShowSecurityType { get; set; } = true;
-   public bool ShowSecurityName { get; set; } = true;
-   public bool ShowSecurityId { get; set; } = true;
-   public bool ShowCashPerPoint { get; set; } = true;
-   public bool ShowPrice { get; set; } = true;
-   public bool ShowQuantity { get; set; } = true;
-   public bool ShowNotional { get; set; } = true;
-
-   public static BookStreamParams FromStringArg(string str, VenueArgumentParser venueParser, SecurityTypeArgumentParser secTypeParser) {
-      var opts = new BookStreamParams();
-      var arg = new StreamArgument(null, null, str);
-      if (venueParser.TryParse(arg, out var venue, out _)) {
-         opts.Venue = venue;
-      } else if (secTypeParser.TryParse(arg, out var secType, out _)) {
-         opts.SecurityType = secType;
-      } else {
-         opts.NameOrAsset = str;
-      }
-      return opts;
-   }
-
-   public Func<SecurityImmutable, bool> GetSecurityFilter() {
-      var nameRx = CreateRegex(Name);
-      var nameOrAssetRx = CreateRegex(NameOrAsset);
-      return SecurityFilter;
-
-      static Regex? CreateRegex(string? pattern) =>
-         pattern is null ? null : new(pattern);
-
-      bool SecurityFilter(SecurityImmutable s) {
-         if (Venue.HasValue && Venue.Value != s.Exchange) {
-            return false;
-         }
-         if (SecurityType.HasValue && SecurityType != s.security_type()) {
-            return false;
-         }
-         if (Asset is not null && Asset != s.Symbol) {
-            return false;
-         }
-         if (nameRx is not null && !nameRx.IsMatch(s.ToString()!)) {
-            return false;
-         }
-         if (nameOrAssetRx is not null && NameOrAsset != s.Symbol && !nameOrAssetRx.IsMatch(s.ToString()!)) {
-            return false;
-         }
-         return true;
-      }
-   }
-
-   public BookSchemaBuilder GetSchemaBuilder() =>
-      new() {
-         NumLevels = NumLevels,
-         ShowVenue = ShowVenue,
-         ShowAsset = ShowAsset,
-         ShowSecurityType = ShowSecurityType,
-         ShowSecurityName = ShowSecurityName,
-         ShowSecurityId = ShowSecurityId,
-         ShowCashPerPoint = ShowCashPerPoint,
-         ShowPrice = ShowPrice,
-         ShowQuantity = ShowQuantity,
-         ShowNotional = ShowNotional,
-      };
-}

src/DataStreamSuite/SharedStreams/BlueOrb/MarketData/Live/LiveBBOStream.cs

@@ -1,35 +1,26 @@
-namespace bts.BlueOrb {
+namespace bts.BlueOrb
+   {
+   using System;
+
    using BCG;
 
-   using Bts.DataStream;
-   using Bts.DataStreamCore;
-   using Bts.SharedStreams.MarketData;
+   using bts.api;
 
-   using Spark.Api.MarketData;
-
-   [LiveMarketDataStream]
-   [DataStreamUserRespresentation("LiveBBOStream")]
+   [DataStreamUserRespresentation(false)]
    public class LiveBBOStream : ViewStream {
-      public LiveBBOStream([MarketDataClientParam] IMarketDataClient mdClient, ViewStreamDatabase vsd)
-         : this(mdClient, vsd, null, null, null) { }
-
-      public LiveBBOStream([MarketDataClientParam] IMarketDataClient mdClient, ViewStreamDatabase vsd, string? venueStr, string? nameOrAsset)
-         : this(mdClient, vsd, venueStr, null, nameOrAsset) { }
-
-      public LiveBBOStream([MarketDataClientParam] IMarketDataClient mdClient, ViewStreamDatabase vsd, string? venueStr, string? secTypeStr, string? nameOrAsset) {
-         var opts = new BookStreamParams() {
-            NameOrAsset = nameOrAsset,
-         };
-         if (VenueArgumentParser.TryParse(venueStr, out var venue)) {
-            opts.Venue = venue;
+      public LiveBBOStream(string target)
+         : this(SecurityType.Future, target) { }
+      public LiveBBOStream(SecurityType sec_type, string target) {
+         ViewStream rs = null;
+         switch (sec_type) {
+            case SecurityType.Future:
+               rs = new LiveFuturesBBOStream(target);
+               break;
+            default:
+               Log.assert(false);
+               break;
          }
-         if (SecurityTypeArgumentParser.TryParse(secTypeStr, out var secType)) {
-            opts.SecurityType = secType;
-         }
-         var schemaBd = opts.GetSchemaBuilder();
-         var adapter = schemaBd.BboAdapter();
-         var secFilter = opts.GetSecurityFilter();
-         mySetRealSource(mdClient.Books(secFilter).AsDataPackets(adapter).OnWorkScheduler(vsd).AsViewStream());
+         mySetRealSource(rs);
       }
    }
 }

src/DataStreamSuite/SharedStreams/BlueOrb/MarketData/SecurityTypeArgumentParser.cs

@@ -42,9 +42,9 @@ public class SecurityTypeArgumentParser : IStreamArgumentParser<SecurityType> {
          if (TryParse(str, out result)) {
             return true;
          }
-         error = $"Failed to parse SecurityType from string '{str}'";
+         error = $"Failed to parse secType from string '{str}'";
       } else {
-         error = $"SecurityType parser received input of unexpected type '{val?.GetType()}'";
+         error = $"Venue parser received input of unexpected type '{val?.GetType()}'";
       }
       return false;
    }

src/DataStreamSuite/SharedStreams/BlueOrb/MarketData/VenueArgumentParser.cs

@@ -7,20 +7,20 @@ using Bts.DataStreamCore;
 public class VenueArgumentParser : IStreamArgumentParser<Venue> {
    public static readonly VenueArgumentParser Instance = new();
 
-   public static bool TryParse(string? str, out Venue result) {
+   public static bool TryParse(string? str, out Venue v) {
       if (str is null) {
-         result = Venue.Unset;
+         v = Venue.Unset;
          return false;
       }
-      if (Enum.TryParse(str, true, out result) || Enum.TryParse("VENUE_" + str, true, out result)) {
+      if (Enum.TryParse(str, true, out v) || Enum.TryParse("VENUE_" + str, true, out v)) {
          return true;
       }
-      result = str.ToUpper() switch {
+      v = str.ToUpper() switch {
          "CME" or "XCME" or "CEC" or "XCEC" => Venue.Glbx,
          "CBOT" or "CBT" or "XCBT" => Venue.Cboe,
          _ => Venue.Unset,
       };
-      return result != Venue.Unset;
+      return v != Venue.Unset;
    }
 
    public bool CanParse(StreamArgument arg, Type outputType) =>

src/DataStreamSuite/SharedStreams/StreamFactories/BookSchemaBuilder.cs

@@ -1,26 +1,25 @@
-namespace Bts.SharedStreams.MarketData;
+namespace Bts.SharedStreams.StreamFactories;
 
 using BCG;
 
 using bts.api;
 using bts.md;
 
-using Bts.DataStream;
 using Bts.DataStreamCore;
 
 using Spark.Api.MarketData;
 
 public class BookSchemaBuilder {
-   public int NumLevels { get; set; } = 5;
-   public bool ShowVenue { get; set; } = true;
-   public bool ShowAsset { get; set; } = true;
-   public bool ShowSecurityType { get; set; } = true;
-   public bool ShowSecurityName { get; set; } = true;
-   public bool ShowSecurityId { get; set; } = true;
-   public bool ShowCashPerPoint { get; set; } = true;
-   public bool ShowPrice { get; set; } = true;
+   public int NumLevels { get; set; }
+   public bool ShowNotional { get; set; }
    public bool ShowQuantity { get; set; } = true;
-   public bool ShowNotional { get; set; } = true;
+   public bool ShowPrice { get; set; } = true;
+   public bool ShowVenue { get; set; }
+   public bool ShowAsset { get; set; } = true;
+   public bool ShowSecurityType { get; set; }
+   public bool ShowSecurityName { get; set; }
+   public bool ShowSecurityId { get; set; } = true;
+   public bool ShowCashPerPoint { get; set; }
    public string TableName { get; set; } = "Book";
 
    public IBlueTableSchema ToSchema() {
@@ -30,12 +29,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          bd.Add(WellKnownColumns.Venue);
       }
-      if (ShowAsset) {
-         bd.Add(WellKnownColumns.Asset);
-      }
       if (ShowSecurityType) {
          bd.Add(WellKnownColumns.SecurityType);
       }
+      if (ShowAsset) {
+         bd.Add(WellKnownColumns.Asset);
+      }
       if (ShowSecurityName) {
          bd.Add(WellKnownColumns.SecurityName);
       }
@@ -67,12 +66,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          adapter.AddColumn(WellKnownColumns.Venue.Name, b => b.security.Exchange.ToString());
       }
-      if (ShowAsset) {
-         adapter.AddColumn(WellKnownColumns.Asset.Name, b => b.security.Symbol);
-      }
       if (ShowSecurityType) {
          adapter.AddColumn(WellKnownColumns.SecurityType.Name, b => b.security.security_type().ToString());
       }
+      if (ShowAsset) {
+         adapter.AddColumn(WellKnownColumns.Asset.Name, b => b.security.Symbol);
+      }
       if (ShowSecurityName) {
          adapter.AddColumn(WellKnownColumns.SecurityName.Name, b => b.security.ToString());
       }
@@ -104,12 +103,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          adapter.AddColumn(WellKnownColumns.Venue.Name, b => b.security.Exchange.ToString());
       }
-      if (ShowAsset) {
-         adapter.AddColumn(WellKnownColumns.Asset.Name, b => b.security.Symbol);
-      }
       if (ShowSecurityType) {
          adapter.AddColumn(WellKnownColumns.SecurityType.Name, b => b.security.security_type().ToString());
       }
+      if (ShowAsset) {
+         adapter.AddColumn(WellKnownColumns.Asset.Name, b => b.security.Symbol);
+      }
       if (ShowSecurityName) {
          adapter.AddColumn(WellKnownColumns.SecurityName.Name, b => b.security.ToString());
       }
@@ -149,4 +148,3 @@ public class BookSchemaBuilder {
    static Func<BookTuple, double> notionalGetter(Side side, int level) =>
       b => b.book[side, level].NotionalValue(b.security.ContractSpec.CashPerPoint);
 }
-

src/DataStreamSuite/SharedStreams/StreamFactories/BookStreamFactory.cs

@@ -2,113 +2,202 @@ namespace Bts.SharedStreams.StreamFactories;
 
 using BCG;
 
-using bts.api;
 using bts.BlueOrb;
 using bts.collections;
 
 using Bts.DataStream;
 using Bts.DataStreamCore;
+using Bts.DataStreamCore.StreamFactories;
 using Bts.PubSub;
-using Bts.SharedStreams.MarketData;
 
 using Microsoft.Extensions.Logging;
 
 using Spark.Api.MarketData;
+using Spark.Core.Reactive;
 
 using System;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Reactive.Linq;
 
-public class BookStreamFactory : DataStreamFactory {
-   readonly VenueArgumentParser venueParser = VenueArgumentParser.Instance;
-   readonly SecurityTypeArgumentParser secTypeParser = SecurityTypeArgumentParser.Instance;
-   readonly StringArgumentParser strParser = StringArgumentParser.Instance;
-   readonly IMarketDataClient mdClient;
+public class BookStreamFactory : DataStreamFactory, IRecursiveDataStreamFactory {
+   readonly LiveBboStreamFactory liveBboFactory;
+   readonly IMarketDataClient? mdClient;
    readonly ILogger logger;
    readonly IWorkScheduler workScheduler;
+   IDataStreamFactory? compositeFactory = null;
 
-   public BookStreamFactory(IMarketDataClient mdClient, ViewStreamDatabase viewStreamDatabase, ILogger<BookStreamFactory> logger)
-      : this(mdClient, viewStreamDatabase, (ILogger)logger) { }
-
-   BookStreamFactory(IMarketDataClient mdClient, IWorkScheduler workScheduler, ILogger logger) {
+   public BookStreamFactory(IMarketDataClient? mdClient, ILogger<BookStreamFactory> logger, ViewStreamDatabase workScheduler) {
       this.mdClient = mdClient;
       this.logger = logger;
       this.workScheduler = workScheduler;
-      this.Add("book", createBook, venueParser);
-      this.Add("book", createBook, secTypeParser);
-      this.Add("book", createBook, strParser);
-      this.Add("book", createBook, venueParser, secTypeParser);
-      this.Add("book", createBook, venueParser, strParser);
-      this.Add("book", createBook, secTypeParser, strParser);
-      this.Add("bbo", createBbo, strParser);
-      this.Add("bbo", createBbo, venueParser);
-      this.Add("bbo", createBbo, venueParser, secTypeParser);
-      this.Add("bboPriceSummary", createBboPriceSummary, StringArgumentParser.Instance);
-      this.Add("bboPriceSummary", createBboPriceSummary, venueParser);
-      this.Add("bboPriceSummary", createBboPriceSummary, venueParser, secTypeParser);
+      this.Add<BookStreamParams>(new[] { "book" }, CreateBook, logger);
+      this.Add<BookStreamParams>(new[] { "bbo" }, CreateBbo, logger);
+      this.Add<BookStreamParams>(new[] { "bboPriceSummary" }, createBboPriceSummary, logger);
+      liveBboFactory = new LiveBboStreamFactory(mdClient, logger);
+      Add(liveBboFactory);
+
+      ViewStream? CreateBook(BookStreamParams streamParams) {
+         if (!checkHasLiveMd()) {
+            return null;
+         }
+         return this.mdClient.Books(streamParams.GetSecurityFilter())
+            .AsDataPackets(streamParams.GetSchemaBuilder().BookTupleAdapter())
+            .AsViewStream()
+            .TimeSeries(streamParams.BufferSize, WellKnownColumnNames.Timestamp);
+      }
+
+      ViewStream? CreateBbo(BookStreamParams streamParams) {
+         if (!checkHasLiveMd()) {
+            return null;
+         }
+         return this.mdClient.Books(streamParams.GetSecurityFilter())
+            .AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter())
+            .AsViewStream()
+            .TimeSeries(streamParams.BufferSize, WellKnownColumnNames.Timestamp);
+      }
    }
 
-   ViewStream? createBook(Venue venue) =>
-      createBook(new BookStreamParams() { Venue = venue });
+   public void OnCompositeFactory(IDataStreamFactory compositeFactory) {
+      this.compositeFactory = compositeFactory;
+      liveBboFactory.OnCompositeFactory(compositeFactory);
+   }
 
-   ViewStream? createBook(SecurityType secType) =>
-      createBook(new BookStreamParams() { SecurityType = secType });
-
-   ViewStream? createBook(Venue venue, SecurityType secType) =>
-      createBook(new BookStreamParams() { Venue = venue, SecurityType = secType });
-
-   ViewStream? createBook(Venue venue, string nameOrAsset) =>
-      createBook(new BookStreamParams() { Venue = venue, NameOrAsset = nameOrAsset });
-
-   ViewStream? createBook(SecurityType secType, string nameOrAsset) =>
-      createBook(new BookStreamParams() { SecurityType = secType, NameOrAsset = nameOrAsset });
-
-   ViewStream? createBook(string str) =>
-      createBook(BookStreamParams.FromStringArg(str, venueParser, secTypeParser));
-
-   ViewStream? createBook(BookStreamParams streamParams) =>
-      mdClient.Books(streamParams.GetSecurityFilter())
-         .AsDataPackets(streamParams.GetSchemaBuilder().BookTupleAdapter())
-         .AsViewStream();
-
-   ViewStream? createBbo(string str) =>
-      createBbo(BookStreamParams.FromStringArg(str, venueParser, secTypeParser));
-
-   ViewStream? createBbo(Venue venue) =>
-      createBbo(new BookStreamParams() { Venue = venue });
-
-   ViewStream? createBbo(Venue venue, SecurityType secType) =>
-      createBbo(new BookStreamParams() { Venue = venue, SecurityType = secType });
-
-   ViewStream? createBbo(BookStreamParams streamParams) =>
-      mdClient.Books(streamParams.GetSecurityFilter())
-         .AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter())
-         .AsViewStream();
-
-   ViewStream? createBboPriceSummary(string str) =>
-      createBboPriceSummary(BookStreamParams.FromStringArg(str, venueParser, secTypeParser));
-
-   ViewStream? createBboPriceSummary(Venue venue) =>
-      createBboPriceSummary(new BookStreamParams() { Venue = venue });
-
-   ViewStream? createBboPriceSummary(SecurityType secType) =>
-      createBboPriceSummary(new BookStreamParams() { SecurityType = secType });
-
-   ViewStream? createBboPriceSummary(Venue venue, SecurityType secType) =>
-      createBboPriceSummary(new BookStreamParams() { Venue = venue, SecurityType = secType });
-
-   ViewStream? createBboPriceSummary(BookStreamParams opts) {
-      var bboStream = createBbo(opts);
-      if (bboStream is null) {
-         logger.LogCreatingStreamError("bboPriceSummary", "failed to create bboStream");
+   ViewStream? createBboPriceSummary(BookStreamParams streamParams) {
+      if (!checkHasLiveMd()) {
          return null;
       }
-      var bboTimeSeriesStream = new TimeSeriesStream(bboStream, opts.BufferSize, WellKnownColumnNames.Timestamp);
-      var tradeSchemaBd = new TradeSchemaBuilder();
-      var tradeSchema = tradeSchemaBd.ToSchema();
-      var tradeAdapter = tradeSchemaBd.TradeInfoAdapter();
-      var secFilter = opts.GetSecurityFilter();
-      var tradeStream = LiveTradeStream.TradeStream(mdClient.Trades(secFilter), tradeSchema, tradeAdapter).AsViewStream();
+      var secCriteria = streamParams.GetSecurityFilter();
+      mdClient.Books(secCriteria).WithHistory().Select(t => t.Table);
+      // var bboStream = mdClient.Books(secCriteria).AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter());
+      var bboStream = liveBboFactory.Create(streamParams)
+         ?? throw new ArgumentException("Failed to create bbo stream", nameof(streamParams));
+      var bboTimeSeriesStream = new TimeSeriesStream(bboStream, streamParams.BufferSize, WellKnownColumnNames.Timestamp);
+      ViewStream? tradeStream = null;
+      if (streamParams.Asset == "ES" && compositeFactory is not null) {
+         tradeStream = compositeFactory.Create("ReplayFutureLiveTradeStream", new (string?, int?, object?)[] { ("contract", 0, streamParams.Asset) });
+         if (tradeStream is null) {
+            logger.LogError("Failed to create trade stream for bboPriceSummary stream");
+            return null;
+         }
+         tradeStream = tradeStream.RenameColumns(true, ("trade_price", WellKnownColumnNames.Price), ("trade_size", WellKnownColumnNames.Quantity));
+      } else {
+         var tradeSchemaBd = new TradeSchemaBuilder();
+         var tradeSchema = tradeSchemaBd.ToSchema();
+         var tradeAdapter = tradeSchemaBd.TradeInfoAdapter();
+         tradeStream = LiveTradeStream.TradeStream(mdClient.Trades(secCriteria), tradeSchema, tradeAdapter).AsViewStream();
+      }
       var bboPriceSummaryStream = new BboPriceSummaryStream(bboStream.AsObservable(), tradeStream.AsObservable(), workScheduler, TimeSpan.FromSeconds(1));
       return bboPriceSummaryStream;
    }
+
+   static bool checkHasLiveMd([NotNullWhen(true)] IMarketDataClient? mdClient, ILogger? logger = null) {
+      if (mdClient is not null) {
+         return true;
+      }
+      logger?.LogError("Market data is unavailable (md client is null)");
+      return false;
+   }
+
+   [MemberNotNullWhen(true, nameof(mdClient))]
+   bool checkHasLiveMd() =>
+      checkHasLiveMd(mdClient, logger);
+
+   class LiveBboStreamFactory : IDataStreamFactory, IRecursiveDataStreamFactory {
+      static readonly IDataStreamMetadata metadata = new DataStreamMetadata() {
+         Name = "LiveBBOStream",
+         AlternateNames = new(),
+         Signatures = new() {
+            new (string?, Type?)[] {
+               ("symbol", typeof(string))
+            },
+            new (string?, Type?)[] {
+               ("secType", typeof(string)),
+               ("symbol", typeof(string)),
+            }
+         }
+      };
+      readonly IMarketDataClient? mdClient;
+      readonly ILogger logger;
+      IDataStreamFactory? compositeFactory = null;
+
+      public LiveBboStreamFactory(IMarketDataClient? mdClient, ILogger logger) {
+         this.mdClient = mdClient;
+         this.logger = logger;
+      }
+
+      public void OnCompositeFactory(IDataStreamFactory compositeFactory) =>
+         this.compositeFactory = compositeFactory;
+
+      public ViewStream? Create(string streamId, IEnumerable<StreamArgument> args) {
+         if (!metadata.IsMatch(streamId)) {
+            return null;
+         }
+         var argList = args.ToList();
+         switch (argList.Count) {
+            case 1: {
+                  var sym = argList[0].Value?.ToString() ?? throw new ArgumentException("Symbol arg is null or empty", nameof(args));
+                  if (sym == "ES" && compositeFactory is not null) {
+                     return compositeFactory.Create("ReplayFutureLiveBBOStream", new (string?, int?, object?)[] { ("contract", 0, sym) });
+                  }
+                  var streamParams = new BookStreamParams() {
+                     Asset = sym
+                  };
+                  return mdClient?.Books(streamParams.GetSecurityFilter())
+                     ?.AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter())
+                     ?.AsViewStream()
+                     ?? throw new InvalidOperationException("Live md not available");
+               }
+            case 2: {
+                  var secTypeStr = argList[0].Value?.ToString() ?? string.Empty;
+                  if (!MarketDataArgStore.TryParseSecurityType(secTypeStr, out var secType)) {
+                     throw new ArgumentException("Failed to parse SecurityType arg", nameof(args));
+                  }
+                  var sym = argList[1].Value?.ToString() ?? throw new ArgumentException("Symbol arg is null or empty", nameof(args));
+                  if (sym == "ES" && secType == bts.api.SecurityType.Future && compositeFactory is not null) {
+                     return compositeFactory.Create("ReplayFutureLiveBBOStream", new (string?, int?, object?)[] { ("contract", 0, sym) });
+                  }
+                  var streamParams = new BookStreamParams() {
+                     Type = secType,
+                     Asset = sym,
+                  };
+                  return mdClient?.Books(streamParams.GetSecurityFilter())
+                     ?.AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter())
+                     ?.AsViewStream()
+                     ?? throw new InvalidOperationException("Live md not available");
+               }
+            default:
+               return null;
+         }
+      }
+
+      public ViewStream? Create(BookStreamParams streamParams) {
+         var secType = streamParams.Type;
+         if (streamParams.Asset == "ES" && (!secType.HasValue || secType.Value == bts.api.SecurityType.Future) && compositeFactory is not null) {
+            return compositeFactory.Create("ReplayFutureLiveBBOStream", new (string?, int?, object?)[] { ("contract", 0, streamParams.Asset) });
+         }
+         if (!checkHasLiveMd()) {
+            return null;
+         }
+         return mdClient.Books(streamParams.GetSecurityFilter())
+            .AsDataPackets(streamParams.GetSchemaBuilder().BboAdapter())
+            .AsViewStream();
+      }
+
+      public IEnumerable<IDataStreamMetadata> GetSupportedStreams() {
+         yield return metadata;
+      }
+
+      [MemberNotNullWhen(true, nameof(mdClient))]
+      bool checkHasLiveMd() =>
+         BookStreamFactory.checkHasLiveMd(mdClient, logger);
+   }
+
+   public class LiveBboStreamParams {
+      [StreamArgument(RegexNames = new[] { WellKnownStreamArgumentAliases.SecurityTypeRegex }, Position = 0, IsRequired = false)]
+      public string? SecurityType { get; set; }
+
+      [StreamArgument(RegexNames = new[] { WellKnownStreamArgumentAliases.AssetRegex }, Position = 1, IsRequired = false)]
+      public string? ContinuousContract { get; set; }
+   }
 }

src/DataStreamSuite/SharedStreams/StreamFactories/BookStreamParams.cs

@@ -0,0 +1,55 @@
+namespace Bts.SharedStreams.StreamFactories;
+
+using bts.api;
+
+public class BookStreamParams {
+   public Venue? Venue { get; set; }
+   public string? Asset { get; set; }
+   public string? Name { get; set; }
+   public string? NameOrAsset { get; set; }
+   public SecurityType? Type { get; set; }
+   public string? SecurityName { get; set; }
+   public int NumLevels { get; set; } = 5;
+   public bool ShowPrice { get; set; } = true;
+   public bool ShowQuantity { get; set; } = true;
+   public bool ShowNotional { get; set; }
+   public bool ShowCashPerPoint { get; set; } = true;
+   public int BufferSize { get; set; } = 10_000;
+
+   public Func<SecurityImmutable, bool> GetSecurityFilter() {
+      return SecurityFilter;
+
+      bool SecurityFilter(SecurityImmutable s) {
+         if (Venue.HasValue && Venue.Value != s.Exchange) {
+            return false;
+         }
+         if (Type.HasValue && Type != s.security_type()) {
+            return false;
+         }
+         if (Asset is not null && Asset != s.Symbol) {
+            return false;
+         }
+         if (Name is not null && Name != s.ToString()) {
+            return false;
+         }
+         if (NameOrAsset is not null && NameOrAsset != s.Symbol && NameOrAsset != s.ToString()) {
+            return false;
+         }
+         return true;
+      }
+   }
+
+   public BookSchemaBuilder GetSchemaBuilder() =>
+      new() {
+         NumLevels = NumLevels,
+         ShowVenue = true,
+         ShowSecurityType = true,
+         ShowAsset = true,
+         ShowSecurityName = true,
+         ShowSecurityId = true,
+         ShowPrice = ShowPrice,
+         ShowQuantity = ShowQuantity,
+         ShowNotional = ShowNotional,
+         ShowCashPerPoint = ShowCashPerPoint,
+      };
+}

src/DataStreamSuite/SharedStreams/StreamFactories/HistoricParquetDataLoader.cs

@@ -2,11 +2,14 @@ namespace Bts.SharedStreams.StreamFactories;
 
 using BCG;
 
+using bts;
 using bts.api;
+using bts.BlueOrb;
 using bts.BlueOrb.Parquet;
 using bts.collections;
 
 using Bts.DataStream;
+using Bts.DataStream.BlueOrb;
 using Bts.DataStreamCore;
 using Bts.DataStreamCore.StreamFactories;
 using Bts.SharedStreams.MarketData;
@@ -15,8 +18,11 @@ using Microsoft.Extensions.Logging;
 
 using Spark.Core.Collections;
 
+using System.Collections;
 using System.Collections.Immutable;
+using System.Diagnostics.CodeAnalysis;
 using System.Reactive.Linq;
+using System.Threading.Tasks;
 
 class HistoricParquetDataLoader : DataStreamFactory {
    public HistoricParquetDataLoader(ViewStreamDatabase viewStreamDatabase, ILogger<HistoricParquetDataLoader> logger) {

src/DataStreamSuite/SharedStreams/StreamFactories/WellKnownColumns.cs

@@ -1,69 +1,71 @@
-namespace Bts.SharedStreams;
+namespace Bts.SharedStreams.StreamFactories;
 
 using bts.api;
 
 using Spark.Api.Orders;
 
+public static class WellKnownColumnNames {
+   public const string Price = "price";
+   public const string Quantity = "qty";
+   public const string Timestamp = "ts_event";
+   public const string BidPrice = "bid_price";
+   public const string BidQuantity = "bid_size";
+   public const string BidNotional = "bid_notl";
+   public const string AskPrice = "ask_price";
+   public const string Currency = "currency";
+   public const string AskQuantity = "ask_size";
+   public const string AskNotional = "ask_notl";
+   public const string SecurityType = "sec_type";
+   public const string Asset = "asset";
+   public const string SecurityName = "sec_name";
+   public const string SecurityId = "SecId";
+   public const string Expiration = "expiration";
+   public const string Venue = "exchange";
+   public const string TradingStatus = "trading_status";
+   public const string NumLegs = "num_legs";
+   public const string CashPerPoint = "cpp";
+
+   public static string DepthPrice(Side side, int level) => $"{side.BidAskName()}Price{level}";
+   public static string DepthQuantity(Side side, int level) => $"{side.BidAskName()}Qty{level}";
+   public static string DepthNotional(Side side, int level) => $"{side.BidAskName()}Notl{level}";
+
+   public static string LegSecurityId(int legNum) => $"Leg{legNum}SecId";
+   public static string LegVenue(int legNum) => $"Leg{legNum}Venue";
+   public static string LegSecurityType(int legNum) => $"Leg{legNum}SecType";
+   public static string LegAsset(int legNum) => $"Leg{legNum}Asset";
+   public static string LegSecurityName(int legNum) => $"Leg{legNum}Sec";
+   public static string LegExpiration(int legNum) => $"Leg{legNum}Exp";
+   public static string LegCashPerPoint(int legNum) => $"Leg{legNum}Cpp";
+}
+
 public static class WellKnownColumns {
-   public static (string Name, Type Type) AskNotional => (WellKnownColumnNames.AskNotional, typeof(double));
-   public static (string Name, Type Type) AskPrice => (WellKnownColumnNames.AskPrice, typeof(double));
-   public static (string Name, Type Type) AskQuantity => (WellKnownColumnNames.AskQuantity, typeof(double));
+   public static (string Name, Type Type) SecurityId => (WellKnownColumnNames.SecurityId, typeof(uint));
+   public static (string Name, Type Type) Venue => (WellKnownColumnNames.Venue, typeof(string));
+   public static (string Name, Type Type) SecurityType => (WellKnownColumnNames.SecurityType, typeof(string));
    public static (string Name, Type Type) Asset => (WellKnownColumnNames.Asset, typeof(string));
-   public static (string Name, Type Type) BidNotional => (WellKnownColumnNames.BidNotional, typeof(double));
-   public static (string Name, Type Type) BidPrice => (WellKnownColumnNames.BidPrice, typeof(double));
-   public static (string Name, Type Type) BidQuantity => (WellKnownColumnNames.BidQuantity, typeof(double));
+   public static (string Name, Type Type) SecurityName => (WellKnownColumnNames.SecurityName, typeof(string));
    public static (string Name, Type Type) CashPerPoint => (WellKnownColumnNames.CashPerPoint, typeof(double));
    public static (string Name, Type Type) Price => (WellKnownColumnNames.Price, typeof(double));
    public static (string Name, Type Type) Quantity => (WellKnownColumnNames.Quantity, typeof(double));
-   public static (string Name, Type Type) SecurityId => (WellKnownColumnNames.SecurityId, typeof(uint));
-   public static (string Name, Type Type) SecurityName => (WellKnownColumnNames.SecurityName, typeof(string));
-   public static (string Name, Type Type) SecurityType => (WellKnownColumnNames.SecurityType, typeof(string));
    public static (string Name, Type Type) Timestamp => (WellKnownColumnNames.Timestamp, typeof(DateTime));
-   public static (string Name, Type Type) Venue => (WellKnownColumnNames.Venue, typeof(string));
+   public static (string Name, Type Type) TimestampNanos => ("ts_event_ns", typeof(long));
+
+   public static (string Name, Type Type) BidPrice => (WellKnownColumnNames.BidPrice, typeof(double));
+   public static (string Name, Type Type) BidQuantity => (WellKnownColumnNames.BidQuantity, typeof(double));
+   public static (string Name, Type Type) BidNotional => ("bid_notl", typeof(double));
+   public static (string Name, Type Type) AskPrice => (WellKnownColumnNames.AskPrice, typeof(double));
+   public static (string Name, Type Type) AskQuantity => (WellKnownColumnNames.AskQuantity, typeof(double));
+   public static (string Name, Type Type) AskNotional => ("ask_notl", typeof(double));
 
-   public static (string Name, Type Type) DepthNotional(Side side, int level) => (WellKnownColumnNames.DepthNotional(side, level), typeof(double));
    public static (string Name, Type Type) DepthPrice(Side side, int level) => (WellKnownColumnNames.DepthPrice(side, level), typeof(double));
    public static (string Name, Type Type) DepthQuantity(Side side, int level) => (WellKnownColumnNames.DepthQuantity(side, level), typeof(double));
+   public static (string Name, Type Type) DepthNotional(Side side, int level) => (WellKnownColumnNames.DepthNotional(side, level), typeof(double));
 
-   public static (string Name, Type Type) LegAsset(int legNum) => (WellKnownColumnNames.LegAsset(legNum), typeof(string));
-   public static (string Name, Type Type) LegCashPerPoint(int legNum) => (WellKnownColumnNames.LegCashPerPoint(legNum), typeof(double));
-   public static (string Name, Type Type) LegExpiration(int legNum) => (WellKnownColumnNames.LegExpiration(legNum), typeof(DateTime));
    public static (string Name, Type Type) LegSecurityId(int legNum) => (WellKnownColumnNames.LegSecurityId(legNum), typeof(uint));
-   public static (string Name, Type Type) LegSecurityName(int legNum) => (WellKnownColumnNames.LegSecurityName(legNum), typeof(string));
-   public static (string Name, Type Type) LegSecurityType(int legNum) => (WellKnownColumnNames.LegSecurityType(legNum), typeof(string));
    public static (string Name, Type Type) LegVenue(int legNum) => (WellKnownColumnNames.LegVenue(legNum), typeof(string));
-}
-
-public static class WellKnownColumnNames {
-   public const string AskNotional = "ask_notl";
-   public const string AskPrice = "ask_price";
-   public const string AskQuantity = "ask_qty";
-   public const string Asset = "asset";
-   public const string BidNotional = "bid_notl";
-   public const string BidPrice = "bid_price";
-   public const string BidQuantity = "bid_qty";
-   public const string CashPerPoint = "cpp";
-   public const string Currency = "currency";
-   public const string Expiration = "exp";
-   public const string NumLegs = "num_legs";
-   public const string Price = "price";
-   public const string Quantity = "qty";
-   public const string SecurityId = "sec_id";
-   public const string SecurityName = "sec_name";
-   public const string SecurityType = "sec_type";
-   public const string Timestamp = "ts_event";
-   public const string TradingStatus = "trd_status";
-   public const string Venue = "venue";
-
-   public static string DepthNotional(Side side, int level) => $"{side.BidAskNameLower()}_notl_{level}";
-   public static string DepthPrice(Side side, int level) => $"{side.BidAskNameLower()}_price_{level}";
-   public static string DepthQuantity(Side side, int level) => $"{side.BidAskNameLower()}_qty_{level}";
-
-   public static string LegAsset(int legNum) => $"leg_{legNum}_{Asset}";
-   public static string LegCashPerPoint(int legNum) => $"leg_{legNum}_{CashPerPoint}";
-   public static string LegExpiration(int legNum) => $"leg_{legNum}_{Expiration}";
-   public static string LegSecurityId(int legNum) => $"leg_{legNum}_{SecurityId}";
-   public static string LegSecurityName(int legNum) => $"leg_{legNum}_{SecurityName}";
-   public static string LegSecurityType(int legNum) => $"leg_{legNum}_{SecurityType}";
-   public static string LegVenue(int legNum) => $"leg_{legNum}_{Venue}";
+   public static (string Name, Type Type) LegSecurityType(int legNum) => (WellKnownColumnNames.LegSecurityType(legNum), typeof(string));
+   public static (string Name, Type Type) LegAsset(int legNum) => (WellKnownColumnNames.LegAsset(legNum), typeof(string));
+   public static (string Name, Type Type) LegSecurityName(int legNum) => (WellKnownColumnNames.LegSecurityName(legNum), typeof(string));
+   public static (string Name, Type Type) LegExpiration(int legNum) => (WellKnownColumnNames.LegExpiration(legNum), typeof(DateTime));
+   public static (string Name, Type Type) LegCashPerPoint(int legNum) => (WellKnownColumnNames.LegCashPerPoint(legNum), typeof(double));
 }

src/client/spark/view/Converters/PartialConverter.cs

@@ -22,7 +22,7 @@ namespace mbts.ui.converters {
                break;
             default:
                if (values[0] is IMultiValueConverter mvc) {
-                  values = ArrayUtils.RemoveAt(values, 0);
+                  ArrayUtils.remove_at(ref values, 0);
                   return mvc.Convert(values, targetType, parameter, culture);
                }
                break;

src/client/spark_api/Orders/OrdersExtensions.cs

@@ -405,7 +405,7 @@ namespace Spark.Api.Orders {
          side switch {
             Side.Buy => "bid",
             Side.Sell => "ask",
-            _ => side.ToString().ToLower(),
+            _ => side.ToString(),
          };
 
       // public static Brush badge_fill(AlgoDefinitionMetadata? algo_meta, OrderMetadata order_meta) =>

src/client/spark_core/Threading/AtomicGate.cs

@@ -10,24 +10,25 @@ using System.Threading;
 using System.Threading.Tasks;
 
 public class AtomicGate {
-   const int maxAttemptsBeforeSleep = 32;
+   const int MAX_ATTEMPTS_BEFORE_SLEEP = 32;
    static readonly int[] increasingSleeps = new[] { 1, 10, 100, 200, 500, 1000 };
-   static int idCounter = 0;
+   static int id_counter = 0;
 
    readonly string name;
-   readonly int id = Interlocked.Increment(ref idCounter);
+   readonly int id = Interlocked.Increment(ref id_counter);
    readonly object syncRoot = new();
    int versionIn = 0;
    int versionOut = 0;
 
-   public AtomicGate(string? name = null) =>
+   public AtomicGate(string? name = null) {
       this.name = name ?? string.Empty;
+   }
 
    public bool IsBusy => Volatile.Read(ref versionIn) != Volatile.Read(ref versionOut);
 
    public void Enter() {
       int n;
-      for (n = 0; n < maxAttemptsBeforeSleep; n++) {
+      for (n = 0; n < MAX_ATTEMPTS_BEFORE_SLEEP; n++) {
          if (TryEnter()) {
             return;
          }
@@ -45,13 +46,13 @@ public class AtomicGate {
    }
 
    public bool TryEnter() {
-      // var vin = Volatile.Read(ref versionIn);
-      // var vout = Volatile.Read(ref versionOut);
-      // Log.DataFlow.diag($$"""[atomic_gate_try_entering.json]{ "id": {{id}}, "name": "{{name}}", "vin": {{vin}}, "vout": {{vout}}, "ts": "{{DateTime.Now}}" }""");
+      var vin = Volatile.Read(ref versionIn);
+      var vout = Volatile.Read(ref versionOut);
+      Log.DataFlow.diag($"[atomic_gate_try_entering.json]{{ \"id\": {id}, \"name\": \"{name}\", \"vin\": {vin}, \"vout\": {vout}, \"ts\": \"{DateTime.Now}\" }}");
       bool res = Impl();
-      // vin = Volatile.Read(ref versionIn);
-      // vout = Volatile.Read(ref versionOut);
-      // Log.DataFlow.diag($$"""[atomic_gate_try_entered.json]{ "id": {{id}}, "name": "{{name}}", "vin": {{vin}}, "vout": {{vout}}, "res": {{res}}, "ts": "{{DateTime.Now}}" }""");
+      vin = Volatile.Read(ref versionIn);
+      vout = Volatile.Read(ref versionOut);
+      Log.DataFlow.diag($"[atomic_gate_try_entered.json]{{ \"id\": {id}, \"name\": \"{name}\", \"vin\": {vin}, \"vout\": {vout}, \"res\": {res}, \"ts\": \"{DateTime.Now}\" }}");
       return res;
 
       bool Impl() {
@@ -115,7 +116,7 @@ public class AtomicGate {
       int version = Interlocked.Increment(ref versionIn) - 1;
       int n = 0;
       var end = Disposable.Create(EndTransaction);
-      for (n = 0; n < maxAttemptsBeforeSleep; n++) {
+      for (n = 0; n < MAX_ATTEMPTS_BEFORE_SLEEP; n++) {
          if (Volatile.Read(ref versionOut) == version) {
             return end;
          }
@@ -147,10 +148,9 @@ public class AtomicGate {
    }
 
    void sleep(int sleepMillis, int? version = null) {
-      // var vout = Volatile.Read(ref versionOut);
-      // var vin = Volatile.Read(ref versionIn);
-      // Log.DataFlow.warn($$"""[atomic_gate_sleep.json]{ "id": {id}, "name": "{{name}}", "vin": {vin}, "vout": {{vout}}, "v": {{version}}, "dur": {{sleepMillis}}ms, "ts": "{{DateTime.Now}}" }""");
+      var vout = Volatile.Read(ref versionOut);
+      var vin = Volatile.Read(ref versionIn);
+      Log.DataFlow.warn($"[atomic_gate_sleep.json]{{ \"id\": {id}, \"name\": \"{name}\", \"vin\": {vin}, \"vout\": {vout}, \"v\": {version}, \"dur\": {sleepMillis}ms, \"ts\": \"{DateTime.Now}\" }}");
       Thread.Sleep(sleepMillis);
    }
 }
-