check pt

Add config for PubSub tag handler factory
devcontainer: Install dotnet8 expose ports for DataStream Web App
2024-07-07 22:19:25 +00:00 · 2024-07-05 13:50:39 +00:00 · 2024-07-04 17:33:37 +00:00 · 2024-07-04 17:33:37 +00:00 · 2024-07-04 17:33:37 +00:00 · 2024-07-04 17:33:37 +00:00
45 changed files with 976 additions and 985 deletions
@@ -1,12 +1,16 @@
 using BCG.Utilities.General;

+using Bts.PubSub;
+
+using Microsoft.Extensions.DependencyInjection;
+
 using System;

 namespace BCG {
-    public class TCPConnectionTestStream : ViewStream, IWorker, ITCPConnectionHandler, ITCPMessageHandler {
+    public class TCPConnectionTestStream : ViewStream, IParameterizedWorker, ITCPConnectionHandler, ITCPMessageHandler {
        class State {
-            public int n_sends = 0;
-            public int n_recvs = 0;
+           public int n_sends = 0;
+           public int n_recvs = 0;
        }
        private readonly string _ip;
        private readonly int _port;
@@ -17,7 +21,8 @@ namespace BCG {
        private TimeSpan _send_period;
        private ITCPConnection _connection = null;
        private byte[] _buff;
-
+        WorkActivator workActivator;
+     
        public TCPConnectionTestStream(string ip, int port, int msg_size, int send_period_seconds) {
            _ip = ip;
            _port = port;
@@ -31,16 +36,21 @@ namespace BCG {
            for (int i = 0; i < _buff.Length; i++) {
                _buff[i] = (byte)i;
            }
+           workActivator = new(this);
+        }
+
+        TimeSpan IParameterizedWorker.Interval => _send_period;
+
+        protected override void OnServiceProvider(IServiceProvider serviceProvider) {
+           workActivator.OnWorkScheduler(serviceProvider.GetRequiredService<IWorkScheduler>());
        }

        protected override void myOnFirstSubscriber() {
-            MasterWorker.getInstance(false).addWorker(this, _send_period);
-            var server_rci = new RemoteConnectionInfo(_ip, _port);
-            bts.AsyncTCPConnector_APM.connect_to(server_rci, this);
+           workActivator.Start();
        }

        protected override void myOnLastSubscriber() {
-            MasterWorker.getInstance(false).removeWorker(this);
+            workActivator.Stop();
        }

        public void doWork() {
@@ -56,18 +66,18 @@ namespace BCG {
        }

        public void onConnection(ITCPConnection connection) {
-            _connection = connection;
-            _connection.register(this);
+           _connection = connection;
+           _connection.register(this);
        }
        public bool isDisposed { get; }

        public void onData(RemoteMessage msg) {
-            bts.Log.Gen.info($"TCPConnectionTestStream received msg: {msg}");
-            _state.n_recvs++;
+           bts.Log.Gen.info($"TCPConnectionTestStream received msg: {msg}");
+           _state.n_recvs++;
        }

        public void Dispose() {
-
+            workActivator.Stop();
        }


@@ -1,178 +0,0 @@
-using BCG;
-using BCG.Utilities.General;
-using System.Collections.Generic;
-
-//using MongoDB.Driver;
-
-using System;
-using System.IO;
-using System.Linq;
-using System.Runtime.CompilerServices;
-using bts.api;
-
-namespace bts {
-   [DataStreamUserRespresentation("pss_echo")]
-   public class PubsubEchoTestStream : ViewStream, IWorker {
-      public class PayloadMessage {
-         public string guid;
-         public Int64 timestamp = 0;
-         public byte[] payload = null;
-         public PayloadMessage(string guid_, Int64 timestamp_, int payload_size, byte payload_byte) {
-            guid = guid_;
-            timestamp = timestamp_;
-            payload = new byte[payload_size];
-            for (int i = 0; i < payload_size; i++) {
-               payload[i] = payload_byte;
-            }
-         }
-         private PayloadMessage() { }
-         public byte[] serialize() {
-            var str = new MemoryStream();
-            BinaryWriter bw = new BinaryWriter(str);
-            bw.Write(guid);
-            bw.Write(timestamp);
-            bw.Write(payload.Length);
-            bw.Write(payload);
-            str.Position = 0;
-            var bytes = str.ToArray();
-            return bytes;
-         }
-
-         public static PayloadMessage deserialize(byte[] ser_bytes) {
-            var str = new MemoryStream(ser_bytes);
-            BinaryReader br = new BinaryReader(str);
-            string guid = br.ReadString();
-            Int64 timestamp_ = br.ReadInt64();
-            int len = br.ReadInt32();
-            var payload = br.ReadBytes(len);
-            return new PayloadMessage() {
-               guid = guid,
-               timestamp = timestamp_,
-               payload = payload
-            };
-         }
-      }
-      //public void
-      private readonly int _msg_size;
-      private readonly TimeSpan _publish_period;
-      private DateTime _last_publish_time = DateTime.MinValue;
-      private PublisherCore _pub_core;
-      private ISubscriberCore _sub_core;
-      private BlueTable _table;
-      private int _n_pubs = 0;
-      private readonly int _max_pubs;
-      private int _n_updates = 0;
-      private ViewStreamDatabase _vsd;
-      private PayloadMessage _last_sent_msg, _last_recvd_msg;
-      private const byte _payload_byte = (byte)'A';
-      private readonly string _tag_name;
-      private readonly int _payload_sz;
-      private readonly IPubSubRouter _router;
-      private enum IOState {
-         SEND = 0,
-         POLL = 1
-      };
-      IOState _state = IOState.SEND;
-
-      public PubsubEchoTestStream(string server_url, TimeSpan publish_period, int payload_size, string tag_name, ViewStreamDatabase vsd) {
-         var route = new PubSubRoute(server_url, ".*");
-         _router = new PubSubRouter(route);
-         _vsd = vsd;
-         _payload_sz = payload_size;
-         var schema = BlueDataUtility.form_schema("PubSubTest",
-         Tuple.Create("id", TypeCode.Int32),
-         Tuple.Create("sent_guid", TypeCode.String),
-         Tuple.Create("sent_time", TypeCode.Int64),
-         Tuple.Create("recv_guid", TypeCode.String),
-         Tuple.Create("recv_time", TypeCode.Int64),
-         Tuple.Create("match", TypeCode.Boolean),
-         Tuple.Create("echo_time", TypeCode.Int32));
-         _table = new BlueTable(schema);
-         _publish_period = publish_period;
-         _tag_name = tag_name;
-      }
-
-      public PubsubEchoTestStream(string server_url, int publish_period_msecs, int payload_size, string tag_name)
-      : this(server_url, TimeSpan.FromMilliseconds(publish_period_msecs), payload_size, tag_name, ViewStreamDatabase.instance) { }
-
-      protected override void myOnFirstSubscriber() {
-         _pub_core = new PublisherCore(_router);
-         _sub_core = new SubscriberCore(_router);
-         _sub_core.subscribe(_tag_name);
-         _vsd.addWorker(this);
-
-      }
-
-      protected override void myOnLastSubscriber() {
-         _vsd.removeWorker(this);
-         _sub_core.unsubscribe(_tag_name);
-      }
-
-      private void _update_table_and_fire_dp() {
-         if (_last_sent_msg == null) {
-            return;
-         }
-         int id = _table.Count;
-         var sent_guid = _last_sent_msg.guid;
-         var sent_time = _last_sent_msg.timestamp;
-         if (_last_recvd_msg == null) {
-            BlueRow r = new BlueRow(_table.schema, id, sent_guid, sent_time, DBNull.Value, DBNull.Value,
-            DBNull.Value, DBNull.Value);
-            _table.addRow(r);
-         } else {
-            BlueRow r = new BlueRow(_table.schema, id, sent_guid, sent_time, _last_recvd_msg.guid, _last_recvd_msg.timestamp, String.Compare(_last_recvd_msg.guid, _last_sent_msg.guid),
-               epoch_now() - _last_sent_msg.timestamp);
-            _table.addRow(r);
-         }
-         var dp_out = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
-         myFireEvent(dp_out);
-      }
-
-      private void _publish_next_packet() {
-         _last_sent_msg = new PayloadMessage(Guid.NewGuid().ToString(), epoch_now(), _payload_sz, _payload_byte);
-         var ser_bytes = _last_sent_msg.serialize();
-         _n_pubs++;
-         _pub_core.publish("Blob", _tag_name, ser_bytes);
-         _last_publish_time = DateTime.Now;
-      }
-
-      public void doWork() {
-         Func<IEnumerable<PubSubServerMessage>, PayloadMessage> find_packet = (updates) => {
-            foreach (var u in updates) {
-               var recv_msg = PayloadMessage.deserialize(u.data);
-               if (recv_msg.guid == _last_sent_msg.guid) {
-                  return recv_msg;
-               }
-            }
-            return null;
-         };
-         switch (_state) {
-            case IOState.SEND:
-               if (DateTime.Now - _last_publish_time >= _publish_period) {
-                  _publish_next_packet();
-                  _state = IOState.POLL;
-                  _last_publish_time = DateTime.Now;
-                  _update_table_and_fire_dp();
-               }
-               break;
-            case IOState.POLL:
-               var updates = _sub_core.get_updates(TimeSpan.Zero);
-               updates = updates.Where(u => u.tag == _tag_name);
-               var recv_msg = find_packet(updates);
-               if (recv_msg != null) {
-                  recv_msg.timestamp = epoch_now();
-                  _last_recvd_msg = recv_msg;
-                  _state = IOState.SEND;
-                  _update_table_and_fire_dp();
-                  break;
-               }
-               break;
-         }
-      }
-
-      public static long epoch_now() {
-         return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-      }
-
-   }
-}
@@ -24,6 +24,7 @@ namespace bts {
            var dp_out = BlueDataUtility.build_dp_inc(_rows, myLastValue as DataPacket);
            myFireEvent(dp_out);

+
        }


@@ -1,54 +1,6 @@
 using System;

-using BCG.Utilities.General;
-
-using bts;
-
 namespace BCG {
-   [DataStreamUserRespresentation("milliseconds")]
-   public class MillisecondStream : ViewStream, IWorker {
-      private DateTime _last_fire_time = DateTime.MinValue;
-      private readonly TimeSpan _period;
-      private readonly int _size;
-      private readonly ViewStreamDatabase _vsd;
-      private IBlueTableSchema _schema;
-      private bts.CircularBufferTable _table;
-
-      public MillisecondStream(int interval_msecs, int size)
-      : this(ViewStreamDatabase.instance, interval_msecs, size) {
-      }
-
-      public MillisecondStream(ViewStreamDatabase vsd, int interval_msecs, int size) {
-         _vsd = vsd;
-         _period = TimeSpan.FromMilliseconds(interval_msecs);
-         _size = size;
-         _schema = BlueDataUtility.form_schema("Milliseconds",
-               Tuple.Create("id", TypeCode.Int32),
-               Tuple.Create("time", TypeCode.DateTime),
-               Tuple.Create("display_time", TypeCode.String));
-         _table = new CircularBufferTable(_schema, _size);
-      }
-
-      protected override void myOnFirstSubscriber() {
-         _vsd.addWorker(this);
-      }
-
-      protected override void myOnLastSubscriber() {
-         _vsd.removeWorker(this);
-      }
-
-      private int _id = 0;
-      public void doWork() {
-         if (DateTime.Now - _last_fire_time >= _period) {
-            _last_fire_time = DateTime.Now;
-            IBlueRow nr = new BlueRow(_schema, _id++, _last_fire_time,
-            _last_fire_time.ToString("yyyy-MM-dd HH:mm:ss.fff"));
-            _table.add_row(nr);
-            var dp = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
-            myFireEvent(dp);
-         }
-      }
-   }
   public class MSecondStream : TimeUnitStream {
      public MSecondStream(int intervalInms, int size)
         : base(DateTime.Now, new TimeSpan(0, 0, 0, 0, intervalInms), size) { }
@@ -2,6 +2,8 @@ using System;

 using BCG.Utilities.General;

+using Bts.DataStreamCore;
+
 namespace BCG {
   public abstract class TimeUnitStream : ViewStream, IWorker {
      protected TimeSpan myOffset;
@@ -31,6 +33,9 @@ namespace BCG {
      #endregion

      protected virtual void myInit() {
+         if (!this.HasSubscribers()) {
+            return;
+         }
         int periods = (int)
            (DateTime.Now - myConstructionTime).TotalMilliseconds / (int)myOffset.TotalMilliseconds;

@@ -16,6 +16,12 @@ namespace bts {
         _buffer_sz = buffer_sz;
      }

+      // public OrderedBlueTable(List<IBlueRow> rows, int buffer_sz) {
+      //    _buffer_sz = buffer_sz;
+      //    foreach (IBlueRow r in rows)
+      //       add_row(r);
+      // }
+
      // returns the row that might need to be deleted to keep
      // the circular buffer intact
      public IBlueRow add_row(IBlueRow row) {
@@ -0,0 +1,8 @@
+namespace Bts.DataStreamCore;
+
+using BCG;
+
+public static class DataStreamCoreExtensions {
+   public static bool HasSubscribers(this DataStream dataStream) =>
+      dataStream.numberOfSubscribers > 0;
+}
@@ -50,13 +50,13 @@ namespace BCG {
      public virtual void make_batch_changes(IBlueRowCollection new_rows,
                                             IBlueRowCollection del_rows,
                                             IBlueRowCollection added_rows) {
-         // string log_msg = $"PubTableBroker.make_batch_changes table: {new_rows.schema.tableName} del: {del_rows.Count} added: {added_rows.Count}";
-         // foreach (IBlueRow r in del_rows)
-         //    log_msg += $"id: {r["id"]} ";
-         // log_msg += "addedRows: ";
-         // foreach (IBlueRow r in added_rows)
-         //    log_msg += $"id: {r["id"]}";
-         // bts.Log.Gen.info(log_msg);
+         string log_msg = $"PubTableBroker.make_batch_changes table: {new_rows.schema.tableName} del: {del_rows.Count} added: {added_rows.Count}";
+         foreach (IBlueRow r in del_rows)
+            log_msg += $"id: {r["id"]} ";
+         log_msg += "addedRows: ";
+         foreach (IBlueRow r in added_rows)
+            log_msg += $"id: {r["id"]}";
+         bts.Log.Gen.info(log_msg);
         SinglePubTableBroker table_publisher = null;
         var name_key = new_rows.schema.tableName.ToLower();
         if (!_publisher_map.TryGetValue(name_key, out table_publisher))
@@ -5,21 +5,18 @@ using System.Linq;

 using bts;

-namespace BCG
-{
+namespace BCG {
   // This class is responsible for mantaining the state fo a table. It gets the
   // bytes from the pubsubserver via the handle_updates call (called by VSD), deserializes
   // updates the state of the table, and fires the TableUpdateDelegate.
-   public class TableAccumulator2 : ITableAccumulator
-   {
+   public class TableAccumulator2 : ITableAccumulator {
      private readonly string _tag;
      private readonly ISubscriberCore _subscriber_core;
      IStreamMsgDeserilizer _deser = new StreamIODeserializationController();
      private BlueTable _table;
      private TableUpdateDelegate _on_update_delegate;

-      public TableAccumulator2(string tag, ISubscriberCore sub_core)
-      {
+      public TableAccumulator2(string tag, ISubscriberCore sub_core) {
         _tag = tag;
         _subscriber_core = sub_core;
      }
@@ -28,13 +25,10 @@ namespace BCG

      //public bool is_active() { return _on_update_delegate != null; }

-      public event TableUpdateDelegate on_update
-      {
-         add
-         {
+      public event TableUpdateDelegate on_update {
+         add {
            Log.Gen.info("TableAccumulator2.on_update.add {_tag} subscribed");
-            if (_on_update_delegate == null)
-            {
+            if (_on_update_delegate == null) {
               //ie first subscription
               Log.Gen.info("TableAccumulator2.on_update {_tag} subscribed for the first time");
               _subscriber_core.subscribe(_tag);
@@ -44,17 +38,14 @@ namespace BCG
            if (_table != null)
               value(_table, _table, new BlueTable(_table.schema));
         }
-         remove
-         {
+         remove {
            Log.Gen.info("TableAccumulator2.on_update.remove {_tag} unsubscribed");
-            if (_on_update_delegate != null)
-            {
+            if (_on_update_delegate != null) {
               _on_update_delegate =
                  Delegate.Remove(_on_update_delegate, value) as TableUpdateDelegate;
               int n_subs = _on_update_delegate == null ? 0 : _on_update_delegate.GetInvocationList().Length;
               Log.Gen.info("TableAccumulator2.on_update.remove {_tag} unsubscribed n_subs: {n_subs}");
-               if (_on_update_delegate == null)
-               {
+               if (_on_update_delegate == null) {
                  _subscriber_core.unsubscribe(_tag);
                  // TODO
                  //_table = null;
@@ -64,19 +55,16 @@ namespace BCG
         }
      }
      private int _update_no = 0;
-      public void handle_updates(IEnumerable<byte[]> msgs)
-      {
+      public void handle_updates(IEnumerable<byte[]> msgs) {
         if (msgs == null || msgs.Count() == 0)
            return;
         Log.Gen.info($"TableAccumulator2.handle_updates _update_no: {_update_no++} msgs: {msgs.Count()} msg_0: {msgs.ElementAt(0).Length}");
         IBlueRowCollection table_0 = _table == null ? null : new BlueTable(_table);
-         foreach (var msg in msgs)
-         {
+         foreach (var msg in msgs) {
            bts.StreamIOMessage str_msg = _deser.deserialize(msg);
-            var state_txt = _table == null ? "NULL" : _table.Count.ToString();// BlueDataUtility.id_string(_table);
+            var state_txt = _table == null ? "NULL" : BlueDataUtility.id_string(_table);
            Log.Gen.info($"TableAccumulator2.handle_update tag: {_tag} sz: {msg.Length} str_msg: {str_msg} _state: {state_txt}");
-            switch (str_msg.op)
-            {
+            switch (str_msg.op) {
               case bts.StreamIOMessage.StreamOp.CREATE:
                  _table = new BlueTable(str_msg.create_message.rows);
                  break;
@@ -89,13 +77,11 @@ namespace BCG
         // this could be made faster if we just do this on the CREATE  message, and pass-through
         // what we get on the MULTIPLE_REMOVE_ADD
         IBlueRowCollection d_rows, a_rows;
-         //BlueDataUtility.compute_differences_id_based(_table, table_0, out d_rows, out a_rows);
-         BlueDataUtility.compute_differences_fast(_table, table_0, out d_rows, out a_rows);
+         BlueDataUtility.compute_differences_id_based(_table, table_0, out d_rows, out a_rows);
         _on_update_delegate(_table, a_rows, d_rows);
      }

-      public void update(Stream[] streams)
-      {
+      public void update(Stream[] streams) {
      }

      #endregion
@@ -95,8 +95,8 @@ namespace bts {
      }

      public static IBlueTableSchema deserialize_schema(ref BinaryReader br) {
-         // var hex_txt = to_hex(br.BaseStream);
-         // bts.Log.Gen.info($"StreamSerializationUtils.deserialize_schema bytes: {hex_txt}");
+         var hex_txt = to_hex(br.BaseStream);
+         bts.Log.Gen.info($"StreamSerializationUtils.deserialize_schema bytes: {hex_txt}");
         string table_name = br.read_string_cpp();
         int n_cols = br.ReadInt32(); // doing a br.Read was only advancing the data 1 byte!!
         List<ColumnInfo> cis = new List<ColumnInfo>();
@@ -137,8 +137,8 @@ namespace bts {

      public static ColumnInfo deserialize_column_info(ref BinaryReader br) {
         //col_no, name, type>
-         // var hex_txt = to_hex(br.BaseStream);
-         // bts.Log.Gen.info($"StreamSerializationUtils.deserialize_column_info bytes: {hex_txt}");
+         var hex_txt = to_hex(br.BaseStream);
+         bts.Log.Gen.info($"StreamSerializationUtils.deserialize_column_info bytes: {hex_txt}");
         int col_no = br.ReadInt32();
         string col_name = br.read_string_cpp();
         int tc_int = br.ReadInt32();
@@ -57,7 +57,7 @@ namespace bts {
         DSTagManager.ValidateStreamTag(tag);
         StreamIOMessage str_msg = _deser.deserialize(msg);
         var state_txt = _state == null ? "NULL" : _state.Count.ToString();
-         Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (pre.1) tag: {tag} sz: {msg.Length} _state: {state_txt}");
+         Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (pre.1) tag: {tag} sz: {msg.Length} str_msg: {str_msg} _state: {state_txt}");
         switch (str_msg.op) {
            case StreamIOMessage.StreamOp.CREATE:
               Log.isnull(_state);
@@ -70,7 +70,7 @@ namespace bts {
               _state.addRows(str_msg.multiple_remove_add_message.added_rows);
               break;
         }
-         Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (post) tag: {tag} sz: {msg.Length}  state: {_state.Count}");
+         Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (post) tag: {tag} sz: {msg.Length} str_msg: {str_msg} state: {_state.Count}");
      }

      public byte[] get_content(string tag) {
@@ -156,7 +156,7 @@ namespace BCG
      protected virtual void OnViewStreamDatabase(IWorkScheduler viewStreamDatabase) {
         var w = this as IWorker;
         if (w is not null) {
-            worker = viewStreamDatabase.Subscribe(w, (this as IParameterizedWorker)?.Interval);
+            worker = viewStreamDatabase.Subscribe(w);
         }
      }

@@ -7,52 +7,17 @@
    "Password": ""
  },
  "MasterWorker": {
-    "Interval": "00:00:00.1"
+    "Interval": "00:00:01"
  },
  "Publisher": {
    "Publishers": [
-      {
-        "Name": "market_data",
-        "Streams": [
-          {
-            "Name": "bbo",
-            "Expression": "bbo()",
-            "IsOnWebsocket": true
-          },
-          {
-            "Name": "instruments",
-            "Expression": "instruments()",
-            "IsOnWebsocket": true
-          },
-          {
-            "Name": "timesales",
-            "Expression": "timesales(10000)",
-            "IsOnWebsocket": true
-          },
-          {
-            "Name": "books",
-            "Expression": "books()",
-            "IsOnWebsocket": true
-          }
-        ]
-      },
      {
        "Name": "test_data",
        "Streams": [
          {
-            "name": "milliseconds_1000_100",
-            "expression": "milliseconds(1000,100)",
-            "is_on_websocket": true
-          },
-          {
-            "name": "test_literal",
-            "expression": "literal(\"x\", \"420.0\")",
-            "is_on_websocket": true
-          },
-          {
-            "name": "test_not_on_websocket",
-            "expression": "literal(\"x\", \"840.0\")",
-            "is_on_websocket": false
+            "Name": "seconds_1_10",
+            "Expression": "second(1,10)",
+            "IsOnWebsocket": true
          }
        ]
      }
@@ -118,7 +83,26 @@
    "UsePrepublishedServerState": false,
    "IpTranslation": null,
    "GcActivationThreshold": 0,
-    "WaitPeriodAfterGc": 1000
+    "WaitPeriodAfterGc": 1000,
+    "TagHandlerFactory": {
+      "Handlers": [
+        {
+          "TagType": "DataStream",
+          "Assembly": "DataStreamCore",
+          "Class": "bts.PubsubServerStreamTagHandler"
+        },
+        {
+          "TagType": "System_Sub",
+          "Assembly": "PubSub",
+          "Class": "BCG.SystemSubscriptionMapPublisher"
+        },
+        {
+          "TagType": "System_Stats",
+          "Assembly": "PubSub",
+          "Class": "BCG.SystemStatsPublisher"
+        }
+      ]
+    }
  },
  "ViewStreamDatabase": {
    "Interval": "00:00:01"
@@ -1,73 +1,91 @@
 {
-  "user": "bts",
-  "pubsub": {
-    "milliseconds_between_sub_info_updates": 1000,
-    "milliseconds_between_time_slice_updates": 5000,
-    "milliseconds_between_no_buffer_space_retries": 100,
-    "retries_on_no_buffer_space": 1,
-    "do_it_all_seconds_between_logs": 0,
-    "heartbeat_interval": 15000,
-    "missing_heartbeats_before_gc": 50,
-    "use_managed_sockets": true,
-    "state_update_period": 5000,
-    "statistics_update_period": 5000,
-    "disposed_connection_check_period": 1000,
-    "message_processing_warning_threshold": 1000,
-    "server_port": 12345,
-    "publisher_routes": [
-      {
-        "name": "default",
-        "server_address": "bluto://localhost:12345",
-        "tag": ".*"
-      }
-    ],
-    "subscriber_routes": [
-      {
-        "name": "default",
-        "server_address": "bluto://localhost:12345",
-        "tag": ".*"
-      }
-    ],
-    "max_packet_size": 500000000,
-    "blue_comm_must_die": true,
-    "crushable_tags": [],
-    "tag_type_to_assembly_map": [
-      {
-        "handler_class": "BCG.PubSubServerTagHandler",
-        "assembly": "DataStreamCore.dll"
-      }
-    ],
-    "server_queue": "PubSubQueue",
-    "port_range": {
-      "first": 30400,
-      "last": 31700
-    },
-    "bind_to_random_port": false,
-    "debug_tags": "^NOTAGS$",
-    "latency_report_threshold_ms": 1500,
-    "use_substations": false,
-    "use_dynamic_clients": false,
-    "is_substation": false,
-    "internally_publish_sub_info": true,
-    "use_managed_clients_in_native_code": false,
-    "compressed_tags": "^NOTAGS$",
-    "tags_to_report_compression_ratio": "^NOTAGS$",
-    "disable_msmq": false,
-    "use_prepublished_server_state": false,
-    "ip_translation": null,
-    "gc_activation_threshold": 0,
-    "wait_period_after_gc": 1000
+  "Spark": {
+    "ConfigServer": "http://localhost:8000/config"
  },
-  "publishers": [
-    {
-      "name": "test_data",
-      "streams": [
+  "SparkdCredentials": {
+    "User": "bts",
+    "Password": ""
+  },
+  "MasterWorker": {
+    "Interval": "00:00:00.1"
+  },
+  "Publisher": {
+    "publishers": [
+      {
+        "Name": "test_data",
+        "Streams": [
+          {
+            "Name": "test_literal",
+            "Expression": "literal(\"x\", \"420.0\")",
+            "IsOnWebsocket": true
+          }
+        ]
+      }
+    ]
+  },
+  "PubSub": {
+    "MillisecondsBetweenSubInfoUpdates": 1000,
+    "MillisecondsBetweenTimeSliceUpdates": 5000,
+    "MillisecondsBetweenNoBufferSpaceRetries": 100,
+    "RetriesOnNoBufferSpace": 1,
+    "DoItAllSecondsBetweenLogs": 0,
+    "HeartbeatInterval": 15000,
+    "MissingHeartbeatsBeforeGc": 50,
+    "UseManagedSockets": true,
+    "StateUpdatePeriod": 5000,
+    "StatisticsUpdatePeriod": 5000,
+    "DisposedConnectionCheckPeriod": 1000,
+    "MessageProcessingWarningThreshold": 1000,
+    "ServerPort": 12345,
+    "Routing": {
+      "PublisherRoutes": [
        {
-          "name": "test_literal",
-          "expression": "literal(\"x\", \"420.0\")",
-          "is_on_websocket": true
+          "Name": "default",
+          "ServerAddress": "bts-cloud://127.0.0.1:12345",
+          "TagRegExp": ".*"
+        }
+      ],
+      "SubscriberRoutes": [
+        {
+          "Name": "default",
+          "ServerAddress": "bts-cloud://127.0.0.1:12345",
+          "TagRegExp": ".*"
        }
      ]
-    }
-  ]
+    },
+    "MaxPacketSize": 500000000,
+    "BlueCommMustDie": true,
+    "CrushableTags": [],
+    "TagTypeToAssemblyMap": {
+      "Entries": [
+        {
+          "HandlerClass": "BCG.PubSubServerTagHandler",
+          "Assembly": "DataStreamCore.dll"
+        }
+      ]
+    },
+    "ServerQueue": "PubSubQueue",
+    "PortRange": {
+      "First": 30400,
+      "Last": 31700
+    },
+    "BindToRandomPort": false,
+    "DebugTags": "^NOTAGS$",
+    "LatencyReportThresholdMs": 1500,
+    "UseSubstations": false,
+    "UseDynamicClients": false,
+    "IsSubstation": false,
+    "InternallyPublishSubInfo": true,
+    "UseManagedClientsInNativeCode": false,
+    "CompressedTags": "^NOTAGS$",
+    "TagsToReportCompressionRatio": "^NOTAGS$",
+    "DisableMsmq": false,
+    "UsePrepublishedServerState": false,
+    "IpTranslation": null,
+    "GcActivationThreshold": 0,
+    "WaitPeriodAfterGc": 1000
+  },
+  "ViewStreamDatabase": {
+    "Interval": "00:00:01"
+  }
 }
@@ -1,25 +1,101 @@
 {
-  "user": "bts",
-  "publishers": [
-    {
-      "name": "test_data",
-      "streams": [
+  "Spark": {
+    "ConfigServer": "http://localhost:8000/config"
+  },
+  "SparkdCredentials": {
+    "User": "bts",
+    "Password": ""
+  },
+  "MasterWorker": {
+    "Interval": "00:00:00.1"
+  },
+  "Publisher": {
+    "publishers": [
+      {
+        "name": "test_data",
+        "streams": [
+          {
+            "name": "seconds_1_10",
+            "expression": "second(1,10)",
+            "is_on_websocket": true
+          },
+          {
+            "name": "test_literal",
+            "expression": "literal(\"x\", \"420.0\")",
+            "is_on_websocket": true
+          },
+          {
+            "name": "test_not_on_websocket",
+            "expression": "literal(\"x\", \"840.0\")",
+            "is_on_websocket": false
+          }
+        ]
+      }
+    ]
+  },
+  "PubSub": {
+    "MillisecondsBetweenSubInfoUpdates": 1000,
+    "MillisecondsBetweenTimeSliceUpdates": 5000,
+    "MillisecondsBetweenNoBufferSpaceRetries": 100,
+    "RetriesOnNoBufferSpace": 1,
+    "DoItAllSecondsBetweenLogs": 0,
+    "HeartbeatInterval": 15000,
+    "MissingHeartbeatsBeforeGc": 50,
+    "UseManagedSockets": true,
+    "StateUpdatePeriod": 5000,
+    "StatisticsUpdatePeriod": 5000,
+    "DisposedConnectionCheckPeriod": 1000,
+    "MessageProcessingWarningThreshold": 1000,
+    "ServerPort": 12345,
+    "Routing": {
+      "PublisherRoutes": [
        {
-          "name": "seconds_1_10",
-          "expression": "milliseconds(1000,10)",
-          "is_on_websocket": true
-        },
+          "Name": "default",
+          "ServerAddress": "bts-cloud://127.0.0.1:12345",
+          "TagRegExp": ".*"
+        }
+      ],
+      "SubscriberRoutes": [
        {
-          "name": "test_literal",
-          "expression": "literal(\"x\", \"420.0\")",
-          "is_on_websocket": true
-        },
-        {
-          "name": "test_not_on_websocket",
-          "expression": "literal(\"x\", \"840.0\")",
-          "is_on_websocket": false
+          "Name": "default",
+          "ServerAddress": "bts-cloud://127.0.0.1:12345",
+          "TagRegExp": ".*"
        }
      ]
-    }
-  ]
-}
+    },
+    "MaxPacketSize": 500000000,
+    "BlueCommMustDie": true,
+    "CrushableTags": [],
+    "TagTypeToAssemblyMap": {
+      "Entries": [
+        {
+          "HandlerClass": "BCG.PubSubServerTagHandler",
+          "Assembly": "DataStreamCore.dll"
+        }
+      ]
+    },
+    "ServerQueue": "PubSubQueue",
+    "PortRange": {
+      "First": 30400,
+      "Last": 31700
+    },
+    "BindToRandomPort": false,
+    "DebugTags": "^NOTAGS$",
+    "LatencyReportThresholdMs": 1500,
+    "UseSubstations": false,
+    "UseDynamicClients": false,
+    "IsSubstation": false,
+    "InternallyPublishSubInfo": true,
+    "UseManagedClientsInNativeCode": false,
+    "CompressedTags": "^NOTAGS$",
+    "TagsToReportCompressionRatio": "^NOTAGS$",
+    "DisableMsmq": false,
+    "UsePrepublishedServerState": false,
+    "IpTranslation": null,
+    "GcActivationThreshold": 0,
+    "WaitPeriodAfterGc": 1000
+  },
+  "ViewStreamDatabase": {
+    "Interval": "00:00:01"
+  }
+}
@@ -1,82 +0,0 @@
-namespace Bts.DataStreamPublisher;
-
-using Bts.PubSub;
-
-using CommandLine;
-
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Text.Json;
-using System.Text.Json.Serialization;
-
-public class DataStreamPublisherConfig {
-   /// <summary>
-   /// The login user for sparkd
-   /// </summary>
-   [JsonPropertyName("user")]
-   [Option("user", HelpText = "The sparkd login user")]
-   public string? User { get; set; }
-
-   /// <summary>
-   /// The login password for sparkd
-   /// </summary>
-   [JsonPropertyName("password")]
-   [Option("pass", HelpText = "The sparkd login password")]
-   public string? Password { get; set; }
-
-   /// <summary>
-   /// The config server url for sparkd
-   /// </summary>
-   [JsonPropertyName("sparkd")]
-   [Option("sparkd", HelpText = "The sparkd config server")]
-   public string? Sparkd { get; set; }
-
-   /// <summary>
-   /// The tables to publish on startup. Each key/value represents a table name/query to publish.
-   /// </summary>
-   [JsonPropertyName("publishers")]
-   public List<PublisherNode>? Publishers { get; set; }
-
-   [JsonPropertyName("pubsub")]
-   public PubSubConfig? PubSub { get; set; }
-
-   [Option(
-      "config",
-      MetaValue = "PATH",
-      HelpText = "Path to json file representing these options (DataStreamPublisherConfig)")]
-   [JsonIgnore]
-   public string? ConfigFile { get; set; }
-
-   public static DataStreamPublisherConfig Default() =>
-      new() {
-         PubSub = PubSubConfig.Default(),
-      };
-
-   public void MergeWith(DataStreamPublisherConfig other) {
-      User ??= other.User;
-      Password ??= other.Password;
-      Sparkd ??= other.Sparkd;
-      Publishers ??= other.Publishers;
-      PubSub ??= other.PubSub;
-      ConfigFile ??= other.ConfigFile;
-   }
-
-   /// <summary>
-   /// Merges options with those defined in `ConfigFile` (does not overwrite
-   /// non-null existing values).
-   /// </summary>
-   /// <exception cref="InvalidOperationException">
-   /// JsonSerializer.Deserialize returns null
-   /// </exception>
-   public void MergeWithConfigFile() {
-      if (ConfigFile is null) {
-         return;
-      }
-      var cfgString = File.ReadAllText(ConfigFile);
-      var parsedCfg = JsonSerializer.Deserialize<DataStreamPublisherConfig>(cfgString) ??
-         throw new InvalidOperationException($"Failed to parse json from {ConfigFile}");
-      MergeWith(parsedCfg);
-      ConfigFile = null;
-   }
-}
@@ -8,7 +8,9 @@ using bts.api.core;

 using Bts.DataStreamCore;
 using Bts.DataStreamExpressions;
+using Bts.PubSub;

+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Options;

@@ -29,17 +31,20 @@ public class PublisherService : IHostedService {
   readonly IPubTableBroker pubTableBroker;
   readonly PrimitiveManager primitiveManager;
   readonly IServiceProvider serviceProvider;
+   readonly IWorkScheduler workScheduler;

   public PublisherService(
      IPubTableBroker pubTableBroker,
      PrimitiveManager primitiveManager,
      IServiceProvider serviceProvider,
-      IOptions<PublisherServiceConfig> config
+      IOptions<PublisherServiceConfig> config,
+      [FromKeyedServices(WorkerGroups.Main)]IWorkScheduler workScheduler
   ) {
      this.pubTableBroker = pubTableBroker;
      this.primitiveManager = primitiveManager;
      this.serviceProvider = serviceProvider;
      this.config = config.Value;
+      this.workScheduler = workScheduler;
   }

   public Task StartAsync(CancellationToken cancellationToken) {
@@ -93,7 +98,9 @@ public class PublisherService : IHostedService {

      foreach (var publisher in publishers) {
         if (Directory.Exists(publisher.CsvDir)) {
-            csvPublishers.Add(new(publisher.CsvDir, pubTableBroker));
+            var csvPublisher = new CSVPublisher(publisher.CsvDir, pubTableBroker);
+            csvPublishers.Add(csvPublisher);
+            workScheduler.Add(csvPublisher);
         }
      }

@@ -122,7 +129,7 @@ public class PublisherService : IHostedService {
   void on_change(object obj) =>
      Log.Gen.info($"DSPublisher on_change dp: {obj}");

-   public class CSVPublisher : IWorker {
+   public class CSVPublisher : IParameterizedWorker {
      private IDictionary<string, ViewPublisher> _pubs = new Dictionary<string, ViewPublisher>();
      private readonly string _csv_directory;
      private IPubTableBroker _pub_table_broker;
@@ -130,8 +137,12 @@ public class PublisherService : IHostedService {
      public CSVPublisher(string csv_directory, IPubTableBroker pub_table_broker) {
         _csv_directory = csv_directory;
         _pub_table_broker = pub_table_broker;
-         MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
      }
+
+      string IParameterizedWorker.Group => WorkerGroups.Main;
+
+      TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(1);
+
      public void doWork() {
         var file_names = Directory.GetFiles(_csv_directory);
         foreach (var fn in file_names) {
@@ -1,5 +1,3 @@
-using bts;
-
 using mbts.utils;

 using System;
@@ -10,7 +8,7 @@ using System.Text.RegularExpressions;

 namespace BCG;

-public class PubSubRoute : IEquatable<PubSubRoute>, IComparable<PubSubRoute>, ICloneable<PubSubRoute> {
+public class PubSubRoute : IEquatable<PubSubRoute>, IComparable<PubSubRoute> {
   [JsonPropertyName("server_address")]
   public string? ServerAddress { get; set; }

@@ -44,13 +42,6 @@ public class PubSubRoute : IEquatable<PubSubRoute>, IComparable<PubSubRoute>, IC
      return myRegex.IsMatch(tag);
   }

-   public PubSubRoute Clone() =>
-      new() {
-         ServerAddress = ServerAddress,
-         TagRegExp = TagRegExp,
-         Name = Name,
-      };
-
   public int CompareTo(PubSubRoute? other) =>
      other is null ? -1 : comparer.Compare(this, other);

@@ -46,7 +46,7 @@ namespace BCG {
         if (!_pub_routes.TryGetFirst(r => r?.IsMatch(tag) ?? false, out var route)) {
            throw new ArgumentException($"Failed to find route for tag '{tag}'", nameof(tag));
         }
-         Log.Gen.info($"PubSubRouter.get_sub_url_for_tag tag: {tag} route: {route}");
+         Log.Gen.info($"PubSubRouter.get_sub_url_for_tag. tag: {tag} route: {route}");
         ArgumentNullException.ThrowIfNull(route.ServerAddress);
         return RemoteConnectionInfo.parse_url(route.ServerAddress);
      }
@@ -55,7 +55,7 @@ namespace BCG {
         if (!_sub_routes.TryGetFirst(r => r?.IsMatch(tag) ?? false, out var route)) {
            throw new ArgumentException($"Failed to find route for tag '{tag}'", nameof(tag));
         }
-         Log.Gen.info($"PubSubRouter.get_sub_url_for_tag tag: {tag} route: {route}");
+         Log.Gen.info($"PubSubRouter.get_sub_url_for_tag. tag: {tag} route: {route}");
         ArgumentNullException.ThrowIfNull(route.ServerAddress);
         return RemoteConnectionInfo.parse_url(route.ServerAddress);
      }
@@ -2,39 +2,10 @@ namespace Bts.PubSub.Routing;

 using BCG;

-using bts;
-
 using System.Collections.Generic;
-using System.Linq;

-public class PubSubRoutingConfig : ICloneable<PubSubRoutingConfig> {
+public class PubSubRoutingConfig {
   public List<PubSubRoute>? PublisherRoutes { get; set; }

   public List<PubSubRoute>? SubscriberRoutes { get; set; }
-
-   public void MergeWith(PubSubRoutingConfig other) {
-      PublisherRoutes = MergeRoutes(PublisherRoutes, other.PublisherRoutes);
-      SubscriberRoutes = MergeRoutes(SubscriberRoutes, other.SubscriberRoutes);
-
-      List<PubSubRoute>? MergeRoutes(List<PubSubRoute>? routes, List<PubSubRoute>? otherRoutes) {
-         if (otherRoutes is null) {
-            return routes;
-         }
-         if (routes is null) {
-            return otherRoutes;
-         }
-         foreach (var route in otherRoutes) {
-            if (!routes.Any(r => r.TagRegExp == route.TagRegExp || r.Name == route.Name)) {
-               routes.Add(route);
-            }
-         }
-         return routes;
-      }
-   }
-   
-   public PubSubRoutingConfig Clone() =>
-      new() {
-         PublisherRoutes = PublisherRoutes?.Select(r => r.Clone()).ToList(),
-         SubscriberRoutes = SubscriberRoutes?.Select(r => r.Clone()).ToList(),
-      };
 }
@@ -2,8 +2,14 @@ using System;
 using System.Collections.Generic;
 using System.Text;
 using System.Threading;
+using System.Threading.Tasks;
+
 // using BCG.Utilities;
- using BCG.Utilities.General;
+using BCG.Utilities.General;
+
+using Bts.PubSub;
+using Bts.PubSub.Services;
+
 using mbts;

 namespace BCG
@@ -16,15 +22,58 @@ namespace BCG
   /// </summary>
   public class ThreadDrivenMasterWorker : MasterWorker
   {
+      readonly object syncRoot = new();
+      readonly string? name;
+      readonly CancellationToken cancellationToken;
+      readonly TaskCompletionSource taskCompletionSource = new();
+      Thread? thread;
+
      internal ThreadDrivenMasterWorker()
-         : base()
-      { }
+         : this(new WorkSchedulerConfig() { Interval = TimeSpan.FromMilliseconds(100)}) { }
+
+      internal ThreadDrivenMasterWorker(string? name)
+         : this(new WorkSchedulerConfig() { Name = name, Interval = TimeSpan.FromMilliseconds(100) }) { }
+
+      internal ThreadDrivenMasterWorker(WorkSchedulerConfig config) : this(config, CancellationToken.None) { }
+
+      internal ThreadDrivenMasterWorker(WorkSchedulerConfig config, CancellationToken cancellationToken) : base(config.Interval) {
+         name = config.Name;
+         this.cancellationToken = cancellationToken;
+      }
+
+      public static ThreadDrivenMasterWorker StartNew(string? name) {
+         var rv = new ThreadDrivenMasterWorker(name);
+         rv.Start();
+         return rv;
+      }
+
+      public static ThreadDrivenMasterWorker StartNew(IWorker worker, TimeSpan interval) {
+         var rv = new ThreadDrivenMasterWorker();
+         rv.addWorker(worker, interval);
+         rv.Start();
+         return rv;
+      }
+
+      public Task StartAsync() {
+         Start();
+         return taskCompletionSource.Task;
+      }

      protected override void myInit()
      {
-         Thread doWorkThread = new Thread(new ThreadStart(myDoWork));
-         doWorkThread.IsBackground = true;
-         doWorkThread.Start();
+         if (thread is not null) {
+            // Already started
+            return;
+         }
+         lock (syncRoot) {
+            if (thread is null) {
+               thread = new(myDoWork);
+               if (name is not null) {
+                  thread.Name = name;
+               }
+               thread.Start();
+            }
+         }
      }

      public override void addWorker(IWorker worker, TimeSpan period)
@@ -50,7 +99,7 @@ namespace BCG
      private void myDoWork()
      {
         TimeSpan sleepTime = new TimeSpan(myPeriod.Ticks / 2);
-         while (true)
+         while (!cancellationToken.IsCancellationRequested)
         {
            try
            {
@@ -76,11 +125,10 @@ namespace BCG
            }
            catch (Exception ex)
            {
-               // BlueEventLog.logEvent(ex);
-               if (ex is OutOfMemoryException)
-                  throw ex;
+               taskCompletionSource.SetException(ex);
            }
         }
+         taskCompletionSource.SetResult();
      }

   }
@@ -12,46 +12,42 @@ using System.Collections.Generic;
 public class PubSubConfig {
   public short? ServerPort { get; set; }
   public bool? SimpleEchoServer { get; set; }
-
-   // unused
   public int? MillisecondsBetweenSubInfoUpdates { get; set; } = 1000;
-
-   // unused
   public int? MillisecondsBetweenTimeSliceUpdates { get; set; } = 5000;
-
-   // unused
   public int? MillisecondsBetweenNoBufferRetries { get; set; } = 100;
-
-   // unused
   public int? RetriesOnNoBufferSpace { get; set; } = 1;
-
-   // unused
   public int? DoItAllSecondsBetweenLog { get; set; } = 0;
-
-   // Used by the PubSubMessageReactor to pass to the ManagedTCPMessageHb (but unsed within it)
+   // Milliseconds
   public int? HeartbeatInterval { get; set; } = 15000;
   public int? MissingHeartbeatsBeforeGc { get; set; } = 50;
-
-   // unused
+   public bool? UseManagedSockets { get; set; } = true;
   // Milliseconds
   public int? StateUpdatePeriod { get; set; } = 5000;
-
-   // unused
   // Milliseconds
   public int? StatisticsUpdatePeriod { get; set; } = 5000;
   // Milliseconds
   public int? DisposedConnectionCheckPeriod { get; set; } = 1000;
-
+   // Milliseconds
+   public int? MessageProcessingWarningThreshold { get; set; } = 1000;
+   public int? MaxPacketSize { get; set; } = 500000000;
+   public bool? BlueCommMustDie { get; set; } = true;
+   public List<string>? CrushableTags { get; set; }
   public TagTypeToAssembly? TagTypeToAssemblyMap { get; set; }
-
-   // unused
+   public string? ServerQueue { get; set; } = "PubSubQueue";
+   public PubSubPortRange? PortRange { get; set; }
+   public bool? BindToRandomPort { get; set; } = false;
+   public string? DebugTags { get; set; }
   public int? LatencyReportThresholdMs { get; set; }
-
-   // unused
+   public bool? UseSubstations { get; set; }
+   public bool? UseDynamicClients { get; set; }
+   public bool? IsSubstation { get; set; }
   public bool? InternallyPublishSubInfo { get; set; }
-
+   public bool? UseManagedClientsInNativeCode { get; set; }
   public string? CompressedTags { get; set; }
-
+   public string? TagsToReportCompressionRatio { get; set; }
+   public bool? DisableMsmq { get; set; }
+   public bool? UsePrepublishedServerState { get; set; }
+   public string? IpTranslation { get; set; }
   public int? GcActivationThreshold { get; set; }
   // Milliseconds
   public int? WaitPeriodAfterGc { get; set; }
@@ -70,14 +66,30 @@ public class PubSubConfig {
         DoItAllSecondsBetweenLog = 0,
         HeartbeatInterval = 15000,
         MissingHeartbeatsBeforeGc = 50,
+         UseManagedSockets = true,
         StateUpdatePeriod = 5000,
         StatisticsUpdatePeriod = 5000,
         DisposedConnectionCheckPeriod = 1000,
+         MessageProcessingWarningThreshold = 1000,
         ServerPort = 12345,
+         MaxPacketSize = 500000000,
+         BlueCommMustDie = true,
+         CrushableTags = new(0),
         TagTypeToAssemblyMap = new(),
+         ServerQueue = "PubSubQueue",
+         PortRange = new(30400, 31700),
+         BindToRandomPort = false,
+         DebugTags = NOTAGS,
         LatencyReportThresholdMs = 1500,
+         UseSubstations = false,
+         IsSubstation = false,
         InternallyPublishSubInfo = true,
+         UseManagedClientsInNativeCode = false,
         CompressedTags = NOTAGS,
+         TagsToReportCompressionRatio = NOTAGS,
+         DisableMsmq = false,
+         UsePrepublishedServerState = false,
+         IpTranslation = null,
         GcActivationThreshold = 0,
         WaitPeriodAfterGc = 1000,
         SimpleEchoServer = false,
@@ -101,24 +113,33 @@ public class PubSubConfig {
      DoItAllSecondsBetweenLog ??= other.DoItAllSecondsBetweenLog;
      HeartbeatInterval ??= other.HeartbeatInterval;
      MissingHeartbeatsBeforeGc ??= other.MissingHeartbeatsBeforeGc;
+      UseManagedSockets ??= other.UseManagedSockets;
      StateUpdatePeriod ??= other.StateUpdatePeriod;
      StatisticsUpdatePeriod ??= other.StatisticsUpdatePeriod;
      DisposedConnectionCheckPeriod ??= other.DisposedConnectionCheckPeriod;
+      MessageProcessingWarningThreshold ??= other.MessageProcessingWarningThreshold;
      ServerPort ??= other.ServerPort;
+      MaxPacketSize ??= other.MaxPacketSize;
+      BlueCommMustDie ??= other.BlueCommMustDie;
+      CrushableTags ??= other.CrushableTags;
      TagTypeToAssemblyMap ??= other.TagTypeToAssemblyMap;
+      ServerQueue ??= other.ServerQueue;
+      PortRange ??= other.PortRange;
+      BindToRandomPort ??= other.BindToRandomPort;
+      DebugTags ??= other.DebugTags;
      LatencyReportThresholdMs ??= other.LatencyReportThresholdMs;
+      UseSubstations ??= other.UseSubstations;
+      IsSubstation ??= other.IsSubstation;
      InternallyPublishSubInfo ??= other.InternallyPublishSubInfo;
+      UseManagedClientsInNativeCode ??= other.UseManagedClientsInNativeCode;
      CompressedTags ??= other.CompressedTags;
+      TagsToReportCompressionRatio ??= other.TagsToReportCompressionRatio;
+      DisableMsmq ??= other.DisableMsmq;
+      UsePrepublishedServerState ??= other.UsePrepublishedServerState;
+      IpTranslation ??= other.IpTranslation;
      GcActivationThreshold ??= other.GcActivationThreshold;
      WaitPeriodAfterGc ??= other.WaitPeriodAfterGc;
      SimpleEchoServer ??= other.SimpleEchoServer;
-      if (other.Routing is not null) {
-         if (Routing is null) {
-            Routing = other.Routing?.Clone();
-         } else {
-            Routing.MergeWith(other.Routing);
-         }
-      }
   }
 }

@@ -1,78 +0,0 @@
-using System.Net.Sockets;
-using System.Net;
-
-using BCG;
-using System;
-using System.Collections.Generic;
-
-
-namespace bts {
-
-    // opens a TCP port and starts listening to connections
-    public class AsyncTCPConnectionAcceptor_APM {
-        private readonly int _port;
-        private Socket _accept_socket;
-        private ITCPConnectionHandler _connection_handler;
-        private HashSet<string> _key_map = new HashSet<string>();
-        public AsyncTCPConnectionAcceptor_APM(int port, ITCPConnectionHandler connection_handler) {
-            Log.Gen.info($"AsyncTCPConnectionAcceptor_APM. cons started on port: {port}");
-            _port = port;
-            _connection_handler = connection_handler;
-        }
-
-        public void init() {
-            _init();
-            _begin_accepting();
-        }
-
-        private void _init() {
-            Log.Gen.info($"AsyncTCPConnectionAcceptor_APM._init started on port: {_port}");
-            _accept_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-            TCPSocketUtils.set_server_socket_options(ref _accept_socket);
-            IPEndPoint endpoint = new IPEndPoint(IPAddress.Any, _port);
-            _accept_socket.Bind(endpoint);
-            _accept_socket.Listen(250);
-        }
-
-        private void _begin_accepting() {
-            var accept_key = Guid.NewGuid().ToString();
-            Log.Gen.info($"AsyncTCPConnectionAcceptor_APM._begin_accepting port: {_port} accept_key: {accept_key}");
-            var res = _accept_socket.BeginAccept(new AsyncCallback(_on_accept), accept_key);
-            if (res != null && res.IsCompleted && res.CompletedSynchronously)
-                _on_accept(res);
-        }
-
-        private void _on_accept(IAsyncResult res) {
-            var log_msg = $"AsyncTCPConnectionAcceptor_APM._on_accept state: {res.AsyncState}";
-            Log.Gen.info(log_msg);
-            Console.WriteLine(log_msg);
-            string key = res.AsyncState.ToString();
-            if (_key_map.Contains(key)) {
-                return;
-            }
-            _key_map.Add(key);
-            Socket new_sock;
-            try {
-                new_sock = _accept_socket.EndAccept(res);
-            } catch (Exception ex) {
-                bts.Log.Gen.info("AsyncTCPConnectionAcceptor_APM._on_accept exception thrown by EndAccept ex: {ex}");
-                System.Threading.Thread.Sleep(TimeSpan.FromMilliseconds(250));
-                _begin_accepting();
-                return;
-            }
-            new_sock.LingerState = new LingerOption(false, 0);
-            new_sock.Blocking = false;
-            ITCPConnection newConn = new AsyncTCPConnection_APM(new_sock, TimeSpan.FromSeconds(1), ITCPConnection.ConnectionSide.Server);
-            _connection_handler.onConnection(newConn);
-            // go back to accepting more conns
-            _begin_accepting();
-            var peer_rci = newConn.get_peer_address();
-            Log.Gen.info($"AsyncTCPConnectionAcceptor_APM._on_accept peer_rci: {peer_rci}");
-        }
-
-
-
-
-    }
-
-}
@@ -70,7 +70,8 @@ namespace bts {

      #region Sending 
      public void _send_fixed_size_msg(byte[] buffer) {
-         //Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg buffer: {to_hex(buffer, 0, buffer.Length)}");
+         string buff_txt = to_hex(buffer, 0, buffer.Length);
+         Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg buffer: {buff_txt}");

         int total_bytes_sent = 0;
         int n_bytes = buffer.Length;
@@ -113,7 +114,10 @@ namespace bts {

      private SyncQueue<byte[]> _send_queue = new SyncQueue<byte[]>();
      public void send(byte[] data) {
-         //Log.Gen.info($"AsyncTCPConnection_AsyncWrapper.send data: {to_hex(data, 0, data.Length}");
+         var data_txt = to_hex(data, 0, data.Length);
+         Log.Gen.info($"AsyncTCPConnection_AsyncWrapper.send data: {data_txt}");
+         //Task.Run(() => _send_one_msg(data));
+         //ThreadPool.QueueUserWorkItem(_ => _send_one_msg(data));
         _send_queue.enqueue(data);
      }
      #endregion
@@ -21,7 +21,7 @@ namespace bts {
         ThreadPool.QueueUserWorkItem(_ => _keep_trying_blocking_function(connect_once, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5)));
      }

-      // this is the actual fuction thta calls TCPCLient sync methof and establishes a connection
+      // this is the asctual fuction thta calls TCPCLient sync methof and establishes a connection
      private static void _connect_to(RemoteConnectionInfo server_rci, ITCPConnectionHandler handler) {
         // TODO, random sleep
         bts.Log.Gen.info($"AsyncTCPConnector_AsyncWrapper._connect_to server: {server_rci}");
@@ -41,8 +41,6 @@ namespace bts {
         return conn;
      }

-      // this is a ganeric function that calls a sync action function, and if unsuccessful timesout after "timeout"
-      // wais for wait_before_retrying before trying again
      public static void _keep_trying_blocking_function(Action action, TimeSpan timeout, TimeSpan wait_before_retrying) {
         Func<bool> try_once = () => {
            var cts = new CancellationTokenSource();
@@ -66,5 +64,26 @@ namespace bts {
            System.Threading.Thread.Sleep(TimeSpan.FromSeconds(1));
         }
      }
+
+      // public static void connect_to(RemoteConnectionInfo server_rci, TimeSpan socket_timeout, TimeSpan retry_interval, ITCPConnectionHandler handler) {
+      //    while (true) {
+      //       try {
+      //          var client = new TcpClient();
+      //          var result = client.BeginConnect(server_rci.ip, server_rci.port, null, null);
+      //          var success = result.AsyncWaitHandle.WaitOne(socket_timeout, true);
+      //          if (success && client.Connected) {
+      //             client.EndConnect(result);
+      //             handler.onConnection(new AsyncTCPConnection_AsyncWrapper(client));
+      //          } else {
+      //             throw new SocketException();
+      //          }
+      //       } catch (Exception ex) {
+      //          Log.Gen.info($"Connection attempt failed: {ex.Message}");
+      //          Log.Gen.info($"Retrying in {retry_interval.TotalSeconds} seconds...");
+      //          Thread.Sleep(retry_interval);
+      //       }
+      //    }
+      // }
+
   }
 }
@@ -15,8 +15,8 @@ namespace BCG {

      private DateTime myLastDoWorkTime = DateTime.MinValue,
                        myLastLogTime = DateTime.MinValue;
-      protected TimeSpan myPeriod = new TimeSpan(0, 0, 0, 0, 100),
-                           myLogPeriod = new TimeSpan(0, 1, 0);
+      protected readonly TimeSpan myPeriod;
+      protected readonly TimeSpan myLogPeriod = TimeSpan.FromMinutes(1);
      protected Dictionary<IWorker, Pair<TimeSpan, DateTime>> myWorkers = new Dictionary<IWorker, Pair<TimeSpan, DateTime>>(),
                                                                     myJustAdded = new Dictionary<IWorker, Pair<TimeSpan, DateTime>>(),
                                                                     myJustRemoved = new Dictionary<IWorker, Pair<TimeSpan, DateTime>>();
@@ -24,7 +24,18 @@ namespace BCG {
      private static MasterWorker myThreadDriven = null,
                                    myDoWorkDriven = null;

-      internal MasterWorker() {
+      internal MasterWorker() : this(TimeSpan.FromMilliseconds(100)) { }
+      internal MasterWorker(TimeSpan interval) {
+         myPeriod = interval;
+      }
+
+      public static MasterWorker StartNew() {
+         var rv = new MasterWorker();
+         rv.Start();
+         return rv;
+      }
+
+      public void Start() {
         myInit();
      }

@@ -34,11 +45,10 @@ namespace BCG {
         lock (typeof(MasterWorker)) {
            if (threadDriven) {
               // TODO
-               myThreadDriven = myThreadDriven ?? new ThreadDrivenMasterWorker();
+               myThreadDriven ??= ThreadDrivenMasterWorker.StartNew();
               return myThreadDriven;
-               return null;
            } else {
-               myDoWorkDriven = myDoWorkDriven ?? new MasterWorker();
+               myDoWorkDriven ??= StartNew();
               return myDoWorkDriven;
            }
         }
@@ -0,0 +1,6 @@
+namespace Bts.PubSub;
+
+public static class WorkerGroups {
+   public const string Main = "Main";
+   public const string Background = "Background";
+}
@@ -48,7 +48,8 @@ namespace BCG {
         this.publisher = publisher;
      }

-      public TimeSpan Interval => TimeSpan.FromMilliseconds(500);
+      string IParameterizedWorker.Group => WorkerGroups.Background;
+      TimeSpan IParameterizedWorker.Interval => TimeSpan.FromMilliseconds(500);

      #region IWorker Members

@@ -39,8 +39,8 @@ namespace BCG {
      public string TagType => tag_type;
      public string Tag => tag_name;

-      TimeSpan IParameterizedWorker.Interval =>
-         TimeSpan.FromMilliseconds(250);
+      string IParameterizedWorker.Group => WorkerGroups.Main;
+      TimeSpan IParameterizedWorker.Interval => TimeSpan.FromMilliseconds(250);

      internal static SystemStatsPublisher get_instance() {
         return Log.notnull(instance);                  
@@ -23,7 +23,6 @@ namespace BCG {
      private Thread _msg_polling_thread;
      // all the data,list of subscribers etc are in the state
      private IPubSubState _state;
-      private MasterWorker _mainloop_master_worker = MasterWorker.getInstance(false);
      // private static SyncQueue<PublishMsg> _internal_pub_queue = new SyncQueue<PublishMsg>();
      private static IInternalPublisherQueue? _internal_pub_queue;
      static int instanceCount = 0;
@@ -109,7 +108,6 @@ namespace BCG {
                  _handle_client_msg(kv.first, kv.second);
               }
               handle_internal_pubs();
-               _mainloop_master_worker.doWork();
            } catch (Exception ex) {
               bts.Log.Gen.error($"PubSubMessageReactor._polling_thread_func ex: {ex}");
               if (ex is OutOfMemoryException) {
@@ -0,0 +1,70 @@
+namespace Bts.PubSub.Services;
+
+using bts;
+
+using Microsoft.Extensions.Hosting;
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+/// <summary>
+/// An <see cref="IHostedService"/> that runs on a background thread.
+/// </summary>
+public abstract class BackgroundHostedService : IHostedService {
+   readonly TaskCompletionSource tcs = new();
+   readonly CancellationTokenSource cancellationTokenSource = new();
+   readonly object syncRoot = new();
+   Thread? thread;
+   State state;
+
+   protected BackgroundHostedService() { }
+
+   enum State {
+      Idle,
+      Starting,
+      Running,
+      Stopping,
+   }
+
+   string name => GetType().Name;
+
+   public Task StartAsync(CancellationToken cancellationToken) {
+      Log.isnull(thread);
+      lock (syncRoot) {
+         Log.assert(state == State.Idle);
+         state = State.Starting;
+      }
+      Log.Gen.info($"Starting {name}");
+      thread = new Thread(run) {
+         Name = name
+      };
+      thread.Start();
+      return Task.CompletedTask;
+   }
+
+   public Task StopAsync(CancellationToken cancellationToken) {
+      Log.Gen.info($"Stopping {name}...");
+      cancellationTokenSource.Cancel();
+      return tcs.Task;
+   }
+
+   protected abstract void Run(CancellationToken cancellationToken);
+
+   void run() {
+      Log.Gen.info($"{GetType()} STARTED.");
+      lock (syncRoot) {
+         state = State.Running;
+      }
+      try {
+         Run(cancellationTokenSource.Token);
+         Log.Gen.error($"{name} '{nameof(Run)}' finished");
+      } catch (Exception ex) {
+         Log.Gen.error($"{name} '{nameof(Run)}' threw an exception: {ex}");
+      }
+      lock (syncRoot) {
+         state = State.Idle;
+      }
+      tcs.SetResult();
+   }
+}
@@ -1,7 +1,8 @@
 namespace Bts.PubSub.Services;

 using System;
+using System.Collections.Generic;

 public class MasterWorkerConfig {
-   public TimeSpan Interval { get; set; } = TimeSpan.FromMilliseconds(100);
+   public List<WorkSchedulerConfig>? WorkSchedulers { get; set; }
 }
@@ -9,7 +9,9 @@ using bts.collections;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Options;

+using System;
 using System.Collections.Generic;
+using System.Collections.Immutable;
 using System.Linq;
 using System.Text.Json;
 using System.Text.Json.Nodes;
@@ -20,54 +22,57 @@ using System.Threading.Tasks;
 /// An <see cref="IHostedService"/> that imports a collection of <see cref="IWorker"/>s and
 /// calls each <see cref="IWorker.doWork"/> in a loop.
 /// </summary>
-public class MasterWorkerService : IHostedService {
+public class MasterWorkerService : BackgroundHostedService, IWorkScheduler {
   readonly TaskCompletionSource tcs = new();
   readonly CancellationTokenSource cancellationTokenSource = new();
   readonly List<IWorker> workers;
   readonly MasterWorkerConfig config;
-   MasterWorker? masterWorker;
-   Dictionary<string, IWorkScheduler>? backgroundMasterWorkers;
+   readonly object syncRoot = new();
+   readonly Dictionary<string, ThreadDrivenMasterWorker> backgroundWorkers = new();

   public MasterWorkerService(IEnumerable<IWorker> workers, IOptions<MasterWorkerConfig> config) {
      this.workers = workers.ToList();
      this.config = config.Value;
+      Log.notnull(this.config.WorkSchedulers);
+      Log.assert(this.config.WorkSchedulers.Count == 2);
+      Log.assert(this.config.WorkSchedulers[0].Name == WorkerGroups.Main);
+      Log.assert(this.config.WorkSchedulers[1].Name == WorkerGroups.Background);
+      foreach (var w in this.config.WorkSchedulers) {
+         Log.notnull(w.Name);
+         backgroundWorkers.Add(w.Name, new(w.Name));
+      }
   }

-   public Task StartAsync(CancellationToken cancellationToken) {
+   enum State {
+      Idle,
+      Working
+   }
+
+   public void Add(IWorker worker, TimeSpan interval) {
+      var group = getWorkerGroup(worker);
+      var scheduler = getOrAddScheduler(group);
+      scheduler.Add(worker, interval);
+   }
+
+   public void Remove(IWorker worker) {
+      var group = getWorkerGroup(worker);
+      var scheduler = getScheduler(group);
+      Log.notnull(scheduler).Remove(worker);
+   }
+
+   protected override void Run(CancellationToken cancellationToken) {
      var workersDesc = JsonSerializer.Serialize(workersToJson(workers));
      Log.Gen.info($"Starting MasterWorkerService. {workersDesc}");
-      foreach (var w in workers) {
-         GetWorkScheduler((w as IParameterizedWorker)?.Group).Add(w);
-      }
-      new Thread(workLoop).Start();
-      return Task.CompletedTask;
-
-      IWorkScheduler GetWorkScheduler(string? key) =>
-         key is null
-            ? (masterWorker ??= new())
-            : (backgroundMasterWorkers ??= new()).GetOrAdd(key, _ => new ThreadDrivenMasterWorker());
-   }
-
-   public Task StopAsync(CancellationToken cancellationToken) {
-      Log.Gen.info("Stopping MasterWorkerService...");
-      cancellationTokenSource.Cancel();
-      return tcs.Task;
-   }
-
-   void workLoop() {
-      Log.Gen.info("MasterWorkerService STARTED.");
-      if (masterWorker is null) {
-         Log.Gen.warn("MasterWorkerService STOPPED early because there is no work to do.");
-         return;
-      }
-      while (!cancellationTokenSource.IsCancellationRequested) {
-         masterWorker?.doWork();
-         Thread.Sleep(config.Interval);
+      foreach (var workScheduler in backgroundWorkers.Values) {
+         workScheduler.Start();
      }
      Log.Gen.info("MasterWorkerService STOPPED.");
      tcs.SetResult();
   }

+   static string getWorkerGroup(IWorker worker) =>
+      (worker as IParameterizedWorker)?.Group ?? WorkerGroups.Main;
+
   static JsonObject workersToJson(IEnumerable<IWorker> workers) {
      var rootJson = new JsonObject();
      foreach (var wg in workers.GroupBy(w => (w as IParameterizedWorker)?.Group ?? "Main")) {
@@ -77,4 +82,35 @@ public class MasterWorkerService : IHostedService {
      }
      return rootJson;
   }
+
+   IWorkScheduler? getScheduler(string group) {
+      lock (syncRoot) {
+         if (backgroundWorkers.TryGetValue(group, out var bgw)) {
+            return bgw;
+         }
+      }
+      return null;
+   }
+
+   IWorkScheduler getOrAddScheduler(IWorker worker) =>
+      getOrAddScheduler((worker as IParameterizedWorker)?.Group ?? WorkerGroups.Main);
+
+   IWorkScheduler getOrAddScheduler(string group) {
+      var (bgWorker, isNew) = getOrAddBackgroundWorker(group);
+      if (isNew) {
+         bgWorker.Start();
+      }
+      return bgWorker;
+   }
+
+   (ThreadDrivenMasterWorker worker, bool isNew) getOrAddBackgroundWorker(string group) {
+      lock (syncRoot) {
+         if (backgroundWorkers.TryGetValue(group, out var bgWorker)) {
+            return (bgWorker, false);
+         }
+         bgWorker = new(group);
+         backgroundWorkers.Add(group, bgWorker);
+         return (bgWorker, true);
+      }
+   }
 }
@@ -65,6 +65,7 @@ public static class PubSubServices {
      services
         .Configure<PubSubConfig>(configurationSection)
         .AddInternalPublisher()
+         .AddPubSubTagHandlerFactory(configurationSection.GetSection(nameof(PubSubConfig.TagHandlerFactory)))
         .AddSingleton<IPubSubTagHandlerFactory, PubsubTagHandlerFactory>()
         .AddSingleton<IPubSubRouter, PubSubRouter>()
         .AddPublisherCore()
@@ -0,0 +1,8 @@
+namespace Bts.PubSub.Services;
+
+using System;
+
+public class WorkSchedulerConfig {
+   public string? Name { get; set; } 
+   public TimeSpan Interval { get; set; } = TimeSpan.FromMilliseconds(100);
+}
@@ -0,0 +1,22 @@
+namespace Bts.PubSub.Services;
+
+using BCG;
+
+
+using System.Collections.Generic;
+
+public class WorkSchedulerProvider {
+   readonly Dictionary<string, IWorkScheduler> workSchedulers = new();
+   readonly object syncRoot = new();
+
+   public IWorkScheduler GetScheduler(string? group) {
+      group ??= string.Empty;
+      lock (syncRoot) {
+         return workSchedulers.GetOrAdd(
+            group,
+            g => string.IsNullOrEmpty(g)
+               ? new MasterWorker()
+               : new ThreadDrivenMasterWorker());
+      }
+   }
+}
@@ -0,0 +1,123 @@
+namespace Bts.PubSub;
+
+using BCG;
+using BCG.Utilities.General;
+
+using bts;
+using bts.collections;
+
+using Bts.PubSub.Services;
+
+using Microsoft.Extensions.Options;
+
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+using System.Threading;
+
+public class WorkSchedulerService : BackgroundHostedService, IWorkScheduler {
+   readonly List<IWorker> workers;
+   readonly MasterWorkerConfig config;
+   readonly IWorkScheduler mainWorker = new MasterWorker();
+   readonly object syncRoot = new();
+   ImmutableDictionary<string, IWorkScheduler> backgroundWorkers = ImmutableDictionary<string, IWorkScheduler>.Empty;
+
+   public WorkSchedulerService(IEnumerable<IWorker> workers, IOptions<MasterWorkerConfig> config) {
+      this.workers = workers.ToList();
+      this.config = config.Value;
+   }
+
+   public void Add(IWorker worker, TimeSpan interval) {
+      // getOrAddScheduler(worker).Add(worker, interval);
+      var group = (worker as IParameterizedWorker)?.Group;
+      if (string.IsNullOrEmpty(group)) {
+         lock (syncRoot) {
+            mainWorker.Add(worker, interval);
+         }
+         return;
+      }
+      if (backgroundWorkers.TryGetValue(group, out var bgw)) {
+         bgw.Add(worker, interval);
+      }
+      bgw = ThreadDrivenMasterWorker.StartNew(worker, interval);
+      lock (syncRoot) {
+         backgroundWorkers = backgroundWorkers.Add(group, bgw);
+      }
+   }
+
+   public void Remove(IWorker worker) {
+      var group = (worker as IParameterizedWorker)?.Group;
+      if (string.IsNullOrEmpty(group)) {
+         lock (syncRoot) {
+            mainWorker.Remove(worker);
+         }
+         return;
+      }
+      if (backgroundWorkers.TryGetValue(group, out var bgw)) {
+         bgw.Remove(worker);
+      }
+      var scheduler = getScheduler(worker);
+      Log.notnull(scheduler).Remove(worker);
+   }
+
+   protected override void OnStart(CancellationToken cancellationToken) {
+      var workersDesc = JsonSerializer.Serialize(workersToJson(workers));
+      Log.Gen.info($"Starting MasterWorkerService. {workersDesc}");
+      foreach (var w in workers) {
+         getOrAddScheduler(w).Add(w);
+      }
+   }
+
+   static JsonObject workersToJson(IEnumerable<IWorker> workers) {
+      var rootJson = new JsonObject();
+      foreach (var wg in workers.GroupBy(w => (w as IParameterizedWorker)?.Group ?? "Main")) {
+         var wgJson = new JsonArray();
+         wgJson.AddRange(wg.Select(w => JsonValue.Create(w.ToString())));
+         rootJson.Add(wg.Key, wgJson);
+      }
+      return rootJson;
+   }
+
+   IWorkScheduler? getScheduler(IWorker worker) =>
+      getScheduler((worker as IParameterizedWorker)?.Group);
+
+   IWorkScheduler? getScheduler(string? group) {
+      if (string.IsNullOrEmpty(group)) {
+         return mainWorker;
+      }
+      lock (syncRoot) {
+         if (backgroundWorkers.TryGetValue(group, out var bgw)) {
+            return bgw;
+         }
+      }
+      return null;
+   }
+
+   IWorkScheduler getOrAddScheduler(IWorker worker) =>
+      getOrAddScheduler((worker as IParameterizedWorker)?.Group);
+
+   IWorkScheduler getOrAddScheduler(string? group) {
+      if (string.IsNullOrEmpty(group)) {
+         return mainWorker;
+      }
+      lock (syncRoot) {
+         if (backgroundWorkers.TryGetValue(group, out var bgw)) {
+            return bgw;
+         }
+         bgw = new();
+         backgroundWorkers = backgroundWorkers.Add(group, bgw);
+         bgw.Start();
+         return bgw;
+      }
+   }
+
+   protected override void Run(CancellationToken cancellationToken) {
+      while (!cancellationToken.IsCancellationRequested) {
+         mainWorker.doWork();
+         Thread.Sleep(config.Interval);
+      }
+   }
+}
@@ -52,7 +52,7 @@ namespace BCG {
            if (_tags.Contains(msg.tag)) {
               rv.Add(msg);
            } else {
-               Log.Gen.warn($"SingleServerSubscriberCore.get_updates msg: {msg} does not have a subscription");
+               Log.Gen.warn("SingleServerSubscriberCore.get_updates msg: {msg} does not have a subscription");
            }
         }
         return rv;
@@ -0,0 +1,115 @@
+#nullable enable
+namespace Bts.PubSub;
+
+using BCG.Utilities.General;
+
+using bts;
+
+using System;
+
+public class WorkActivator : IDisposable {
+   readonly object syncRoot = new();
+   readonly IWorker worker;
+   IWorkScheduler? workScheduler;
+
+   State state = State.Idle;
+
+   public WorkActivator(IWorker worker) {
+      this.worker = worker;
+   }
+
+   enum State {
+      Idle,
+      WaitingToWork,
+      Working
+   }
+
+   public void OnWorkScheduler(IWorkScheduler workScheduler) {
+      lock (syncRoot) {
+         if (workScheduler == this.workScheduler) {
+            return;
+         }
+         switch (state) {
+            case State.Idle:
+               this.workScheduler = workScheduler;
+               break;
+            case State.WaitingToWork:
+               /// We only transition <see cref="State.WaitingToWork"/> when <see cref="Start"/> is
+               /// called and <see cref="worker"/> is null.
+               Log.isnull(this.workScheduler);
+               this.workScheduler = workScheduler;
+               if (workScheduler is not null) {
+                  state = State.Working;
+                  workScheduler.Add(worker);
+               }
+               break;
+            case State.Working:
+               Log.notnull(this.workScheduler).Remove(worker);
+               this.workScheduler = workScheduler;
+               if (workScheduler is not null) {
+                  workScheduler.Add(worker);
+               } else {
+                  state = State.WaitingToWork;
+               }
+               break;
+         }
+      }
+   }
+
+   public void Start() {
+      lock (syncRoot) {
+         switch (state) {
+            case State.Idle:
+               if (workScheduler is null) {
+                  state = State.WaitingToWork;
+               } else {
+                  state = State.Working;
+                  workScheduler.Add(worker);
+               }
+               break;
+            case State.WaitingToWork:
+               // Sanity check - "WaitingToWork" means we are waiting for a non-null IWorkScheduler.
+               // It follows that if we're "WaitingToWork", then our IWorkScheduler should be null.
+               Log.isnull(workScheduler);
+               break;
+            case State.Working:
+               Log.notnull(workScheduler);
+               break;
+         }
+      }
+   }
+
+   public void Stop() {
+      lock (syncRoot) {
+         switch (state) {
+            case State.Idle:
+               break;
+            case State.WaitingToWork:
+               state = State.Idle;
+               break;
+            case State.Working:
+               Log.notnull(workScheduler).Remove(worker);
+               state = State.Idle;
+               break;
+         }
+      }
+   }
+
+   public void Dispose() {
+      lock (syncRoot) {
+         stopImpl();
+         workScheduler = null;
+      }
+   }
+
+   void stopImpl() {
+      if (workScheduler is not null) {
+         switch (state) {
+            case State.Working:
+               workScheduler.Remove(worker);
+               break;
+         }
+      }
+      state = State.Idle;
+   }
+}
@@ -1,117 +0,0 @@
-Basic TCP:
-ITCPConnection implemented by AsyncTCPConnection_APM is the innermost class resposinsble for TCP sends and received. Its used 
-on both client and server symmetrically. All the hard async send/receive logic is there.
-AsyncTCPConnectionAcceptor_APM opens a port and listens and accepts to connections from the client. It creates the ITCPConnection
-on the server.
-AsyncTCPConnector_APM connects to the server (above) and creates an ITCPConnection on the client side.
-
-IPubSubServerConnection:
-All communication to PubSubServer happens using this. It has 3 main types of calls:
-1. send: to asyncronoulsy end data
-2. get data: either through a polling call or callback
-3. get connection info: through callbacks
-
-The only implementation of the above is PubSubServerManagedTCPConnection
-(there used to be an MSMQ one and another wrapped around native code - but neither exists anymore)
-
-PubSubMessages:
-PubSubClientMessage is the only messages thats sent from client to pubsubserver. Its serialized on the
-client and sent over to the pubsubserver, where uints received ad deserialized
-
-SubscriberCore:ISubscriberCore (retired now)
-|
-V
-SingleServerSubscriberCore:ISingleServerSubscriberCore
-
-PublisherCore:IPublisherCore (transactions not supported for now), keeps AllPublishedTagState to cache state on the publisher
-|
-V
-IPubSubServerConnection
-
-PubSubServer:
-PubSubMessageReactor (top level, called from main), contain PubSubState (which has all data and connection state)
-ManagedTCPMessageHub : IPubsubMessageHub (this is what handles each ITCPConnection for each connection)
-
-PubSubState keeps the total pubsub data in 2 classes
-Dictionary<string, SingleTagData> and 
-Dictionary<string, SingleTagConnectionData>
-SingleTagData has the tag and IPubSubServerTagHandler (or byte[]) (IPubSubServerTagHandler holds the data state)
-for streams the IPubSubServerTagHandler is the PubsubServerTagHandler, which is designed for multiple tags and has a IDictionary<string, TagStateBank>
-TagStateBank containes the PubsubDeserial izationController, which knows how to deserialize the stream publishes and build state, ad simialirly produce the state on demand
-PubsubTagHandlerFactory returns a IPubSubServerTagHandler for each tag, called by the server state handler, SingleTagData and publisher tag state, AllPublishedTagState
-
-SingleTagConnectionData contains connection vs sub, and connection vs pub
-PubSubState calls
-SystemSubscriptionMapPublisher.init(this);
-SystemStatsPublisher.init(this);
-SystemHeartbeatPublisher.init();
---------------------------------------------------------------
-How internal tags are published?
-There are 3 internal publishers (ie compute somerthing an dpublish tit PubSubServer, such that clients can subscribe to it):
-SystemStatsPublisher, SystemSubscriptionMapPublisher and SystemHeartbeatPublisher
-These are somewhat tricky because they are both listeners to PubUsbState event and intrnally publish, but are also IPubSubServerTagHandler
-SystemSubscriptionMapPublisher:
-just one instance, a static init is called by PubSubState and it passes itself. SystemSubscriptionMapPublisher subscribes to the event
-gets the affected tags, marks them dirty, in doWork creates the state and calls the internal_publish.
-Now, as the result of an internal_publish (which is pretty much like a rmote publish as far as the PubsubServer is concerned), SingleTagData
-ends up calling SystemSubscriptionMapPublisher (since it also serves as the tagHandler for the tag it publishes via internal_publish)
-
-
-
-PubSubState has onPublishStats (tag, sz) that is fired on update_pub_data
-onPublishStats is subscribed by SystemStatsPublisher
-SystemStatsPublisher.onPublishStats updates the state and on the next doWork calls PubSubMessageReactor.internal_send
-SystemStatsPublisher is also an IPubSubServerTagHandler so SingleTagData calls it getContent to get full state
-
-Streams:
-PubTableBroker allows publishing of streams through createTable, addRow, deleteRow, changeRow, makeBatchChanges ->
-PubTablePublisher:IPubTablePublisher per tag (addRow, deleteRow, changeRow, createTable, batchRemoveAdd)
-PubTablePublisher has BlueDataSerializationController:IBlueDataSerializer to serialize the rows to byte[]
-BlueDataSerializationController uses a IBlueDataSerializer, which is specialized for the tag (using UniversalSerializer for now)
-
-IBlueDataSerializer has just one function
-void serialize (byte op, IBlueRowCollection[] rows, out byte[] serializedData);
-the second arg is IBlueRowCollection[] because in case of op = StreamOpCode.multipleRemoveAdd, it has multiple IBlueRowCollection for rem and add
-(BlueDataSerializer.serialize's comment has the serialization layout)
-
-For each serializer we have 2 matching deserializers, IPubsubDeserializer (for PubSubservr) and IClientDeserializer (for clients subscribing to streams)
-For UniversalSerializer, we have PubsubUniversalDeserializer and ClientUniversalDeserializer
-
-------------------------------------------------------------------
-TCP publishing:
-To run the base TCP layer run PubSubServer as an echo server, 
-(set run_echo_tcp_server_only in spark\src\DataStreamSuite\PubSubServer\Program.cs)
-to run a client that sends a tcp message tot he echo server, run 
-SimpleTCPEchoStream
-
-
-Pubsub publishing:
-----------------
-PubSubTestingStream (bounces pubs across the pubsub server)
-mutableAppend(pubSubTesting(100,10000000,100000,"test_1"),
-pubSubTesting(250,10000000,100000,"test_2"))
-
-
-Stream publishing:
------------------
-PublisherStream (publishes any stream to the pubsubserver)
-r=second(100)
-r=project(r,"id")
-r=math(r,"x = id*id")
-publisher(r, "test_x")
-
-
-tracking message into the pubsubserver
--------------------------------------
-connection:
-ServerToClientManagedTCPConnectionAcceptor._on_accept
- var log_msg = $"ServerToClientManagedTCPConnectionAcceptor._on_accept called port: {_port} state: {res.AsyncState}";
-         Log.Gen.info(log_msg);
-
-data:
-ManagedTCPMessageHub.onData
-bts.Log.Gen.info($"ManagedTCPMessgageHub.onData recv msg: {message}");
-PubSubMessageReactor._polling_thread_func
-bts.Log.Gen.info($"PubSubMessageReactor._polling_thread_func n_msgs: {n_msgs}");
-PubSubMessageReactor._handle_client_msg
-bts.Log.Gen.info($"PubSubMessageReator._handle_client_msg ps_client_msg: {ps_client_msg}");
@@ -1,58 +1,117 @@
-If the DSQA comes up empty, we need to take it one step at a time and ensure it in the following order:
+Basic TCP:
+ITCPConnection implemented by AsyncTCPConnection_APM is the innermost class resposinsble for TCP sends and received. Its used 
+on both client and server symmetrically. All the hard async send/receive logic is there.
+AsyncTCPConnectionAcceptor_APM opens a port and listens and accepts to connections from the client. It creates the ITCPConnection
+on the server.
+AsyncTCPConnector_APM connects to the server (above) and creates an ITCPConnection on the client side.

--------------------
-PubsubServer routes:
--------------------
-the client (both publisher and subscriber) need to connect tot he rigth pubsubserver. There are 3 days to do this:
-1. Bring up a DSQA and check the title bar, it should be tag:regExp: PubSunRoute
-the ip and port matter, the prefix is just fo the human eye.
-2. Run the stream pss_routes().
-It will give a row per tag reg exp vs the pss route (pss == pubsub server0)
-3. Check the clirnt logs for:
-PubSubRouter.get_sub_url_for_tag tag: and PubSubRouter.get_pub_url_for_tag tag:
+IPubSubServerConnection:
+All communication to PubSubServer happens using this. It has 3 main types of calls:
+1. send: to asyncronoulsy end data
+2. get data: either through a polling call or callback
+3. get connection info: through callbacks

-------------------
-Connection Attempt:
-------------------
-Check the pss logs for:
-"AsyncTCPConnectionAcceptor_AsyncWrapper.cons started on port:"
-Make sure it is listening to the correct port
+The only implementation of the above is PubSubServerManagedTCPConnection
+(there used to be an MSMQ one and another wrapped around native code - but neither exists anymore)

-On the client, make sure it is attemping to connect to the right pss
-"AsyncTCPConnector_AsyncWrapper.connect_to server:"
+PubSubMessages:
+PubSubClientMessage is the only messages thats sent from client to pubsubserver. Its serialized on the
+client and sent over to the pubsubserver, where uints received ad deserialized
+
+SubscriberCore:ISubscriberCore (retired now)
+|
+V
+SingleServerSubscriberCore:ISingleServerSubscriberCore
+
+PublisherCore:IPublisherCore (transactions not supported for now), keeps AllPublishedTagState to cache state on the publisher
+|
+V
+IPubSubServerConnection
+
+PubSubServer:
+PubSubMessageReactor (top level, called from main), contain PubSubState (which has all data and connection state)
+ManagedTCPMessageHub : IPubsubMessageHub (this is what handles each ITCPConnection for each connection)
+
+PubSubState keeps the total pubsub data in 2 classes
+Dictionary<string, SingleTagData> and 
+Dictionary<string, SingleTagConnectionData>
+SingleTagData has the tag and IPubSubServerTagHandler (or byte[]) (IPubSubServerTagHandler holds the data state)
+for streams the IPubSubServerTagHandler is the PubsubServerTagHandler, which is designed for multiple tags and has a IDictionary<string, TagStateBank>
+TagStateBank containes the PubsubDeserial izationController, which knows how to deserialize the stream publishes and build state, ad simialirly produce the state on demand
+PubsubTagHandlerFactory returns a IPubSubServerTagHandler for each tag, called by the server state handler, SingleTagData and publisher tag state, AllPublishedTagState
+
+SingleTagConnectionData contains connection vs sub, and connection vs pub
+PubSubState calls
+SystemSubscriptionMapPublisher.init(this);
+SystemStatsPublisher.init(this);
+SystemHeartbeatPublisher.init();
+---------------------------------------------------------------
+How internal tags are published?
+There are 3 internal publishers (ie compute somerthing an dpublish tit PubSubServer, such that clients can subscribe to it):
+SystemStatsPublisher, SystemSubscriptionMapPublisher and SystemHeartbeatPublisher
+These are somewhat tricky because they are both listeners to PubUsbState event and intrnally publish, but are also IPubSubServerTagHandler
+SystemSubscriptionMapPublisher:
+just one instance, a static init is called by PubSubState and it passes itself. SystemSubscriptionMapPublisher subscribes to the event
+gets the affected tags, marks them dirty, in doWork creates the state and calls the internal_publish.
+Now, as the result of an internal_publish (which is pretty much like a rmote publish as far as the PubsubServer is concerned), SingleTagData
+ends up calling SystemSubscriptionMapPublisher (since it also serves as the tagHandler for the tag it publishes via internal_publish)


------------------------------------
-Successful connection establishment:
------------------------------------

-On pss, a successful connection should result in the log:
-"calling onConnection now"
-The whole log is 
-Log.Gen.info($"AsyncTCPConnectionAcceptor_AsyncWrapper._begin_accepting port: {_accept_port}, new connection, self: {conn.get_self_address()} peer: {conn.get_peer_address()} calling onConnection now");
+PubSubState has onPublishStats (tag, sz) that is fired on update_pub_data
+onPublishStats is subscribed by SystemStatsPublisher
+SystemStatsPublisher.onPublishStats updates the state and on the next doWork calls PubSubMessageReactor.internal_send
+SystemStatsPublisher is also an IPubSubServerTagHandler so SingleTagData calls it getContent to get full state

-On the client, a successful connection should result in the log:
-"AsyncTCPConnector_AsyncWrapper._connect_to new connection,"
+Streams:
+PubTableBroker allows publishing of streams through createTable, addRow, deleteRow, changeRow, makeBatchChanges ->
+PubTablePublisher:IPubTablePublisher per tag (addRow, deleteRow, changeRow, createTable, batchRemoveAdd)
+PubTablePublisher has BlueDataSerializationController:IBlueDataSerializer to serialize the rows to byte[]
+BlueDataSerializationController uses a IBlueDataSerializer, which is specialized for the tag (using UniversalSerializer for now)
+
+IBlueDataSerializer has just one function
+void serialize (byte op, IBlueRowCollection[] rows, out byte[] serializedData);
+the second arg is IBlueRowCollection[] because in case of op = StreamOpCode.multipleRemoveAdd, it has multiple IBlueRowCollection for rem and add
+(BlueDataSerializer.serialize's comment has the serialization layout)
+
+For each serializer we have 2 matching deserializers, IPubsubDeserializer (for PubSubservr) and IClientDeserializer (for clients subscribing to streams)
+For UniversalSerializer, we have PubsubUniversalDeserializer and ClientUniversalDeserializer
+
+-------------------------------------------------------------------
+TCP publishing:
+To run the base TCP layer run PubSubServer as an echo server, 
+(set run_echo_tcp_server_only in spark\src\DataStreamSuite\PubSubServer\Program.cs)
+to run a client that sends a tcp message tot he echo server, run 
+SimpleTCPEchoStream


------------------------------------
-Successful message sent from client:
------------------------------------
-"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg sent"
+Pubsub publishing:
+-----------------
+PubSubTestingStream (bounces pubs across the pubsub server)
+mutableAppend(pubSubTesting(100,10000000,100000,"test_1"),
+pubSubTesting(250,10000000,100000,"test_2"))

+
+Stream publishing:
+------------------
+PublisherStream (publishes any stream to the pubsubserver)
+r=second(100)
+r=project(r,"id")
+r=math(r,"x = id*id")
+publisher(r, "test_x")
+
+
+tracking message into the pubsubserver
 --------------------------------------
-Successful message received on client:
--------------------------------------
-"AsyncTCPConnection_AsyncWrapper._read_fixed_size_msg read"
+connection:
+ServerToClientManagedTCPConnectionAcceptor._on_accept
+ var log_msg = $"ServerToClientManagedTCPConnectionAcceptor._on_accept called port: {_port} state: {res.AsyncState}";
+         Log.Gen.info(log_msg);

-
-pss's send/receive is the same log, since it uses the same class AsyncTCPConnection_AsyncWrapper
---------------------------------
-Successful message sent from pss:
---------------------------------
-"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg sent"
-
-----------------------------------
-Successful message received on pss:
-----------------------------------
-"AsyncTCPConnection_AsyncWrapper._read_fixed_size_msg read"
+data:
+ManagedTCPMessageHub.onData
+bts.Log.Gen.info($"ManagedTCPMessgageHub.onData recv msg: {message}");
+PubSubMessageReactor._polling_thread_func
+bts.Log.Gen.info($"PubSubMessageReactor._polling_thread_func n_msgs: {n_msgs}");
+PubSubMessageReactor._handle_client_msg
+bts.Log.Gen.info($"PubSubMessageReator._handle_client_msg ps_client_msg: {ps_client_msg}");
@@ -1,6 +1,6 @@
 {
  "MasterWorker": {
-    "Interval": "00:00:00.1"
+    "Interval": "00:00:01"
  },
  "SparkLogging": {
    "ConfigServer": "http://localhost:8000/config"
@@ -61,7 +61,7 @@ namespace BCG {
      }

      private void _on_books(TableUpdate<BookTuple, uint> update) {
-         //_log_updates(update);
+         _log_updates(update);
         bts.Log.Gen.info($"BookStream._on_books state: {update.state.Count()} changes: {update.changes.Count()} added: {update.added.Count()}");
         _cached_state = update.state;
      }
@@ -1,114 +0,0 @@
-namespace bts
-{
-   using BCG;
-   using BCG.Utilities.General;
-
-   public class TableMetaDataStream : ViewStream, IWorker
-   {
-      private class TableState
-      {
-         public ViewStream table_stream;
-         public IBlueTableSchema schema = null;
-         public bool processed = false;
-         public TableState(ViewStream table_stream_)
-         {
-            table_stream = table_stream_;
-            table_stream.onChange += _handle_table_update;
-         }
-         private void _handle_table_update(object dp_obj)
-         {
-            DataPacket dp = dp_obj as DataPacket;
-            schema = dp.schema;
-         }
-         public void unsub()
-         {
-            table_stream.onChange -= _handle_table_update;
-         }
-      }
-      private IBlueTableSchema _schema = BlueDataUtility.form_schema("TableMetaData",
-         Tuple.Create("id", TypeCode.Int32),
-         Tuple.Create("table_name", TypeCode.String),
-         Tuple.Create("col_idx", TypeCode.Int32),
-         Tuple.Create("col_type_code", TypeCode.Int32),
-         Tuple.Create("col_name", TypeCode.String));
-
-      private ViewStreamDatabase _vsd;
-      private ViewStream _base_stream;
-      private IDictionary<string, TableState> _table_map = new Dictionary<string, TableState>();
-      private BlueTable _state;
-      public TableMetaDataStream()
-      : this(ViewStreamDatabase.instance) { }
-      public TableMetaDataStream(ViewStreamDatabase vsd)
-      {
-         _vsd = vsd;
-         _base_stream = new PubSubServerStateStream(_vsd);
-         _state = new BlueTable(_schema);
-      }
-      protected override void myOnFirstSubscriber()
-      {
-         _base_stream.onChange += _handle_base_stream;
-         _vsd.addWorker(this);
-      }
-
-      protected override void myOnLastSubscriber()
-      {
-         _base_stream.onChange -= _handle_base_stream;
-         _vsd.removeWorker(this);
-      }
-
-      public void doWork()
-      {
-         bool changed = false;
-         foreach (TableState ts in _table_map.Values)
-         {
-            if (!ts.processed && ts.schema != null)
-            {
-               foreach (ColumnInfo ci in ts.schema.columnInfo)
-               {
-                  int id = _state.Count;
-                  var table_name = ts.schema.tableName;
-                  var col_idx = ci.colNo;
-                  var col_name = ci.colName;
-                  TypeCode type_code = BlueDataUtility.get_typecode(ci.colType);
-                  BlueRow nr = new BlueRow(_schema, id, table_name, col_idx, Convert.ToInt32(type_code),
-                                          col_name);
-                  changed = true;
-                  _state.addRow(nr);
-               }
-               ts.processed = true;
-               ts.unsub();
-            }
-         }
-         if (changed)
-         {
-            var dp_out = BlueDataUtility.build_dp_inc(_state, myLastValue as DataPacket);
-            myFireEvent(dp_out);
-         }
-      }
-
-      private void _handle_base_stream(object dp_obj)
-      {
-         DataPacket dp = dp_obj as DataPacket;
-         foreach (IBlueRow r in dp.newRows)
-         {
-            string tag = r.get<string>("tag");
-            if (tag.StartsWith("DS_"))
-            {
-               _handle_table(tag);
-            }
-         }
-      }
-
-      private void _handle_table(string tag_name)
-      {
-         Log.assert(tag_name.StartsWith("DS_"));
-         string table_name = tag_name.Remove(0, 3);
-         TableState ts = null;
-         if (!_table_map.TryGetValue(table_name, out ts))
-         {
-            ts = new TableState(new TableStream(table_name, _vsd));
-            _table_map.Add(table_name, ts);
-         }
-      }
-   }
-}
@@ -2,8 +2,7 @@
 namespace bts {
   using System;

-   public interface ICloneable<T> : ICloneable where T : notnull {
+   public interface ICloneable<T> : ICloneable {
      new T Clone();
-      object ICloneable.Clone() => Clone();
   }
 }
Author	SHA1	Message	Date
John Colvin	554ceea2ee	check pt	2024-07-07 22:19:25 +00:00
John Colvin	d54b0437b0	Add config for PubSub tag handler factory	2024-07-05 13:50:39 +00:00
John Colvin	81f51352b4	devcontainer: Install dotnet8 expose ports for DataStream Web App	2024-07-04 17:33:37 +00:00
John Colvin	6f521f8f6a	fix build: Copy libspark to net8.0 dir For the DataStream Web App	2024-07-04 17:33:37 +00:00
John Colvin	7a0f36f686	fix build: Copy dll to all dirs in DOTNET_BINARY_DIR Needed to wrap the var in double quotes	2024-07-04 17:33:37 +00:00
John Colvin	a14aa58a24	build: Add scripts to publish data stream apps	2024-07-04 17:33:37 +00:00
John Colvin	667ec4ca10	Fix typo in log msg	2024-07-04 17:33:37 +00:00
John Colvin	7698fa15f6	Fix work scheduler issues	2024-07-04 13:32:22 -04:00
John Colvin	e78b405139	refactor: Remove JsonPropertyName attributes from PubSubConfig The ApplicationBuilder doesn't support these attributes when parsing config.	2024-07-04 13:32:22 -04:00
John Colvin	bf96899fa3	refactor: Add config for IPubSubTagHandlerFactory	2024-07-04 13:32:22 -04:00
John Colvin	e919aa9d05	fix: Make sure transient streams are attached to IServiceProvider So they can retrieve services necessary to function (e.g., add themselves to the master IWorkScheduler so doWork gets called, etc.)	2024-07-04 13:32:22 -04:00
John Colvin	a05c7410dc	refactor: Use DependencyInjection paradigm in DataStreamSuite This allows us to create loosely coupled services with automatic dependency resolution	2024-07-02 11:30:14 -04:00
John Colvin	6cc0ad4f66	Fix net7 -> net8 typo in publish_data_stream_web_app.py The DataStream Web App is on net8	2024-07-01 23:36:04 -04:00
John Colvin	8fff414c4c	Fix build scripts so they propagate the exit code back to the shell I noticed this was broken because launch.json configurations would still start the debugger even when the preLaunchTask (build) failed!	2024-07-01 23:36:04 -04:00
John Colvin	8421e4ae1a	Change PubSub scheme prefix from 'bluto' -> 'bts-cloud'	2024-07-01 23:36:04 -04:00
Taha	a8d0712e30	Added pubsub routes to the Title of DSQA	2024-07-01 23:36:04 -04:00
John Colvin	aa36a7393c	vscode: Add DataStreamPublisher debug cfg to skip build	2024-07-01 11:23:46 -04:00
John Colvin	7e0d5e4f59	build: Use separate cmd to build DataStreamSuite/QueryAnalyzer.sln The DataStreamSuite.sln builds the new webapp now, which will (eventually) replace DSQA. We'll keep it around for now, but with a clean separation from the main build.	2024-07-01 11:23:46 -04:00
John Colvin	d26e653a6b	Tweak grid colors in DataStream web app	2024-07-01 11:23:46 -04:00
John Colvin	56e087ab8b	refactor: Move DataStreamWebApp themes to their own js file	2024-07-01 11:23:45 -04:00
Taha	c71bb8c301	Fixing the bug where messages were being sent by the pubsubserver out of order	2024-06-27 01:32:18 -04:00
John Colvin	44d6173ddf	feat: Add DataStream WebApp Included in DataStreamSuite.sln, but not DataStreamQueryAnalyzer.sln	2024-06-26 14:33:04 -04:00
John Colvin	322152c743	build: Copy libspark to all dotnet dirs in single cmake target This is cleaner than the previous impl, which used separate build targets to copy libspark to "net7.0", and "net7.0-windows" (dirs). Note: I decided to clean this up now because I'm about to upgrade the new DataStream WebApp to net8.0, and did not want to add yet another build target to copy libspark to "net8.0".	2024-06-26 14:33:04 -04:00
John Colvin	9ab39587e2	fix: Parse ServerAddress as normal string The "JsonValueOrArrayConverter<T>" doesn't work the way I thought it would. When it is associated with a property via the JsonConverter attribute, dotnet json parsing code throws an exception if "T" isn't assignable to the property type.	2024-06-26 14:33:04 -04:00
John Colvin	72f59924ea	fix build (api server): Change namespace to bts.websocket_core	2024-06-26 14:33:04 -04:00
John Colvin	98ab9741b5	fix build: Update vcpkg On some vs versions, vcpkg cmake scripts would fail w/ an error saying "Could not locate a complete Visual Studio instance"	2024-06-26 14:33:04 -04:00
John Colvin	11f0143845	Remove docker/windows again?	2024-06-26 14:33:04 -04:00
John Colvin	35d47f9cce	Remove docker/linux again?	2024-06-26 14:33:04 -04:00
John Colvin	18662fcc0d	Change externals url from ssh to https style So we don't need ssh keys to clone it (the builtin credential manager makes https authentication safe/easy)	2024-06-26 14:33:04 -04:00
John Colvin	16ffc48846	cleanup: Fix benign typos in build scripts Change BUILD_SPARK_PY var name to BUILD_PY (because these scripts to actually build 'spark')	2024-06-26 14:33:04 -04:00
John Colvin	3cf388f041	fix build (warn): Only use Registry on windows	2024-06-26 14:33:03 -04:00
John Colvin	3e401a766d	cleanup: Fix numerous build warnings in DataStreamSuite Nearly all warnings were due to nullability issues, since the concept of nullable reference types did not exist when the DataStream code was originally written. * PubSubClientMessage - Normalize indentation/whitespace - Fix some typos - Address nullability issues to fix build warnings - Modernize code * Compression related files - ZipEntry - ZipFile - ZipInputStream - TarHeader * RemoteMessage - make 'message' nullable to resolve warnings elsewhere (in calling code) * PubSubSingleTagStateEvent.cs * PubSubMessageReactor * PubSubState * SystemHeartbeatPublisher * SystemStatsPublisher * SingleTagData * ManagedTCPMessageHub * PubsubDeserializationController * PublisherCore * InflaterHuffmanTree * AsyncTCPConnection_APM * ThreadSafeQueue * ManagedSubscriber * SubscriberCore * PubSubServerManagedTCPConnection * AsyncTCPConnection_APM * SuperMapStream * MapStream * TradeBarsStream	2024-06-26 14:33:03 -04:00
John Colvin	b16ff50120	cleanup: Normalize indentation in TradeBarsStream	2024-06-26 14:33:03 -04:00
John Colvin	44564f7d56	cleanup: Disable nullability warnings * BZip2InputStream * TarInputStream * Inflater * InflaterDynHeader * BZip2OutputStream * DeflaterHuffman * ParallelJoinTest	2024-06-26 14:33:03 -04:00
John Colvin	eaa5580e89	cleanup: Modernize PubSubRoute code, json parsing	2024-06-26 14:33:03 -04:00
John Colvin	5de4e31880	cleanup: Remove unused DateTime grammar from exp parser	2024-06-26 14:33:03 -04:00
John Colvin	7cf90d95da	Add JsonRpcRequestObservable, AsyncEnumerable	2024-06-26 14:33:03 -04:00
John Colvin	ec36d84412	Add json facilities for DataStream websocket api	2024-06-26 14:33:03 -04:00
John Colvin	95aad773a2	refactor: Add handy WebsocketExtensions * IsSocketStateOpen * IsClosedProperly	2024-06-26 14:33:03 -04:00
John Colvin	fb476de0a3	feat: Add JsonRpc types for websocket paradigm	2024-06-26 14:33:03 -04:00
John Colvin	41b42719ce	cleanup: Remove empty file (clearly added by accident)	2024-06-26 14:33:03 -04:00
John Colvin	c1f8aadd5d	fix: Fix improper 'OfType' linq cast Previously, animations were expressed through polymorphism (derived from the base class CellAnimation), but the new impl uses composition. This 'OfType' call was leftover from the polymorphism implementation, an attempt to linq cast elements of base class (CellAnimation) to BackgroundAnimation.	2024-06-26 14:33:03 -04:00
John Colvin	ad5e0c77fe	editorconfig: Use standard c# capitalization rules * Use 'PascalCase' for public members (instead of 'snake_case') * Use 'camelCase' for private members (instead of '_snake_case')	2024-06-26 14:33:03 -04:00
John Colvin	90ee45db30	refactor: Add WebsocketClient class Basically an ISubject (I/O) wrapper for IWebsocketConnectionGateway	2024-06-26 14:33:03 -04:00
John Colvin	6dd701dd33	refactor: Change WebsocketServerConfig.port to uint32 It should always be a number between 1-65535	2024-06-26 14:33:03 -04:00
John Colvin	456c953839	refactor: Rename WebsocketServerConfig proto pkg 'bts.websocket_core' This matches the namespace used in spark_websocket csproj. Eventually, I'd like to use conventional capitalization (TitleCase namespaces/types, etc.), but let's shelf that until we have enough time/energy to do a more comprehensive refactoring.	2024-06-26 14:33:03 -04:00
John Colvin	492f02eb21	refactor: Remove nullability constraint in GetValueOrDefault	2024-06-26 14:33:03 -04:00
John Colvin	41d2fd9c51	refactor: Use events to avoid circular deps for ws sync/gateway	2024-06-26 14:33:03 -04:00
John Colvin	59f3c446ba	refactor: Let caller specify nullability of T in json parse fn	2024-06-26 14:33:03 -04:00
Taha	9aadda2429	some buf gixes + cleanup	2024-06-26 14:29:22 -04:00
Taha	57ac078469	PubSuMessageReactor empties the message queue (instead oif one elem ata a time) in the main loop	2024-06-26 14:28:52 -04:00
Taha	c39fc70a64	PubSubServer's linux issues fixed	2024-06-26 14:28:51 -04:00