Merged in dripdrip (pull request #83)

Dripdrip

Approved-by: John Colvin <john.colvin@bluetradesys.com>

release_notes.txt

@@ -1,3 +1,7 @@
+1.3.1
+- Improved server side logging
+- Fixed leaking of market data subscriptions on market view form closing
+
 1.3.0
 - Significantly improved order execution latency
 - Added latency monitoring tools

src/client/spark/Daemons/SecurityPickerDaemon.cs

@@ -53,8 +53,8 @@ namespace bts {
          }
          //
          void handle_bad_icon(string icon_error_desc) {
-            Log.Api.warn($"{nameof(SecurityPickerDaemon)} received a message '{message.title}' with {icon_error_desc}, will use {nameof(Icons.AdminFrozen)}.");
-            icon = Icons.AdminFrozen;
+            Log.Api.warn($"{nameof(SecurityPickerDaemon)} received a message '{message.title}' with {icon_error_desc}, will use {nameof(Icons.Admin)}.");
+            icon = Icons.Admin;
          }
          //
          var cpv = new ContractPickerView(vm);

src/client/spark/Images/Icons.cs

@@ -16,11 +16,6 @@ namespace mbts.ui {
          ToolsGeometryKey = nameof(ToolsGeometryKey),
          DraggableGeometryKey = nameof(DraggableGeometryKey);
 
-      static Icons() {
-         // TODO: probably do this for all icons, only reason for holding back is the uncertainity of any unexpected impacts
-         AdminFrozen = Admin.frozen();
-      }
-
       public static Geometry PlayGeometry => Geometry.Parse(_get_geometric_path(PlayGeometryKey).FirstOrDefault() ?? string.Empty);
 
       static readonly string base_uri = "pack://application:,,,/Images/Icons/";
@@ -32,40 +27,43 @@ namespace mbts.ui {
 
       [Obsolete("Please only add new vector icons using icons.xaml")]
       private static ImageSource load_image(string filename) {
-         return new BitmapImage(uri(filename));
+         return new BitmapImage(uri(filename)).frozen();
+      }
+
+      private static ImageSource load_resource_and_freeze(string resource_name) {
+         return ((DrawingImage)_resources[resource_name]).frozen();
       }
 
 #pragma warning disable CS0618 // Type or member is obsolete
       public static ImageSource Logo { get { return load_image("logo.ico"); } }
-      public static ImageSource Admin { get { return (DrawingImage)_resources["Admin"]; } }
-      public static ImageSource AdminFrozen { get; }
-      public static ImageSource Analytics { get { return (DrawingImage)_resources["Analytics"]; } }
-      public static ImageSource Sheets { get { return (DrawingImage)_resources["Sheets"]; } }
-      public static ImageSource Trades { get { return (DrawingImage)_resources["Trades"]; } }
-      public static ImageSource Depth { get { return (DrawingImage)_resources["Depth"]; } }
-      public static ImageSource Ladder { get { return (DrawingImage)_resources["Ladder"]; } }
+      public static ImageSource Admin { get { return load_resource_and_freeze("Admin"); } }
+      public static ImageSource Analytics { get { return load_resource_and_freeze("Analytics"); } }
+      public static ImageSource Sheets { get { return load_resource_and_freeze("Sheets"); } }
+      public static ImageSource Trades { get { return load_resource_and_freeze("Trades"); } }
+      public static ImageSource Depth { get { return load_resource_and_freeze("Depth"); } }
+      public static ImageSource Ladder { get { return load_resource_and_freeze("Ladder"); } }
       public static ImageSource SheetsRed { get { return load_image("SheetsRed.png"); } }
       public static ImageSource SheetsGreen { get { return load_image("SheetsGreen.png"); } }
-      public static ImageSource Spreads { get { return (DrawingImage)_resources["Spreads"]; } }
+      public static ImageSource Spreads { get { return load_resource_and_freeze("Spreads"); } }
       public static ImageSource BlueSpreads { get { return load_image("BlueSpread.ico"); } }
-      public static ImageSource Reports { get { return (DrawingImage)_resources["Reports"]; } }
-      public static ImageSource Risk { get { return (DrawingImage)_resources["Risk"]; } }
-      public static ImageSource Eye { get { return (DrawingImage)_resources["Eye"]; } }
-      public static ImageSource Coins { get { return (DrawingImage)_resources["Coins"]; } }
+      public static ImageSource Reports { get { return load_resource_and_freeze("Reports"); } }
+      public static ImageSource Risk { get { return load_resource_and_freeze("Risk"); } }
+      public static ImageSource Eye { get { return load_resource_and_freeze("Eye"); } }
+      public static ImageSource Coins { get { return load_resource_and_freeze("Coins"); } }
 
 
       public static ImageSource DataCopy { get { return load_image("DataCopy.ico"); } }
       public static ImageSource Refresh { get { return load_image("Refresh.png"); } }
       public static ImageSource FavoriteWindow { get { return load_image("FavoriteWindow.png"); } }
 
-      public static ImageSource Browser => (DrawingImage)_resources["Browser"];
+      public static ImageSource Browser => load_resource_and_freeze("Browser");
 
       public static ImageSource Date { get { return load_image("Date.png"); } }
 
-      public static ImageSource Cube3d { get { return (DrawingImage)_resources["Cube"]; } }
+      public static ImageSource Cube3d { get { return load_resource_and_freeze("Cube"); } }
       public static ImageSource Algo { get { return Cube3d; } }
       public static ImageSource Quote { get { return Math; } }
-      public static ImageSource Math { get { return (DrawingImage)_resources["Surface"]; } }
+      public static ImageSource Math { get { return load_resource_and_freeze("Surface"); } }
       public static ImageSource Abacus { get { return load_image("Abacus.png"); } }
 
       public static ImageSource Delete { get { return load_image("Delete.png"); } }
@@ -77,58 +75,58 @@ namespace mbts.ui {
 #pragma warning restore CS0618
 
       // Vector icons
-      public static ImageSource Bookmarks { get { return (DrawingImage)_resources["Bookmarks"]; } }
-      public static ImageSource Window { get { return (DrawingImage)_resources["Window"]; } }
-      public static ImageSource ActiveWindow { get { return (DrawingImage)_resources["ActiveWindow"]; } }
-      public static ImageSource Trash { get { return (DrawingImage)_resources["Trash"]; } }
-      public static ImageSource Trash2 { get { return (DrawingImage)_resources["Trash2"]; } }
-      public static ImageSource Crucible { get { return (DrawingImage)_resources["Crucible"]; } }
+      public static ImageSource Bookmarks { get { return load_resource_and_freeze("Bookmarks"); } }
+      public static ImageSource Window { get { return load_resource_and_freeze("Window"); } }
+      public static ImageSource ActiveWindow { get { return load_resource_and_freeze("ActiveWindow"); } }
+      public static ImageSource Trash { get { return load_resource_and_freeze("Trash"); } }
+      public static ImageSource Trash2 { get { return load_resource_and_freeze("Trash2"); } }
+      public static ImageSource Crucible { get { return load_resource_and_freeze("Crucible"); } }
 
-      public static ImageSource Accounting { get { return (DrawingImage)_resources["Accounting"]; } }
-      public static ImageSource Add { get { return (DrawingImage)_resources["Add"]; } }
-      public static ImageSource AddEncircled { get { return (DrawingImage)_resources["AddEncircled"]; } }
-      public static ImageSource Beaker { get { return (DrawingImage)_resources["Beaker"]; } }
-      public static ImageSource Book { get { return (DrawingImage)_resources["Book"]; } }
+      public static ImageSource Accounting { get { return load_resource_and_freeze("Accounting"); } }
+      public static ImageSource Add { get { return load_resource_and_freeze("Add"); } }
+      public static ImageSource AddEncircled { get { return load_resource_and_freeze("AddEncircled"); } }
+      public static ImageSource Beaker { get { return load_resource_and_freeze("Beaker"); } }
+      public static ImageSource Book { get { return load_resource_and_freeze("Book"); } }
 
       public static ImageSource Orders => create_icon(OrdersGeometryKey);
-      public static ImageSource AddOrders { get { return (DrawingImage)_resources["AddOrders"]; } }
-      public static ImageSource CancelOrders { get { return (DrawingImage)_resources["CancelOrders"]; } }
-      public static ImageSource CancelBuys { get { return (DrawingImage)_resources["CancelBuys"]; } }
-      public static ImageSource CancelSells { get { return (DrawingImage)_resources["CancelSells"]; } }
-      public static ImageSource CancelSelected { get { return (DrawingImage)_resources["CancelSelected"]; } }
+      public static ImageSource AddOrders { get { return load_resource_and_freeze("AddOrders"); } }
+      public static ImageSource CancelOrders { get { return load_resource_and_freeze("CancelOrders"); } }
+      public static ImageSource CancelBuys { get { return load_resource_and_freeze("CancelBuys"); } }
+      public static ImageSource CancelSells { get { return load_resource_and_freeze("CancelSells"); } }
+      public static ImageSource CancelSelected { get { return load_resource_and_freeze("CancelSelected"); } }
       
-      public static ImageSource LadderCancelOrders { get { return (DrawingImage)_resources["LadderCancelOrders"]; } }
-      public static ImageSource LadderCancelBuys { get { return (DrawingImage)_resources["LadderCancelBuys"]; } }
-      public static ImageSource LadderCancelSells { get { return (DrawingImage)_resources["LadderCancelSells"]; } }
+      public static ImageSource LadderCancelOrders { get { return load_resource_and_freeze("LadderCancelOrders"); } }
+      public static ImageSource LadderCancelBuys { get { return load_resource_and_freeze("LadderCancelBuys"); } }
+      public static ImageSource LadderCancelSells { get { return load_resource_and_freeze("LadderCancelSells"); } }
 
-      public static ImageSource Synthetics { get { return (DrawingImage)_resources["Synthetics"]; } } 
-      public static ImageSource AddSynthetic { get { return (DrawingImage)_resources["AddSynthetic"]; } }
-      public static ImageSource RemoveSynthetic { get { return (DrawingImage)_resources["RemoveSynthetic"]; } }
-      public static ImageSource ModifySynthetic { get { return (DrawingImage)_resources["ModifySynthetic"]; } }
-      public static ImageSource CloneSynthetic { get { return (DrawingImage)_resources["CloneSynthetic"]; } }
+      public static ImageSource Synthetics { get { return load_resource_and_freeze("Synthetics"); } } 
+      public static ImageSource AddSynthetic { get { return load_resource_and_freeze("AddSynthetic"); } }
+      public static ImageSource RemoveSynthetic { get { return load_resource_and_freeze("RemoveSynthetic"); } }
+      public static ImageSource ModifySynthetic { get { return load_resource_and_freeze("ModifySynthetic"); } }
+      public static ImageSource CloneSynthetic { get { return load_resource_and_freeze("CloneSynthetic"); } }
 
-      public static ImageSource ModifyOrder { get { return (DrawingImage)_resources["ModifyOrder"]; } }
-      public static ImageSource Clock { get { return (DrawingImage)_resources["Clock"]; } }
-      public static ImageSource Defrag { get { return (DrawingImage)_resources["Defrag"]; } }
-      public static ImageSource Display { get { return (DrawingImage)_resources["Display"]; } }
+      public static ImageSource ModifyOrder { get { return load_resource_and_freeze("ModifyOrder"); } }
+      public static ImageSource Clock { get { return load_resource_and_freeze("Clock"); } }
+      public static ImageSource Defrag { get { return load_resource_and_freeze("Defrag"); } }
+      public static ImageSource Display { get { return load_resource_and_freeze("Display"); } }
       public static ImageSource Draggable => create_icon(DraggableGeometryKey);
-      public static ImageSource Export => (DrawingImage)_resources["Export"];
-      public static ImageSource Filter { get { return (DrawingImage)_resources["Filter"]; } }
-      public static ImageSource FilterActive { get { return (DrawingImage)_resources["FilterActive"]; } }
-      public static ImageSource FilterError { get { return (DrawingImage)_resources["FilterError"]; } }
+      public static ImageSource Export => load_resource_and_freeze("Export");
+      public static ImageSource Filter { get { return load_resource_and_freeze("Filter"); } }
+      public static ImageSource FilterActive { get { return load_resource_and_freeze("FilterActive"); } }
+      public static ImageSource FilterError { get { return load_resource_and_freeze("FilterError"); } }
       public static ImageSource Gear => create_icon(GearGeometryKey);
-      public static ImageSource Import => (DrawingImage)_resources["Import"];
-      public static ImageSource Knight { get { return (DrawingImage)_resources["Knight"]; } }
+      public static ImageSource Import => load_resource_and_freeze("Import");
+      public static ImageSource Knight { get { return load_resource_and_freeze("Knight"); } }
       public static ImageSource Payment => create_icon(PaymentGeometryKey);
-      public static ImageSource Spawn { get { return (DrawingImage)_resources["Spawn"]; } }
-      public static ImageSource Target { get { return (DrawingImage)_resources["Target"]; } }
-      public static ImageSource Tetris { get { return (DrawingImage)_resources["Tetris"]; } }
+      public static ImageSource Spawn { get { return load_resource_and_freeze("Spawn"); } }
+      public static ImageSource Target { get { return load_resource_and_freeze("Target"); } }
+      public static ImageSource Tetris { get { return load_resource_and_freeze("Tetris"); } }
       public static ImageSource Tools => create_icon(ToolsGeometryKey);
-      public static ImageSource ViewSettings => (DrawingImage)_resources["ViewSettings"];
-      public static ImageSource Wrench { get { return (DrawingImage)_resources["Wrench"]; } }
-      public static ImageSource RightArrow { get { return ((DrawingImage)_resources["RightArrow"]); } }
-      public static ImageSource DownArrow { get { return ((DrawingImage)_resources["DownArrow"]); } }
-      public static ImageSource Support => (DrawingImage)_resources["SupportIcon"];
+      public static ImageSource ViewSettings => load_resource_and_freeze("ViewSettings");
+      public static ImageSource Wrench { get { return load_resource_and_freeze("Wrench"); } }
+      public static ImageSource RightArrow { get { return (load_resource_and_freeze("RightArrow")); } }
+      public static ImageSource DownArrow { get { return (load_resource_and_freeze("DownArrow")); } }
+      public static ImageSource Support => load_resource_and_freeze("SupportIcon");
 
       public static ImageSource create_icon(string geometric_path_key) =>
          create_icon(geometric_path_key, Brushes.Black);

src/client/spark/Modules/Settings/SettingsShellModule.cs

@@ -42,7 +42,7 @@ namespace bts.ui {
 
          _bus = bus;
       }
-      public ImageSource icon => Icons.AdminFrozen;
+      public ImageSource icon => Icons.Admin;
 
       public string name => "Settings";
 

src/client/spark/Modules/Trading/OrderBook/OrderBookRowVM.cs

@@ -30,7 +30,6 @@ namespace mbts.spark {
       string _symbol;
       OrderBookRowVM _parent;
       OrderBookVM.BranchFilterSource _branch_filter_source;
-      IDisposable _branch_filter_sub;
       // Core
       IOrderBookCellStore<uint> _order_key_store;
       IOrderBookCellStore<uint> _parent_order_key_store;
@@ -114,15 +113,11 @@ namespace mbts.spark {
          _show_order_modify_cmd = new ObservantCommand(() => bus.post(SparkTags.ShowModifyOrderTag, _order.Key));
          _show_order_modify_cmd.OnNext(status.value == OrderStatus.Working);
          row_passes_filter = _book_vm.columns.filter(this);
-         _passes_filter = backing(nameof(passes_filter),
-            Observable.FromEvent<Action<IObservable<bool>>, IObservable<bool>>(
-               cb => _branch_filter_source_changed += cb, cb => _branch_filter_source_changed -= cb)
-         .SelectMany(o => o));
+         _passes_filter = backing(nameof(passes_filter), row_passes_filter);
       }
 
       public event Action invalidated;
       public event Action<(OrderRecordTuple old_val, OrderRecordTuple new_val)> updated;
-      event Action<IObservable<bool>> _branch_filter_source_changed;
 
       public OrderRecordImmutable order => _order;
       public OrderMetadata order_metadata => _model.factory.get_metadata(order_type.value);
@@ -228,11 +223,15 @@ namespace mbts.spark {
       }
 
       public void join_branch_filter_source(OrderBookVM.BranchFilterSource new_branch_filter_source) {
+         if (_branch_filter_source != null) {
+            _branch_filter_source.passes_filter_changed -= _passes_filter.set;
+         }
          _branch_filter_source = new_branch_filter_source;
          _branch_filter_source.set_row_passes_filter(_order.Key, row_passes_filter);
-         _branch_filter_source_changed?.Invoke(_branch_filter_source);
          foreach (var child_row in _child_rows)
             child_row.join_branch_filter_source(_branch_filter_source);
+         _passes_filter.set(_branch_filter_source.passes_filter);
+         _branch_filter_source.passes_filter_changed += _passes_filter.set;
       }
 
       /// <summary>
@@ -311,7 +310,13 @@ namespace mbts.spark {
       public override int GetHashCode() => (int)_order.Key;
       public override string ToString() => $"{order_key} {last_change} {parent_order_key}";
 
-      protected override void Dispose(bool disposing) => Disposable.dispose(ref _branch_filter_sub);
+      protected override void Dispose(bool disposing) {
+         // idempotent
+         if (_branch_filter_source != null) {
+            _branch_filter_source.passes_filter_changed -= _passes_filter.set;
+            _branch_filter_source = null;
+         }
+      }
 
       static IEnumerable<OrderBookRowVM> _descendents(OrderBookRowVM root) {
          foreach (var child in root.child_orders) {

src/client/spark/Modules/Trading/OrderBook/OrderBookVM.cs

@@ -27,10 +27,7 @@ namespace mbts.spark {
    partial class OrderBookVM : DisposableViewModelBase, IHandleRowSelection {
       readonly LiveSortingColumKeys _live_sorting_columns;
       readonly Property<OrderBookRowVM> _current_row;
-      readonly Subject<(IEnumerable<SecurityImmutable> added, IEnumerable<SecurityImmutable> removed)> _order_securities_subject =
-         new Subject<(IEnumerable<SecurityImmutable> added, IEnumerable<SecurityImmutable> removed)>();
       OrderBookStartupData _state;
-
       Dictionary<uint, OrderBookRowVM> _order_rows_cache = new Dictionary<uint, OrderBookRowVM>();
       Dictionary<uint, BranchFilterSource> _branch_filter_roots = new Dictionary<uint, BranchFilterSource>();
       ObservableCollection<OrderBookRowVM> _order_rows = new ObservableCollection<OrderBookRowVM>();
@@ -39,7 +36,8 @@ namespace mbts.spark {
       SetMap<uint, uint> _security_orders = new SetMap<uint, uint>();
 
       public OrderBookVM(IObservable<string> owner_title, IBus bus, INotificationService notification_service, OrdersModel model) {
-         columns = new OrderBookColumnCollection(bus, model, _order_securities_subject);
+         var order_securities_subject = new Subject<(IEnumerable<SecurityImmutable> added, IEnumerable<SecurityImmutable> removed)>();
+         columns = new OrderBookColumnCollection(bus, model, order_securities_subject);
          rows_view = CollectionViewSource.GetDefaultView(_order_rows);
          rows_view.Filter = obj => obj is OrderBookRowVM vm && vm.is_visible && vm.passes_filter;
          columns.filter_changed += () => {
@@ -73,7 +71,8 @@ namespace mbts.spark {
             bus.subscribe(SparkTags.SecuritiesTag).OnDispatcher().Subscribe(on_securities),
             model.on_orders.OnDispatcher().AddRemoveOnly().Subscribe(on_orders),
             grid_settings_manager,
-            columns);
+            columns,
+            order_securities_subject);
 
          void on_orders(LegacyTableUpdate<OrderRecordImmutable, uint> table) {
             // clear everything (most likely the server restarted)
@@ -170,7 +169,7 @@ namespace mbts.spark {
                _order_rows.Add(row);
             }
             if (added_securities.Count > 0 || removed_securities.Count > 0)
-               _order_securities_subject.OnNext((added_securities, removed_securities));
+               order_securities_subject.OnNext((added_securities, removed_securities));
 
             void try_add_children(OrderBookRowVM row) {
                if (_parent_to_children.try_lookup(row.order.Key, out ISet<uint> children))
@@ -206,7 +205,7 @@ namespace mbts.spark {
                }
             }
             if (added_securities.Count > 0)
-               _order_securities_subject.OnNext((added_securities, new List<SecurityImmutable>(0)));
+               order_securities_subject.OnNext((added_securities, new List<SecurityImmutable>(0)));
          }
       }
 
@@ -243,6 +242,9 @@ namespace mbts.spark {
       internal void remove_branch_filter_root(uint key) =>
          _branch_filter_roots.Remove(key);
 
+      protected override void Dispose(bool disposing) =>
+         Disposable.dispose(_order_rows);
+
       static GridSettingsManager<OrderBookRowVM, OrderRecordTuple, INotificationServiceDataCache, OrderRecordTuplePredicate> _create_grid_settings_manager(
          IObservable<string> owner_title, IBus bus, INotificationService notification_service, OrdersModel model,
          ObservableCollection<OrderBookRowVM> rows, ICollectionView rows_view, ObservableCollection<DIYGrid.ColumnBase> columns) {
@@ -297,24 +299,21 @@ namespace mbts.spark {
          }
       }
 
-      internal class BranchFilterSource : IObservable<bool> {
+      internal class BranchFilterSource {
          readonly HashSet<uint> _orders_pass_filter = new HashSet<uint>();
-         event Action<bool> _passes_filter_changed;
+
+         public event Action<bool> passes_filter_changed;
+
+         public bool passes_filter => _orders_pass_filter.Count > 0;
 
          public void set_row_passes_filter(uint key, bool value) {
             if (value) {
                if (_orders_pass_filter.Add(key) && _orders_pass_filter.Count == 1)
-                  _passes_filter_changed?.Invoke(value);
+                  passes_filter_changed?.Invoke(value);
             } else if (_orders_pass_filter.Remove(key) && _orders_pass_filter.Count == 0) {
-               _passes_filter_changed?.Invoke(value);
+               passes_filter_changed?.Invoke(value);
             }
          }
-
-         public IDisposable Subscribe(IObserver<bool> observer) {
-            _passes_filter_changed += observer.OnNext;
-            observer.OnNext(_orders_pass_filter.Count > 0);
-            return Disposable.Create(() => _passes_filter_changed -= observer.OnNext);
-         }
       }
 
       class LiveSortingColumKeys : HashSet<Columns> {

src/client/spark/core/Bus/SparkCommands.cs

@@ -70,7 +70,7 @@ namespace bts.spark
       public static SparkCommand ShowWorkspaces =>
          SparkCommand.create("Workspaces", Icons.FavoriteWindow, SparkTags.ShowWorkspaces);
       public static SparkCommand ShowSettings =>
-         SparkCommand.create("Settings", Icons.AdminFrozen, SparkTags.ShowSettings);
+         SparkCommand.create("Settings", Icons.Admin, SparkTags.ShowSettings);
 
       // Support
       public static SparkCommand ShowManual =>
@@ -113,12 +113,12 @@ namespace bts.spark
 
       // TODO: remove this method, eventually want all windows to use window_settings_command()
       public static SparkCommand change_securities_command(Func<(PartialCommand<SecurityCriteria>, SecurityCriteria)> get_param) =>
-         SparkCommand.create("Settings", Icons.AdminFrozen, SparkTags.ShowAndEditPickMultiple, get_param);
+         SparkCommand.create("Settings", Icons.Admin, SparkTags.ShowAndEditPickMultiple, get_param);
 
       public static PartialCommand<SecurityCriteria> change_securities_callback(string window_name) =>
-         SparkCommand.create(window_name, Icons.AdminFrozen, new SparkTags.SparkTag<SecurityCriteria>($"Change {window_name} securities callback"));
+         SparkCommand.create(window_name, Icons.Admin, new SparkTags.SparkTag<SecurityCriteria>($"Change {window_name} securities callback"));
 
       public static SparkCommand window_settings_command(IDialogCommand launch_settings_editor) =>
-         new SparkCommandWrapper("Settings", Icons.AdminFrozen, launch_settings_editor);
+         new SparkCommandWrapper("Settings", Icons.Admin, launch_settings_editor);
    }
 }

src/client/spark/core/Utility/KeyDrivenObservable.cs

@@ -66,15 +66,73 @@ namespace mbts.utils {
 
       class AnonymousConnectable<TResult> : IConnectableObservable<TResult> {
          readonly IObservable<TSource> _source;
-         readonly KeyDrivenObservableSeries<TSource, TResult> _driver;
+         readonly Func<TSource, IObservable<TResult>> _select_next;
+         List<LazySubscription> _lazy_subs = new List<LazySubscription>();
+         KeyDrivenObservableSeries<TSource, TResult> _driver;
+         IDisposable _driver_sub;
+         bool _is_connected = false;
 
          public AnonymousConnectable(IObservable<TSource> source, Func<TSource, IObservable<TResult>> select_next) {
             _source = source;
-            _driver = new KeyDrivenObservableSeries<TSource, TResult>(select_next);
+            _select_next = select_next;
          }
 
-         public IDisposable Connect() => _source.Subscribe(_driver);
-         public IDisposable Subscribe(IObserver<TResult> observer) => _driver.Subscribe(observer);
+         // idempotent
+         public IDisposable Connect() {
+            lock (this) {
+               if (!_is_connected) {
+                  _driver = new KeyDrivenObservableSeries<TSource, TResult>(_select_next);
+                  foreach (var ls in _lazy_subs)
+                     ls.attach();
+                  _lazy_subs.Clear();
+                  _driver_sub = _source.Subscribe(_driver);
+                  _is_connected = true;
+               }
+            }
+            return Disposable.Create(_deconnect);
+         }
+
+         void _deconnect() {
+            lock (this) {
+               if (_is_connected) {
+                  _driver_sub.Dispose();
+                  _driver.Dispose();
+                  _is_connected = false;
+               }
+            }
+         }
+
+         public IDisposable Subscribe(IObserver<TResult> observer) {
+            lock (this) {
+               if (_is_connected)
+                  return _driver.Subscribe(observer);
+               var ls = new LazySubscription(this, observer);
+               _lazy_subs.Add(ls);
+               return ls;
+            }
+            throw new InvalidOperationException();
+         }
+
+         class LazySubscription : IDisposable {
+            readonly IObserver<TResult> _observer;
+            readonly AnonymousConnectable<TResult> _owner;
+            IDisposable _sub;
+
+            public LazySubscription(AnonymousConnectable<TResult> owner, IObserver<TResult> observer) {
+               _owner = owner;
+               _observer = observer;
+            }
+
+            public void attach() =>
+               _sub = _owner._driver.Subscribe(_observer);
+
+            public void Dispose() {
+               lock (_owner) {
+                  Disposable.dispose(ref _sub);
+                  _owner._lazy_subs.Remove(this);
+               }
+            }
+         }
       }
    }
 

src/client/spark/view/Shell/FloatingWindow.xaml.cs

@@ -154,6 +154,7 @@ namespace mbts.ui.shell
          foreach (TabDocument document in _tabs.Items.OfType<TabDocument>().ToList()) {
             if (document != null && document.ViewModel.CloseCommand.CanExecute(this)) {
                document.ViewModel.CloseCommand.Execute(this);
+               document.ViewModel.Closing -= TabClosing;
             }
          }
       }

src/client/spark/view/Shell/ViewModels/ShellThreadFactory.cs

@@ -10,6 +10,7 @@ using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using System.Windows.Threading;
+using System.Windows.Interop;
 
 namespace mbts.ui.shell
 {
@@ -19,9 +20,8 @@ namespace mbts.ui.shell
       class UIThreadData : IDisposable {
          public Dispatcher dispatcher;
          public Thread thread;
-         public List<ISlowWindow> windows = new List<ISlowWindow>();
-         public System.Windows.Interop.MSG last_msg;
-         public void Dispose() { lock(windows) Disposable.dispose(windows); }
+         public MSG last_msg;
+         public void Dispose() {}
       }
       int _count = 0;
       ConcurrentDictionary<string, UIThreadData> _shared_threads = new ConcurrentDictionary<string, UIThreadData>();
@@ -38,7 +38,7 @@ namespace mbts.ui.shell
          uid.thread = new Thread(() => {
             uid.dispatcher = Dispatcher.CurrentDispatcher;
             uid.dispatcher.UnhandledException += BTSApplicationBootstrapper.CurrentDispatcher_HandleClipboardError;
-            System.Windows.Interop.ComponentDispatcher.ThreadFilterMessage += (ref System.Windows.Interop.MSG m,ref bool h) => ComponentDispatcher_ThreadFilterMessage(uid, ref m, ref h);
+            ComponentDispatcher.ThreadFilterMessage += (ref MSG m,ref bool h) => ComponentDispatcher_ThreadFilterMessage(uid, ref m, ref h);
             try {
                Dispatcher.Run();
             }
@@ -62,7 +62,7 @@ namespace mbts.ui.shell
          return uid;
       }
 
-      private void ComponentDispatcher_ThreadFilterMessage(UIThreadData d, ref System.Windows.Interop.MSG msg, ref bool handled) {
+      private void ComponentDispatcher_ThreadFilterMessage(UIThreadData d, ref MSG msg, ref bool handled) {
          d.last_msg = msg;
 
          // We definitely don't need to reload themes on background threads
@@ -75,7 +75,6 @@ namespace mbts.ui.shell
 
       private Dispatcher attach_window(UIThreadData uid, ISlowWindow window) {
          uid.dispatcher.BeginInvoke(DispatcherPriority.Send, (Action)(() => {
-            lock(uid.windows) uid.windows.Add(window);
             window.thread_init();
          }));
          return uid.dispatcher;

src/client/spark/view/Shell/ViewModels/ShellWindowVM.cs

@@ -29,8 +29,8 @@ namespace mbts.ui.shell
       public ShellWindowVM(IShellWindow model, IWindowManager wm, IShellThreadFactory threads, string host = "") {
          Log.assert(model != null);
          _model = model;
-         _model.TitleChanged += () => { if (!HasCustomTitle) RaisePropertyChanged("Title"); };
-         _model.IconChanged += () => { _icon = null; RaisePropertyChanged("Icon"); };
+         _model.TitleChanged += title_changed;
+         _model.IconChanged += icon_changed;
          _wm = wm;
          host_id = host;
 
@@ -46,6 +46,16 @@ namespace mbts.ui.shell
          _edit_mode = backing(nameof(EditMode), false);
       }
 
+      void title_changed() {
+         if (!HasCustomTitle) {
+            RaisePropertyChanged("Title");
+         }
+      }
+      void icon_changed() {
+         _icon = null;
+         RaisePropertyChanged("Icon");
+      }
+
       public string host_id { get; set; }
       public Dispatcher Dispatcher { get { return _dispatcher; } }
 
@@ -208,6 +218,9 @@ namespace mbts.ui.shell
 
 
       public void Dispose() {
+         _model.TitleChanged -= title_changed;
+         _model.IconChanged -= icon_changed;
+
          if (IsShared) {
             // Shared threads continue running
             _dispatcher.Invoke(() => Disposable.dispose(ref _model));
@@ -221,6 +234,7 @@ namespace mbts.ui.shell
             Disposable.dispose(ref _model);
          }
 
+         _ui = null;
          Disposable.dispose(ref _hwnd_host);
       }
 

src/client/spark/view/Shell/WindowManager.cs

@@ -31,11 +31,13 @@ namespace mbts.ui.shell {
       readonly Lazy<IBookmarks> _bookmarks;
       IDisposable _dispose;
       object _lock = new object();
+      readonly IBus _bus;
 
       [ImportingConstructor]
       public WindowManager(
          IShellThreadFactory threads, Lazy<ISettingsCollector> settings, 
          ISavedWindowFactory factory, Lazy<IBookmarks> bookmarks, IBus bus) {
+         _bus = bus;
          _threads = threads;
          _window_settings = settings;
          _window_factory = factory;
@@ -92,11 +94,15 @@ namespace mbts.ui.shell {
       public IReadOnlyList<TabDocument> list_documents() => _documents.Values.ToList();
 
       public async Task add(IShellWindow window) {
-         if (!Application.Current.Dispatcher.CheckAccess())
-            await Application.Current.Dispatcher.InvokeAsync(() => add(window));
-         Log.Ui.info($"Opening window {window.diagnostic_name()}.");
-         lock (_lock) {
-            _documents[window] = _create_document(window);
+         if (window != null) {
+            if (!Application.Current.Dispatcher.CheckAccess())
+               await Application.Current.Dispatcher.InvokeAsync(() => add(window));
+            Log.Ui.info($"Opening window {window.diagnostic_name()}.");
+            lock (_lock) {
+               _documents[window] = _create_document(window);
+            }
+            // Prevent the bus from holding on to a reference to the window
+            _bus.post(SparkTags.ShowWindow, null);
          }
       }
 

src/client/spark/view/Spark/Ladder/LadderGridHost.cs

@@ -73,6 +73,10 @@ namespace mbts.spark {
          };
          RecenterCommand = new DelegateCommand(_scroll_to_center);
          DataContextChanged += (s, e) => _connect_vm();
+         Unloaded += (s, e) => {
+            Disposable.dispose(ref _settings_sub);
+            Disposable.dispose(ref _tick_sub);
+         };
       }
 
       enum ColumnType { Order, Bid, Price, Ask, Last, None }

src/client/spark/view/Spark/SparkWindowView.xaml.cs

@@ -36,7 +36,7 @@ namespace bts.view {
             _menu.ShowText = false;
             _grid.Children.Add(_menu);
             if (ContextMenu != null) {
-               ContextMenu.ItemsSource = commands.Select(c => new SparkCommandVM(c, bus));
+               ContextMenu.ItemsSource = commands.Select(c => new SparkCommandVM(c, bus)).ToList();
             }
          } else {
             expand_button.Visibility = Visibility.Collapsed;

src/lib/CMakeDeps.cmake

@@ -1 +1 @@
-set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/dynamic_orders.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/dynamic_orders.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/quickbus_cfg.proto;config/protos/spreader.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;jetfix/codec/FIXArray.cpp;jetfix/common/OutFIX.cpp;jetfix/connection/AsyncReader.cpp;jetfix/connection/IConnection.cpp;jetfix/connection/SocketConnection.cpp;jetfix/gateway/CoreGateway.cpp;jetfix/gateway/JetFIXGateway.cpp;jetfix/gateway/Session.cpp;jetfix/store/FIXFileStore.cpp;jetfix/store/session_name.cpp;jetfix/tests/async_reader_tests.cpp;jetfix/tests/decoding_tests.cpp;jetfix/tests/fix_encoding_stream_tests.cpp;jetfix/tests/fix_fragment_tests.cpp;jetfix/tests/jetfix_performance_tests.cpp;jetfix/tests/sample_fix.cpp;jetfix/tests/store_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;mercury/actors/IPollableActor.cpp;mercury/broker/Broker.cpp;mercury/broker/BrokerProxy.cpp;mercury/broker/ILocalBroker.cpp;mercury/broker/IRemoteBroker.cpp;mercury/common/mercury_common.cpp;mercury/events/TimerEvent.cpp;mercury/executor/Executor.cpp;mercury/topics/Topic.cpp;metrics/metrics.cpp;metrics/system_metrics.cpp;orderman/actors/AsyncFactory.cpp;orderman/actors/ExecutionGateway.cpp;orderman/actors/FillStore.cpp;orderman/actors/OrderStore.cpp;orderman/actors/SODPosition.cpp;orderman/actors/SecuritySrc.cpp;orderman/actors/ServiceBridge.cpp;orderman/actors/order_manager/OrderBuilder.cpp;orderman/actors/order_manager/OrderManager.cpp;orderman/actors/order_manager/OrderServicesData.cpp;orderman/events/CancelRejectEvent.cpp;orderman/events/NewExecutionReportEvent.cpp;orderman/limiter/LimitStore.cpp;orderman/limiter/Limiter.cpp;orderman/limiter/ProductLimitNode.cpp;orderman/limiter/ProductLimitTree.cpp;orderman/limiter/SecurityLimitNode.cpp;orderman/limiter/SecurityLimitTree.cpp;orderman/limiter/tests/limit_tests.Line truncated
+set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/dynamic_orders.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/threads.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/dynamic_orders.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/quickbus_cfg.proto;config/protos/spreader.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/threads.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;jetfix/codec/FIXArray.cpp;jetfix/common/OutFIX.cpp;jetfix/connection/AsyncReader.cpp;jetfix/connection/IConnection.cpp;jetfix/connection/SocketConnection.cpp;jetfix/gateway/CoreGateway.cpp;jetfix/gateway/JetFIXGateway.cpp;jetfix/gateway/Session.cpp;jetfix/store/FIXFileStore.cpp;jetfix/store/session_name.cpp;jetfix/tests/async_reader_tests.cpp;jetfix/tests/decoding_tests.cpp;jetfix/tests/fix_encoding_stream_tests.cpp;jetfix/tests/fix_fragment_tests.cpp;jetfix/tests/jetfix_performance_tests.cpp;jetfix/tests/sample_fix.cpp;jetfix/tests/store_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;mercury/actors/IPollableActor.cpp;mercury/broker/Broker.cpp;mercury/broker/BrokerProxy.cpp;mercury/broker/ILocalBroker.cpp;mercury/broker/IRemoteBroker.cpp;mercury/common/mercury_common.cpp;mercury/events/TimerEvent.cpp;mercury/executor/Executor.cpp;mercury/topics/Topic.cpp;metrics/metrics.cpp;metrics/system_metrics.cpp;orderman/actors/AsyncFactory.cpp;orderman/actors/ExecutionGateway.cpp;orderman/actors/FillStore.cpp;orderman/actors/OrderStore.cpp;orderman/actors/SODPosition.cpp;orderman/actors/SecuritySrc.cpp;orderman/actors/ServiceBridge.cpp;orderman/actors/order_manager/OrderBuilder.cpp;orderman/actors/order_manager/OrderManager.cpp;orderman/actors/order_manager/OrderServicesData.cpp;orderman/events/CancelRejectEvent.cpp;orderman/events/NewExecutionReportEvent.cpp;orderman/limiter/LimitStore.cpp;orderman/limiter/Limiter.cpp;orderman/limiter/ProductLimitNode.cpp;orderman/limiter/ProductLimitTree.cpp;orderman/limiter/SecurityLimitNode.cpp;orderman/limiter/SecurityLimitTree.cpp;ordLine truncated

src/lib/config/protos/mercury_cfg.proto

@@ -0,0 +1,7 @@
+syntax='proto2';
+
+package bts.config;
+
+message  Mercury {
+  optional uint32 queue_size = 1 [default=1000];
+}

src/lib/mercury/broker/BrokerProxy.cpp

@@ -1,15 +1,18 @@
 #include "lib/mercury/broker/BrokerProxy.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/mercury_cfg.pb.h"
 
 namespace bts {
 namespace mercury {
 
 BrokerProxy::BrokerProxy(ILocalBroker &real_broker)
-   :  _real_broker(real_broker), _id(real_broker.id()), _outgoing(1000), _incoming(1000), _action_pool()
-{}
+    : _real_broker(real_broker), _id(real_broker.id()),
+      _outgoing(config::get<config::Mercury>().queue_size()),
+      _incoming(config::get<config::Mercury>().queue_size()),
+      _action_pool() {}
 
 BrokerProxy::~BrokerProxy() {}
 
-
 void BrokerProxy::add_neighbour(IRemoteBroker &n) {
    auto m = _action_pool.construct();
    m->msg_type = BrokerMsgType::ADD_NEIGHBOUR;
@@ -87,8 +90,7 @@ void BrokerProxy::local_unsubscribe(LocalSubID lsid) {
    send(*m);
 }
 
-void BrokerProxy::schedule_action (std::function<void()> const&action)
-{
+void BrokerProxy::schedule_action(std::function<void()> const &action) {
    auto m = _action_pool.construct();
    m->msg_type = BrokerMsgType::EXECUTE_ACTION;
    m->action = action;

src/lib/mercury/broker/BrokerProxy.h

@@ -3,8 +3,8 @@
 #include "lib/mercury/broker/ILocalBroker.h"
 #include "lib/mercury/common/mercury_common.h"
 #include "lib/mercury/topics/Topic.h"
+#include "lib/utils/GrowableSPSCQueue.h"
 
-#include <boost/lockfree/spsc_queue.hpp>
 #include <boost/pool/object_pool.hpp>
 #include <memory>
 
@@ -59,6 +59,7 @@ class BrokerProxy : public ILocalBroker {
    void local_unsubscribe(LocalSubID) override;
    // ILocalBroker
 
+   void drain() { _outgoing.drain();}
    void schedule_action(std::function<void()> const &action);
 
    // these are just for performance, we want to allow for
@@ -117,6 +118,7 @@ class BrokerProxy : public ILocalBroker {
             bts_assert(false);
             break;
          }
+         _incoming.drain();
          _incoming.push(m);
       });
    }
@@ -124,8 +126,8 @@ class BrokerProxy : public ILocalBroker {
  private:
    ILocalBroker &_real_broker;
    const BrokerID _id;
-   boost::lockfree::spsc_queue<BrokerMsg *> _outgoing;
-   boost::lockfree::spsc_queue<BrokerMsg *> _incoming;
+   utils::GrowableSPSCQueue<BrokerMsg *> _outgoing;
+   utils::GrowableSPSCQueue<BrokerMsg *> _incoming;
    boost::object_pool<BrokerMsg> _action_pool;
 
    void send(BrokerMsg &m) {

src/lib/mercury/executor/Executor.cpp

@@ -9,8 +9,9 @@ namespace mercury {
 
 Executor::Executor() :
    _broker(),
-   _proxies(),
-   _actor_broker(make_proxy()),
+   _incoming_proxies(),
+   _my_proxy(make_proxy()),
+   _actor_broker(_my_proxy),
    _actors() {}
 
 Executor::~Executor() {
@@ -20,11 +21,10 @@ Executor::~Executor() {
  
 void Executor::init_neighbours(std::vector<Executor*> const&neighbours)
 {
-   std::vector<IRemoteBroker*> broker_neighbours;
    for (auto e : neighbours)
       if (e != this) 
-         broker_neighbours.push_back(&e->make_proxy());
-   for (auto n : broker_neighbours)
+         _outgoing_proxies.push_back(&e->make_proxy());
+   for (auto n : _outgoing_proxies)
       _broker.add_neighbour(*n);
 }
 
@@ -40,8 +40,8 @@ void Executor::activate(Actor *actor)
 
 BrokerProxy& Executor::make_proxy()
 {
-   _proxies.push_back(std::make_unique<BrokerProxy>(_broker));
-   return *_proxies.back();
+   _incoming_proxies.push_back(std::make_unique<BrokerProxy>(_broker));
+   return *_incoming_proxies.back();
 }
 
 } // namespace mercury

src/lib/mercury/executor/Executor.h

@@ -25,8 +25,11 @@ class Executor {
    void do_work() {
       for (auto p : _pollable_actors)
          p->on_poll();
-      for (auto &bp : _proxies)
+      for (auto &bp : _incoming_proxies)
          bp->do_work();
+      for (auto &bp : _outgoing_proxies)
+         bp->drain();
+      _my_proxy.drain();
    }
 
    void activate(Actor *actor);
@@ -34,7 +37,9 @@ class Executor {
  private:
 
    Broker _broker;
-   std::vector<std::unique_ptr<BrokerProxy>> _proxies;
+   std::vector<std::unique_ptr<BrokerProxy>> _incoming_proxies;
+   std::vector<BrokerProxy*> _outgoing_proxies;
+   BrokerProxy &_my_proxy;
    ActorBroker _actor_broker;
    std::vector<IPollableActor *> _pollable_actors;
    std::vector<std::unique_ptr<Actor>> _actors;

src/lib/orderman/orders/NativeOrder.cpp

@@ -1032,8 +1032,13 @@ void NativeOrder::transition(api::orderman::OrderStatus new_status,
 void NativeOrder::transition(api::orderman::OrderStatus new_status) {
    auto const &from = OrderStatus_Name(me->state.status());
    auto const &to = OrderStatus_Name(new_status);
-   LOG(OrderMan, Info, "Transitioning order {} from {} to {}", key().value(),
-       from, to);
+   if (from != to)
+      LOG(OrderMan, Info, "Transitioning order {} from {} to {}", key().value(),
+          from, to);
+   else
+      LOG(OrderMan, Diag, "Transitioning order {} from {} to {}", key().value(),
+          from, to);
+      
    me->state.set_status(new_status);
 }
 

src/lib/orderman/orders/spreader_orders/SpreaderOrder.cpp

@@ -202,7 +202,8 @@ void SpreaderOrder::_switch_state (ISpreaderStateWorker* new_state){
 
 void SpreaderOrder::_on_init_new(const bts::api::orderman::OrderDefinition& order_def,
                                  const SpreaderPreActivationOrderServices& spreader_pao_services){
-   LOG(OrderMan, Info, "SpreaderOrder.on_init_new key:{} order_def:{}",
+   LOG(OrderMan, Info, "SpreaderOrder.on_init_new key:{}", key().value());
+   LOG(OrderMan, Diag, "SpreaderOrder.on_init_new key:{} order_def:{}",
        key().value(), bts::utils::pb::to_string(order_def));
    bts_assert(!_inited());   
    _state_worker->on_init_new(order_def, spreader_pao_services);

src/lib/utils/GrowableSPSCQueue.cpp

@@ -0,0 +1,2 @@
+#include "GrowableSPSCQueue.h"
+ 

src/lib/utils/GrowableSPSCQueue.h

@@ -0,0 +1,50 @@
+#pragma once
+
+#include <boost/lockfree/spsc_queue.hpp>
+#include <queue>
+
+namespace bts {
+namespace utils {
+
+// Wraps a bounded single producer, single consumer to provide extra
+// dynamically allocated storage when the wrapped queue is full. The
+// assumption is that filling the queue is a very rare event for which
+// we can afford some performance penalty. Producers must periodically
+// call drain() to ensure that events bufffered while the underlying
+// queue was full are eventually put back in the underlying queue
+
+template <class T> class GrowableSPSCQueue {
+ public:
+   GrowableSPSCQueue(size_t internal_size)
+       : _queue(internal_size), _overflow_buffer() {}
+
+   // producer
+   void drain() {
+      while (_overflow_buffer.size()) {
+         auto const &t = _overflow_buffer.front();
+         if (_queue.push(t))
+            _overflow_buffer.pop();
+         else
+            return;
+      }
+   }
+
+   void push(const T &t) {
+      if (_overflow_buffer.size() || !_queue.push(t))
+         _overflow_buffer.push(t);
+   }
+
+   // consumer
+   bool pop(T &t) { return _queue.pop(t); }
+   template <class C>
+   void consume_all(C const&c) {
+      _queue.consume_all(c);
+   }
+
+ private:
+   boost::lockfree::spsc_queue<T> _queue;
+   std::queue<T> _overflow_buffer;
+};
+
+} // namespace utils
+} // namespace bts

src/web/CMakeDeps.cmake

@@ -1 +1 @@
-set(sources ../../externals/linux/include/google/protobuf/descriptor.proto;../../externals/linux/include/google/protobuf/timestamp.proto;../lib/api/apicore.proto;../lib/api/apigateway.proto;../lib/api/apioptions.proto;../lib/api/services/AdminService.proto;../lib/api/services/ApiUnitTestService.proto;../lib/api/services/ErrorReportService.proto;../lib/api/services/HelloService.proto;../lib/api/services/MarketDataRelay.proto;../lib/api/services/OrderService.proto;../lib/api/services/OutOfProcessURLListener.proto;../lib/api/services/PersistenceService.proto;../lib/api/services/PositionService.proto;../lib/api/services/ShellDiagnosticService.proto;../lib/api/services/ShellLatencyService.proto;../lib/api/services/SyntheticSecurityService.proto;../lib/api/services/UserService.proto;../lib/domain_model/Security.proto;../lib/md/protos/co2.proto;../lib/sparkproto/SparkProtos.proto;api/build.sh;api/node_modules/exit/test/fixtures/create-files.sh;api/node_modules/protobufjs/cli/node_modules/acorn-jsx/node_modules/acorn/bin/update_authors.sh;api/node_modules/watch/release.sh;api/quiet_build.sh;api/src/DynamicService.ts;api/src/ServiceContext.ts;api/src/bts_assert.ts;ui/src/Access.fs;ui/src/Accounts/ClipLimits.fs;ui/src/Accounts/ProductLimits.fs;ui/src/Accounts/SecurityLimits.fs;ui/src/Accounts/State.fs;ui/src/Accounts/Types.fs;ui/src/Accounts/View.fs;ui/src/App.fs;ui/src/Cmd.fs;ui/src/Controls/Button.fs;ui/src/Controls/DataGrid.fs;ui/src/Controls/Fields.fs;ui/src/Controls/Modal.fs;ui/src/Controls/ReactGrid.fs;ui/src/Controls/Table.fs;ui/src/Controls/nbsp.fs;ui/src/Global.fs;ui/src/Imports.fs;ui/src/Lens.fs;ui/src/Login/Login.fs;ui/src/Nav.fs;ui/src/State.fs;ui/src/Types.fs;ui/src/Users/State.fs;ui/src/Users/Types.fs;ui/src/Users/View.fs;ui/src/Util/EnumUtils.fs;ui/src/Util/JSMap.fs)
+set(sources ../../externals/linux/include/google/protobuf/descriptor.proto;../../externals/linux/include/google/protobuf/timestamp.proto;../lib/api/apicore.proto;../lib/api/apigateway.proto;../lib/api/apioptions.proto;../lib/api/services/AdminService.proto;../lib/api/services/ApiUnitTestService.proto;../lib/api/services/ErrorReportService.proto;../lib/api/services/HelloService.proto;../lib/api/services/MarketDataRelay.proto;../lib/api/services/OrderService.proto;../lib/api/services/OutOfProcessURLListener.proto;../lib/api/services/PersistenceService.proto;../lib/api/services/PositionService.proto;../lib/api/services/ShellDiagnosticService.proto;../lib/api/services/ShellLatencyService.proto;../lib/api/services/SyntheticSecurityService.proto;../lib/api/services/UserService.proto;../lib/domain_model/Security.proto;../lib/md/protos/co2.proto;../lib/sparkproto/SparkProtos.proto;api/build.sh;api/node_modules/cssstyle/scripts/run_tests.sh;api/node_modules/exit/test/fixtures/create-files.sh;api/node_modules/protobufjs/cli/node_modules/acorn-jsx/node_modules/acorn/bin/update_authors.sh;api/node_modules/watch/release.sh;api/quiet_build.sh;api/src/DynamicService.ts;api/src/ServiceContext.ts;api/src/bts_assert.ts;ui/src/Access.fs;ui/src/Accounts/ClipLimits.fs;ui/src/Accounts/ProductLimits.fs;ui/src/Accounts/SecurityLimits.fs;ui/src/Accounts/State.fs;ui/src/Accounts/Types.fs;ui/src/Accounts/View.fs;ui/src/App.fs;ui/src/Cmd.fs;ui/src/Controls/Button.fs;ui/src/Controls/DataGrid.fs;ui/src/Controls/Fields.fs;ui/src/Controls/Modal.fs;ui/src/Controls/ReactGrid.fs;ui/src/Controls/Table.fs;ui/src/Controls/nbsp.fs;ui/src/Global.fs;ui/src/Imports.fs;ui/src/Lens.fs;ui/src/Login/Login.fs;ui/src/Nav.fs;ui/src/State.fs;ui/src/Types.fs;ui/src/Users/State.fs;ui/src/Users/Types.fs;ui/src/Users/View.fs;ui/src/Util/EnumUtils.fs;ui/src/Util/JSMap.fs)