check pt: StreamFactoryLogs.cs

Also change snake_case fn names to PascalCase

src/DataStreamSuite/DataStream/BlueOrb/TimeSeriesStream.cs

@@ -47,8 +47,6 @@ 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;
@@ -60,19 +58,21 @@ 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.Add(new(id, ar));
+            addedRowsBd.AddOrUpdate(ar);
             if (oldRow is not null) {
-               deletedRowsBd.Add(new(oldRow.Id(), oldRow));
+               deletedRowsBd.AddOrUpdate(oldRow);
             }
          }
-         var addedRows = new BlueTableImmutable(_table.schema, addedRowsBd.ToImmutable());
-         var deletedRows = new BlueTableImmutable(_table.schema, deletedRowsBd.ToImmutable());
+         var addedRows = addedRowsBd.ToImmutable();
+         var deletedRows = deletedRowsBd.ToImmutable();
          var dp_out = new DataPacket(_table, addedRows, deletedRows);
          myFireEvent(dp_out);
       }

src/DataStreamSuite/DataStream/DataStreamExtensions.cs

@@ -8,7 +8,6 @@ using bts.BlueOrb;
 
 using Bts.DataStreamCore;
 using Bts.PubSub;
-using Bts.SharedStreams.StreamFactories;
 
 using mbts.utils;
 
@@ -139,7 +138,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.Any() || t.Dp.deletedRows.Any()))
+         .Where(t => t.Dp is not null && (t.Dp.addedRows.Count > 0 || t.Dp.deletedRows.Count > 0))
          .Select(t => t.Dp!);
 
    public static IObservable<IDataPacket> AsDataPackets<T>(this IObservable<TableUpdate<T, int>> source, IBlueTableSchema schema) where T : IBlueRow =>
@@ -156,7 +155,7 @@ public static class DataStreamExtensions {
             var resDp = bd.ToDataPacket(out var newRows);
             return (newRows, resDp);
          })
-         .Where(t => t.Dp.addedRows.Any() || t.Dp.deletedRows.Any())
+         .Where(t => t.Dp.addedRows.Count > 0 || t.Dp.deletedRows.Count > 0)
          .Select(t => t.Dp);
 
    public static IObservable<IDataPacket> AsDataPackets<T, TKey>(this IObservable<TableUpdate<T, TKey>> source, BlueTableSchemaAdapter<T> schema) where TKey : IEquatable<TKey> {
@@ -171,9 +170,8 @@ public static class DataStreamExtensions {
             (BlueTableImmutableAdapter<T> Adapter, IDataPacket DataPacket) state, TableUpdate<T, TKey> tableUpdate) {
             var bd = state.Adapter.ToDataPacketBuilder();
             bd.Update(tableUpdate);
-            var dp = bd.ToDataPacket();
-            var tableAdapter = bd.ToImmutable();
-            return (bd.ToImmutable(), dp);
+            var newRows = bd.ToImmutable(out var dp);
+            return (newRows, dp);
          }
       }
    }
@@ -197,6 +195,46 @@ 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

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

src/DataStreamSuite/DataStreamCore/DynamicLoading/LiveMarketDataStreamAttribute.cs

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

src/DataStreamSuite/DataStreamCore/DynamicLoading/MarketDataClientParamAttribute.cs

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

src/DataStreamSuite/DataStreamCore/DynamicLoading/MarketDataStreamFactory.cs

@@ -0,0 +1,49 @@
+#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,9 +8,15 @@ 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 Symbol = "symbol,sym,product,asset";
+   public const string Asset = "symbol,sym,product,asset";
 
    /// <summary>
    /// Aliases for a Type argument.
@@ -29,21 +29,19 @@ public static class WellKnownStreamArgumentAliases {
    /// </summary>
    public const string SecurityName = "name,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 QualifiedSecurityType = "secType,secTyp,securityType,securityTyp";
+   public const string QualifiedSecurityName = "sec,secName,security";
 
    public const string VenueRegex = "^(venue|exchange|exch)$";
-   public const string AssetRegex = "^(symbol|sym|product|asset)$";
+   public const string AssetRegex = "^(symbol|sym|asset|product)$";
    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 ShowSecurityTypeRegex = "^(show|include)(Sec|Security)?(Type|Typ)$";
    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 ShowExpirationRegex = "^(show|include)(Exp(iration|Date)?)$";
    public const string ShowCashPerPointRegex = "^(show|include)(CashPerPoint|Cpp)$";
 
@@ -55,7 +53,6 @@ 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,6 +3,7 @@ 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;
@@ -116,6 +117,8 @@ 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();
@@ -124,6 +127,8 @@ namespace BCG {
          else
             _workers.RemoveFirst(tup => tup.Worker == w);
       }
+      void IWorkScheduler.Remove(IWorker worker) =>
+         removeWorker(worker);
 
       internal static void SetGlobalInstance(ViewStreamDatabase vsd) {
          lock (myInstanceLockObject) {
@@ -132,12 +137,6 @@ 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 class DataStreamPublisherService : IHostedService, IDisposable {
+public sealed 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://localhost:8000/config"
+    "ConfigServer": "http://127.0.0.1:8000/config"
   },
   "SparkApi": {
     "Credentials": {

src/DataStreamSuite/DataStreamTests/DataPacketBuilder/DataPacketBuilderTests.cs

@@ -186,8 +186,7 @@ public class DataPacketBuilderTests {
    }
 
    static BlueTableImmutable getTable(params (int Id, double Value)[] rows) =>
-      new(schema, ImmutableSortedSet.CreateRange(
-         rows.Select(tup => new KeyValuePair<int, IBlueRow>(tup.Id, getBlueRow(tup)))));
+      BlueTableImmutable.CreateRange(schema, rows.Select(tup => new KeyValuePair<int, IBlueRow>(tup.Id, getBlueRow(tup))));
 
    static MutableBlueRow getBlueRow((int id, double value) tup) =>
       getBlueRow(tup.id, tup.value);
@@ -205,11 +204,10 @@ 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(
-         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(newRows))),
-         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(addedRows))),
-         new BlueTableImmutable(schema, ImmutableSortedSet.CreateRange(kvpComp, GetRowKvps(removedRows))));
+         BlueTableImmutable.CreateRange(schema, GetRowKvps(newRows)),
+         BlueTableImmutable.CreateRange(schema, GetRowKvps(addedRows)),
+         BlueTableImmutable.CreateRange(schema, 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,3 +84,4 @@ public class WorkScheduler : IWorkScheduler, IDisposable {
       }
    }
 }
+

src/DataStreamSuite/PubSub/Misc/WorkSchedulerExtensions.cs

@@ -35,4 +35,10 @@ 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/BookSchemaBuilder.cs

@@ -1,25 +1,26 @@
-namespace Bts.SharedStreams.StreamFactories;
+namespace Bts.SharedStreams.MarketData;
 
 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; }
-   public bool ShowNotional { get; set; }
-   public bool ShowQuantity { get; set; } = true;
-   public bool ShowPrice { get; set; } = true;
-   public bool ShowVenue { 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; }
-   public bool ShowSecurityName { get; set; }
+   public bool ShowSecurityType { get; set; } = true;
+   public bool ShowSecurityName { get; set; } = true;
    public bool ShowSecurityId { get; set; } = true;
-   public bool ShowCashPerPoint { get; set; }
+   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 string TableName { get; set; } = "Book";
 
    public IBlueTableSchema ToSchema() {
@@ -29,12 +30,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          bd.Add(WellKnownColumns.Venue);
       }
-      if (ShowSecurityType) {
-         bd.Add(WellKnownColumns.SecurityType);
-      }
       if (ShowAsset) {
          bd.Add(WellKnownColumns.Asset);
       }
+      if (ShowSecurityType) {
+         bd.Add(WellKnownColumns.SecurityType);
+      }
       if (ShowSecurityName) {
          bd.Add(WellKnownColumns.SecurityName);
       }
@@ -66,12 +67,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          adapter.AddColumn(WellKnownColumns.Venue.Name, b => b.security.Exchange.ToString());
       }
-      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 (ShowSecurityType) {
+         adapter.AddColumn(WellKnownColumns.SecurityType.Name, b => b.security.security_type().ToString());
+      }
       if (ShowSecurityName) {
          adapter.AddColumn(WellKnownColumns.SecurityName.Name, b => b.security.ToString());
       }
@@ -103,12 +104,12 @@ public class BookSchemaBuilder {
       if (ShowVenue) {
          adapter.AddColumn(WellKnownColumns.Venue.Name, b => b.security.Exchange.ToString());
       }
-      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 (ShowSecurityType) {
+         adapter.AddColumn(WellKnownColumns.SecurityType.Name, b => b.security.security_type().ToString());
+      }
       if (ShowSecurityName) {
          adapter.AddColumn(WellKnownColumns.SecurityName.Name, b => b.security.ToString());
       }
@@ -148,3 +149,4 @@ 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/BlueOrb/MarketData/BookStreamParams.cs

@@ -0,0 +1,81 @@
+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,26 +1,35 @@
-namespace bts.BlueOrb
-   {
-   using System;
-
+namespace bts.BlueOrb {
    using BCG;
 
-   using bts.api;
+   using Bts.DataStream;
+   using Bts.DataStreamCore;
+   using Bts.SharedStreams.MarketData;
 
-   [DataStreamUserRespresentation(false)]
+   using Spark.Api.MarketData;
+
+   [LiveMarketDataStream]
+   [DataStreamUserRespresentation("LiveBBOStream")]
    public class LiveBBOStream : ViewStream {
-      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;
+      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;
          }
-         mySetRealSource(rs);
+         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());
       }
    }
 }

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 secType from string '{str}'";
+         error = $"Failed to parse SecurityType from string '{str}'";
       } else {
-         error = $"Venue parser received input of unexpected type '{val?.GetType()}'";
+         error = $"SecurityType 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 v) {
+   public static bool TryParse(string? str, out Venue result) {
       if (str is null) {
-         v = Venue.Unset;
+         result = Venue.Unset;
          return false;
       }
-      if (Enum.TryParse(str, true, out v) || Enum.TryParse("VENUE_" + str, true, out v)) {
+      if (Enum.TryParse(str, true, out result) || Enum.TryParse("VENUE_" + str, true, out result)) {
          return true;
       }
-      v = str.ToUpper() switch {
+      result = str.ToUpper() switch {
          "CME" or "XCME" or "CEC" or "XCEC" => Venue.Glbx,
          "CBOT" or "CBT" or "XCBT" => Venue.Cboe,
          _ => Venue.Unset,
       };
-      return v != Venue.Unset;
+      return result != Venue.Unset;
    }
 
    public bool CanParse(StreamArgument arg, Type outputType) =>

src/DataStreamSuite/SharedStreams/BlueOrb/WellKnownColumns.cs

@@ -1,71 +1,69 @@
-namespace Bts.SharedStreams.StreamFactories;
+namespace Bts.SharedStreams;
 
 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) 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) 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) Asset => (WellKnownColumnNames.Asset, typeof(string));
-   public static (string Name, Type Type) SecurityName => (WellKnownColumnNames.SecurityName, 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) 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) 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) Venue => (WellKnownColumnNames.Venue, typeof(string));
 
+   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) LegSecurityId(int legNum) => (WellKnownColumnNames.LegSecurityId(legNum), typeof(uint));
-   public static (string Name, Type Type) LegVenue(int legNum) => (WellKnownColumnNames.LegVenue(legNum), typeof(string));
-   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));
+   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}";
 }

src/DataStreamSuite/SharedStreams/StreamFactories/BookStreamFactory.cs

@@ -2,202 +2,113 @@ 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, IRecursiveDataStreamFactory {
-   readonly LiveBboStreamFactory liveBboFactory;
-   readonly IMarketDataClient? mdClient;
+public class BookStreamFactory : DataStreamFactory {
+   readonly VenueArgumentParser venueParser = VenueArgumentParser.Instance;
+   readonly SecurityTypeArgumentParser secTypeParser = SecurityTypeArgumentParser.Instance;
+   readonly StringArgumentParser strParser = StringArgumentParser.Instance;
+   readonly IMarketDataClient mdClient;
    readonly ILogger logger;
    readonly IWorkScheduler workScheduler;
-   IDataStreamFactory? compositeFactory = null;
 
-   public BookStreamFactory(IMarketDataClient? mdClient, ILogger<BookStreamFactory> logger, ViewStreamDatabase workScheduler) {
+   public BookStreamFactory(IMarketDataClient mdClient, ViewStreamDatabase viewStreamDatabase, ILogger<BookStreamFactory> logger)
+      : this(mdClient, viewStreamDatabase, (ILogger)logger) { }
+
+   BookStreamFactory(IMarketDataClient mdClient, IWorkScheduler workScheduler, ILogger logger) {
       this.mdClient = mdClient;
       this.logger = logger;
       this.workScheduler = workScheduler;
-      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);
-      }
+      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);
    }
 
-   public void OnCompositeFactory(IDataStreamFactory compositeFactory) {
-      this.compositeFactory = compositeFactory;
-      liveBboFactory.OnCompositeFactory(compositeFactory);
-   }
+   ViewStream? createBook(Venue venue) =>
+      createBook(new BookStreamParams() { Venue = venue });
 
-   ViewStream? createBboPriceSummary(BookStreamParams streamParams) {
-      if (!checkHasLiveMd()) {
+   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");
          return null;
       }
-      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 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 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

@@ -1,55 +0,0 @@
-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,14 +2,11 @@ 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;
@@ -18,11 +15,8 @@ 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/client/spark/view/Converters/PartialConverter.cs

@@ -22,7 +22,7 @@ namespace mbts.ui.converters {
                break;
             default:
                if (values[0] is IMultiValueConverter mvc) {
-                  ArrayUtils.remove_at(ref values, 0);
+                  values = ArrayUtils.RemoveAt(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(),
+            _ => side.ToString().ToLower(),
          };
 
       // public static Brush badge_fill(AlgoDefinitionMetadata? algo_meta, OrderMetadata order_meta) =>

src/client/spark_core/Threading/AtomicGate.cs

@@ -10,25 +10,24 @@ using System.Threading;
 using System.Threading.Tasks;
 
 public class AtomicGate {
-   const int MAX_ATTEMPTS_BEFORE_SLEEP = 32;
+   const int maxAttemptsBeforeSleep = 32;
    static readonly int[] increasingSleeps = new[] { 1, 10, 100, 200, 500, 1000 };
-   static int id_counter = 0;
+   static int idCounter = 0;
 
    readonly string name;
-   readonly int id = Interlocked.Increment(ref id_counter);
+   readonly int id = Interlocked.Increment(ref idCounter);
    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 < MAX_ATTEMPTS_BEFORE_SLEEP; n++) {
+      for (n = 0; n < maxAttemptsBeforeSleep; n++) {
          if (TryEnter()) {
             return;
          }
@@ -46,13 +45,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() {
@@ -116,7 +115,7 @@ public class AtomicGate {
       int version = Interlocked.Increment(ref versionIn) - 1;
       int n = 0;
       var end = Disposable.Create(EndTransaction);
-      for (n = 0; n < MAX_ATTEMPTS_BEFORE_SLEEP; n++) {
+      for (n = 0; n < maxAttemptsBeforeSleep; n++) {
          if (Volatile.Read(ref versionOut) == version) {
             return end;
          }
@@ -148,9 +147,10 @@ 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);
    }
 }
+