diff --git a/src/DataStreamSuite/SharedStreams/Business/MarketData/MDStream.cs b/src/DataStreamSuite/SharedStreams/Business/MarketData/MDStream.cs index 643146112..e334c3aba 100644 --- a/src/DataStreamSuite/SharedStreams/Business/MarketData/MDStream.cs +++ b/src/DataStreamSuite/SharedStreams/Business/MarketData/MDStream.cs @@ -1,7 +1,9 @@ namespace BCG { using BCG.Utilities.General; + using bts; using bts.api; + using bts.api.core; using bts.api.domain; using bts.api.synthetic; using bts.md; @@ -14,6 +16,7 @@ namespace BCG { using mbts.spark; using mbts.ui; + using Spark.Api.MarketData; using Spark.Core.Data; using Spark.Core.Disposables; using Spark.Core.Reactive; @@ -39,6 +42,8 @@ namespace BCG { new ColumnInfo("sec", 1, typeof(string))}, "md"); private BlueTable _rows; + readonly MarketDataClient mdClient; + IDisposable? mdSub; public MDStream() : this(String.Empty, ViewStreamDatabase.instance) { @@ -49,14 +54,27 @@ namespace BCG { _symbol = symbol; _rows = new BlueTable(_schema); _md_source = new MDSource(TimeSpan.FromSeconds(0.5)); + mdClient = new(_md_source); } protected override void myOnFirstSubscriber() { _vsd.addWorker(this); _md_source.subscribe_securities().Subscribe(c => on_md(c)); + Log.isnull(mdSub); + mdSub = mdClient.books().Subscribe(onBooks); //base.myOnFirstSubscriber(); } + void onBooks(TableUpdate update) { + foreach (var book in update.added) { + bts.Log.Gen.info($"MDStream.on_md sec: {book.security}"); + var sec = book.security; + IBlueRow r = new BlueRow(_schema, _id++, sec.ToString()); + _rows.addRow(r); + _is_dirty = true; + } + } + protected override void myOnLastSubscriber() { //base.myOnLastSubscriber(); _vsd.removeWorker(this); @@ -82,4 +100,4 @@ namespace BCG { } } } -} \ No newline at end of file +}