Brian tested manually

Merged in ui_permissions (pull request #82)

Ui permissions

Approved-by: Eric Harding <eric.harding@bluetradesys.com>

.vscode/launch.json

@@ -42,7 +42,18 @@
          "stopAtEntry": false,
          "externalConsole": true,
          "requireExactSource": false
-      },        
+      },     
+      {
+         "name": "Client Tests",
+         "type": "clr",
+         "request": "launch",
+         "program": "${workspaceRoot}/build/Debug/bin/net471/sparkcmd.exe",
+         "args": ["test","test_filter"],
+         "cwd": "${workspaceRoot}",
+         "stopAtEntry": false,
+         "externalConsole": true,
+         "requireExactSource": false
+      },           
 		{
             "name": "Client Attach",
             "type": "clr",

ops_notes.txt

@@ -1,3 +1,6 @@
+1.3.4
+- Changed default logging config for Windows.  If Logging config is overridden for a system it should be adjusted to override only linux.
+
 1.1.0
 - With this release we dropped wild cards from limits, so any user
   that has limits expressed in terms of wild cards will have to have

release_notes.txt

@@ -1,3 +1,15 @@
+1.3.4
+- Significantly reduced client memory use
+- Improved OrderBook performance
+- Improved workspace load time
+
+1.3.3
+- Reduced server memory usage
+
+1.3.2
+- Improved OrderBook performance
+- Fix some memory leaks when switching workspaces
+
 1.3.1
 - Improved server side logging
 - Fixed leaking of market data subscriptions on market view form closing

src/client/spark/Admin/AdminModel.cs

@@ -11,18 +11,22 @@ namespace mbts.spark {
    [Export]
    class AdminModel : IDisposable {
       readonly string _user_name;
+      public string user_name => _user_name;
+      
       readonly BehaviorSubject<IReadOnlyDictionary<uint, string>>
          _accounts_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>()),
-         _users_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>());
+         _users_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>());               
       IReadOnlyDictionary<uint, string>
          _account_map = new Dictionary<uint, string>(),
          _user_map = new Dictionary<uint, string>();
-      IDisposable _dispose;
+      readonly BehaviorSubject<AdminModelWrapper> _wrapper_subject = new BehaviorSubject<AdminModelWrapper>(null);         
+      IDisposable _dispose;      
 
       [ImportingConstructor]
       AdminModel(AdminService service, [Import("username")]string user_name) {
          _user_name = user_name;
          _dispose = service.subscribei_admin_data().Select(update => {
+            _wrapper_subject.OnNext(new AdminModelWrapper(update.all.FirstOrDefault(), _user_name));
             IEnumerable<BTSUserImmutable> active_users =
                from admin_data in update.all
                from bts_user in admin_data.Users
@@ -52,9 +56,10 @@ namespace mbts.spark {
       }
 
       public IObservable<IReadOnlyDictionary<uint, string>> on_users => _users_subject.AsObservable();
-      public IObservable<IReadOnlyDictionary<uint, string>> on_accounts => _accounts_subject.AsObservable();
+      public IObservable<IReadOnlyDictionary<uint, string>> on_accounts => _accounts_subject.AsObservable();      
       public IObservable<IReadOnlyList<UserTuple>> on_user_tuples => on_users.Select(d => d.Select(kv => new UserTuple(kv.Key, kv.Value)).ToList());
       public IObservable<IReadOnlyList<AccountTuple>> on_account_tuples => on_accounts.Select(d => d.Select(kv => new AccountTuple(kv.Key, kv.Value)).ToList());
+      public IObservable<AdminModelWrapper> on_admin_data => _wrapper_subject.AsObservable();
 
       public UserTuple? get_current_user() {
          foreach (var kv in _user_map) {

src/client/spark/Admin/AdminModelWrapper.cs

@@ -0,0 +1,65 @@
+using bts;
+using bts.api.admin;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace mbts.spark 
+{
+   class AdminModelWrapper
+   {
+      readonly AdminDataImmutable? _admin_data;
+      readonly string _user_name;
+      readonly List<BTSAccountImmutable> _allowed_accounts;
+      readonly Dictionary<string, string> _allowed_account_dict = new Dictionary<string, string>();
+      public AdminModelWrapper(AdminDataImmutable? admin_data = null, string user_name = null)
+      {
+         _admin_data = admin_data;
+         _user_name = user_name;
+         _allowed_accounts = search_admin(u => u.Name == _user_name).Select(x => x.Item1).ToList();
+         foreach(var val in _allowed_accounts)
+            _allowed_account_dict[val.Name] = val.Name;
+      }
+      public IEnumerable<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>> get_trading_permissions(UInt32 user_id)
+      {
+         return search_admin(u => u.Id == user_id);
+      }         
+      public IEnumerable<BTSAccountImmutable> viewable_accounts()
+      {
+         return _allowed_accounts;
+      }
+      public bool can_view_account(string account)
+      {
+         return _allowed_account_dict.ContainsKey(account);
+      }
+
+      IEnumerable<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>> search_admin(Func<BTSUserImmutable, bool> predicate)
+      {
+         var rv = new List<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>>();
+         if (_admin_data.HasValue)
+         {
+            var users = _admin_data.Value.Users.Where(predicate);
+            if (users != null && users.Count() == 1)
+            {
+               var user = users.ElementAt(0);
+               if (user.HasUserData && user.UserData.HasAccountPermissions)
+               {
+                  foreach (var kv in user.UserData.AccountPermissions.AccountPermissions)
+                  {
+                     var account_id = kv.Key;
+                     var accounts = _admin_data.Value.Accounts.Where(a => a.Id == account_id);
+                     if (accounts != null && accounts.Count() == 1)
+                     {
+                        var account = accounts.ElementAt(0);
+                        var tp = kv.Value.TradingPermission;
+                        var rv_val = new TradingPermissionUnitImmutable(tp);
+                        rv.Add(Tuple.Create(account, rv_val));
+                     }
+                  }
+               }
+            }
+         }
+         return rv;
+      }
+   }
+}

src/client/spark/Admin/PermissionFilters.cs

@@ -0,0 +1,66 @@
+using bts;
+using bts.api;
+using bts.api.core;
+using bts.api.orderman;
+using bts.api.position;
+using bts.spark;
+using mbts.collections;
+using mbts.spark;
+using mbts.spark.utilities;
+using mbts.utils;
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Collections.Specialized;
+using System.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+
+namespace bts.api.core
+{
+
+   static class PermissionFilters
+   {
+      public static IObservable<Update<PositionUnitImmutable, int>> PermissionFilter(this IObservable<Update<PositionUnitImmutable, int>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<PositionUnitImmutable, bool>>( observer =>
+         {
+            return am.on_admin_data.Subscribe(x =>
+            {
+               Func<PositionUnitImmutable, bool> filter = (z => x.can_view_account(z.Account));
+               observer.OnNext(filter);
+            });
+         });
+
+         return source.Filter(pred);
+      }
+
+      public static IObservable<Update<OrderRecordImmutable, uint>> PermissionFilter(this IObservable<Update<OrderRecordImmutable, uint>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<OrderRecordImmutable, bool>>(observer =>
+        {
+           return am.on_admin_data.Subscribe(x =>
+           {
+              Func<OrderRecordImmutable, bool> filter = (z => x.can_view_account(z.account()));
+              observer.OnNext(filter);
+           });
+        });
+
+         return source.Filter(pred);
+      }
+
+      public static IObservable<Update<FillImmutable, string>> PermissionFilter(this IObservable<Update<FillImmutable, string>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<FillImmutable, bool>>(observer =>
+        {
+           return am.on_admin_data.Subscribe(x =>
+           {
+              Func<FillImmutable, bool> filter = (z => x.can_view_account(z.Account));
+              observer.OnNext(filter);
+           });
+        });
+
+         return source.Filter(pred);
+      }
+   }
+}

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/OrderBookCell.cs

@@ -47,7 +47,7 @@ namespace mbts.spark {
             where T : struct =>
             new OrderBookCellStore<T?>(row, column, prop_name, get_value, select_other, format, CompareUtils.nullable_comparison(compare), EqualityComparer.CreateNullable<T>(EqualityComparer<T>.Default.Equals));
 
-         class OrderBookCellStore<T> : IOrderBookCellStore<T> {
+         class OrderBookCellStore<T> : IOrderBookCellStore<T>, INotifyPropertyChanged {
             readonly OrderBookRowVM _row;
             readonly ICVOrderBookColumn _column;
             readonly string _prop_name;
@@ -58,6 +58,10 @@ namespace mbts.spark {
             readonly IEqualityComparer<T> _eq_comparer;
             OrderBookCell<T> _current;
 
+            // Property changed handler to avoid WPF binding leak
+            // https://blog.jetbrains.com/dotnet/2014/09/04/fighting-common-wpf-memory-leaks-with-dotmemory/
+            public event PropertyChangedEventHandler PropertyChanged = (_,__) => { };
+
             public OrderBookCellStore(
               OrderBookRowVM row,
               ICVOrderBookColumn column,

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

@@ -15,7 +15,7 @@ using OrderStatus = bts.api.orderman.OrderStatus;
 
 namespace mbts.spark {
    partial class OrderBookRowVM : DisposableViewModelBase, IEquatable<OrderBookRowVM>, DIYGrid.IRowVM<OrderRecordTuple>, DIYGrid.IActionableRowVM {
-      public readonly PredicateAnimator<OrderBookRowVM, OrderRecordTuple> animator;
+      //public readonly PredicateAnimator<OrderBookRowVM, OrderRecordTuple> animator;
 
       const string _unknown_security_price_format_string = "0.#########";
       const int _unknown_security_price_num_fractional_digits = 9;
@@ -30,31 +30,66 @@ 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;
-      IOrderBookCellStore<uint> _security_key_store;
-      IOrderBookCellStore<SecurityImmutable?> _security_store;
-      IOrderBookCellStore<Venue> _exchange_store;
-      IOrderBookCellStore<string> _symbol_store;
+      IOrderBookCellStore<uint> _order_key_store_backing;
+      IOrderBookCellStore<uint> _parent_order_key_store_backing;
+      IOrderBookCellStore<uint> _security_key_store_backing;
+      IOrderBookCellStore<SecurityImmutable?> _security_store_backing;
+      IOrderBookCellStore<Venue> _exchange_store_backing;
+      IOrderBookCellStore<string> _symbol_store_backing;
       // Def
-      IOrderBookCellStore<CustomOrderDefinition.CustomOneofCase> _order_type_store;
-      IOrderBookCellStore<Side> _side_store;
-      IOrderBookCellStore<string> _user_store;
-      IOrderBookCellStore<string> _account_store;
-      IOrderBookCellStore<TimeInForce> _time_in_force_store;
-      IOrderBookCellStore<DateTime> _time_created_store;
+      IOrderBookCellStore<CustomOrderDefinition.CustomOneofCase> _order_type_store_backing;
+      IOrderBookCellStore<Side> _side_store_backing;
+      IOrderBookCellStore<string> _user_store_backing;
+      IOrderBookCellStore<string> _account_store_backing;
+      IOrderBookCellStore<TimeInForce> _time_in_force_store_backing;
+      IOrderBookCellStore<DateTime> _time_created_store_backing;
       // State
-      IOrderBookCellStore<DateTime> _last_change_store;
-      IOrderBookCellStore<OrderStatus> _status_store;
-      IOrderBookCellStore<double> _price_store;
-      IOrderBookCellStore<uint> _quantity_store;
-      IOrderBookCellStore<uint> _filled_quantity_store;
-      IOrderBookCellStore<double> _last_price_store;
-      IOrderBookCellStore<double> _avg_price_store;
-      IOrderBookCellStore<string> _last_msg_store;
-      IOrderBookCellStore<OrderSubType> _order_sub_type_store;
+      IOrderBookCellStore<DateTime> _last_change_store_backing;
+      IOrderBookCellStore<OrderStatus> _status_store_backing;
+      IOrderBookCellStore<double> _price_store_backing;
+      IOrderBookCellStore<uint> _quantity_store_backing;
+      IOrderBookCellStore<uint> _filled_quantity_store_backing;
+      IOrderBookCellStore<double> _last_price_store_backing;
+      IOrderBookCellStore<double> _avg_price_store_backing;
+      IOrderBookCellStore<string> _last_msg_store_backing;
+      IOrderBookCellStore<OrderSubType> _order_sub_type_store_backing;
+      // Core
+      IOrderBookCellStore<uint> _order_key_store => _order_key_store_backing ?? (_order_key_store_backing = CellFactory.create(this, _book_vm.columns.order_key_column, nameof(order_key), r => r._order.Key, r => r._order_key_store));
+      IOrderBookCellStore<uint> _parent_order_key_store => _parent_order_key_store_backing ?? (_parent_order_key_store_backing = CellFactory.create(this, _book_vm.columns.parent_key_column, nameof(parent_order_key), r => r._order.ParentKey, r => r._parent_order_key_store));
+      IOrderBookCellStore<uint> _security_key_store => _security_key_store_backing ?? (_security_key_store_backing = CellFactory.create(this, _book_vm.columns.security_key_column, nameof(security_key), r => r._order.security_key(), r => r._security_key_store));
+      IOrderBookCellStore<SecurityImmutable?> _security_store => _security_store_backing ?? (_security_store_backing = CellFactory.create_nullable(this, _book_vm.columns.security_column, nameof(security), r => r._security, r => r._security_store, v => v.display_name()));
+      IOrderBookCellStore<Venue> _exchange_store => _exchange_store_backing ?? (_exchange_store_backing = CellFactory.create(this, _book_vm.columns.exchange_column, nameof(exchange), r => r._security.HasValue ? _security.Value.Exchange : Venue.Unset, r => r._exchange_store));
+      IOrderBookCellStore<string> _symbol_store => _symbol_store_backing ?? (_symbol_store_backing = CellFactory.create(this, _book_vm.columns.symbol_column, nameof(symbol), r => r._symbol, r => r._symbol_store, v => !String.IsNullOrEmpty(v) ? v : "unknown symbol"));
+      // Def
+      IOrderBookCellStore<CustomOrderDefinition.CustomOneofCase> _order_type_store => _order_type_store_backing ?? (_order_type_store_backing = CellFactory.create(this, _book_vm.columns.order_type_column, nameof(order_type), r => r._order.order_type(), r => r._order_type_store));
+      IOrderBookCellStore<Side> _side_store => _side_store_backing ?? (_side_store_backing = CellFactory.create(this, _book_vm.columns.side_column, nameof(side), r => r._order.side(), r => r._side_store));
+      IOrderBookCellStore<string> _user_store => _user_store_backing ?? (_user_store_backing = CellFactory.create(this, _book_vm.columns.user_column, nameof(user), r => _order.user(), r => r._user_store));
+      IOrderBookCellStore<string> _account_store => _account_store_backing ?? (_account_store_backing = CellFactory.create(this, _book_vm.columns.account_column, nameof(account), r => r._order.account(), r => r._account_store));
+      IOrderBookCellStore<TimeInForce> _time_in_force_store => _time_in_force_store_backing ?? (_time_in_force_store_backing = CellFactory.create(this, _book_vm.columns.time_in_force_column, nameof(time_in_force), r => r._order.time_in_force(), r => r._time_in_force_store));
+      IOrderBookCellStore<DateTime> _time_created_store => _time_created_store_backing ?? (_time_created_store_backing = CellFactory.create(this, _book_vm.columns.time_created_column, nameof(time_created), r => r._order.created(), r => r._time_created_store, format_date));
+      // State
+      IOrderBookCellStore<DateTime> _last_change_store => _last_change_store_backing ?? (_last_change_store_backing = CellFactory.create(this, _book_vm.columns.last_change_column, nameof(last_change), r => _order.last_change(), r => r._last_change_store, format_date));
+      IOrderBookCellStore<OrderStatus> _status_store => _status_store_backing ?? (_status_store_backing = CellFactory.create(this, _book_vm.columns.status_column, nameof(status), r => _order.status(), r => r._status_store));
+      IOrderBookCellStore<double> _price_store => _price_store_backing ?? (_price_store_backing = CellFactory.create(this, _book_vm.columns.price_column, nameof(price), r => _order.price(), r => r._price_store, format_price));
+      IOrderBookCellStore<uint> _quantity_store => _quantity_store_backing ?? (_quantity_store_backing = CellFactory.create(this, _book_vm.columns.qty_column, nameof(quantity), r => _order.quantity(), r => r._quantity_store));
+      IOrderBookCellStore<uint> _filled_quantity_store => _filled_quantity_store_backing ?? (_filled_quantity_store_backing = CellFactory.create(this, _book_vm.columns.filled_qty_column, nameof(filled_quantity), r => _order.filled(), r => r._filled_quantity_store));
+      IOrderBookCellStore<double> _last_price_store => _last_price_store_backing ?? (_last_price_store_backing = CellFactory.create(this, _book_vm.columns.last_price_column, nameof(last_price), r => _order.last_price(), r => r._last_price_store, format_price));
+      IOrderBookCellStore<double> _avg_price_store => _avg_price_store_backing ?? (_avg_price_store_backing = CellFactory.create(this, _book_vm.columns.avg_price_column, nameof(avg_price), r => _order.avg_price(), r => r._avg_price_store, format_price));
+      IOrderBookCellStore<string> _last_msg_store => _last_msg_store_backing ?? (_last_msg_store_backing = CellFactory.create(this, _book_vm.columns.last_message_column, nameof(last_msg), r => _order.last_msg(), r => r._last_msg_store));
+      IOrderBookCellStore<OrderSubType> _order_sub_type_store => _order_sub_type_store_backing ?? (_order_sub_type_store_backing = CellFactory.create(this, _book_vm.columns.sub_type_column, nameof(order_sub_type), r => r._order.sub_type(), r => r._order_sub_type_store, format_sub_type));
+
+      static string format_date(DateTime v) => $"{v:M/d} {v:h:mm:ss.fff tt}";
+      string format_price(double v) => _security.HasValue ?
+         _security.Value.ContractSpec.format(v) :
+         Math.Round(v, _unknown_security_price_num_fractional_digits, MidpointRounding.AwayFromZero).ToString(_unknown_security_price_format_string);
+      static string format_sub_type(OrderSubType v) {
+         switch (v) {
+            case OrderSubType.HedgingLeg: return "H";
+            case OrderSubType.QuotingLeg: return "Q";
+            default: return string.Empty;
+         }
+      }
 
       public OrderBookRowVM(IBus bus, OrderBookVM source_vm, OrdersModel model, OrderRecordImmutable record, SecurityMD? s) {
          _book_vm = source_vm;
@@ -62,43 +97,7 @@ namespace mbts.spark {
          _order = record;
          _security_md = s;
          _symbol = _security_md.HasValue ? _security_md.Value.security.Symbol : null;
-         animator = new PredicateAnimator<OrderBookRowVM,OrderRecordTuple>(this);
-         // cells
-         cell_stores = new DIYGrid.IGridCellStore[] {
-            _security_store = CellFactory.create_nullable(this, _book_vm.columns.security_column, nameof(security), r => r._security, r => r._security_store, v => v.display_name()),
-            _security_key_store = CellFactory.create(this, _book_vm.columns.security_key_column, nameof(security_key), r => r._order.security_key(), r => r._security_key_store),
-            _parent_order_key_store = CellFactory.create(this, _book_vm.columns.parent_key_column, nameof(parent_order_key), r => r._order.ParentKey, r => r._parent_order_key_store),
-            _order_key_store = CellFactory.create(this, _book_vm.columns.order_key_column, nameof(order_key), r => r._order.Key, r => r._order_key_store),
-            _symbol_store = CellFactory.create(this, _book_vm.columns.symbol_column, nameof(symbol), r => r._symbol, r => r._symbol_store, v => !String.IsNullOrEmpty(v) ? v : "unknown symbol"),            
-            _time_created_store = CellFactory.create(this, _book_vm.columns.time_created_column, nameof(time_created), r => r._order.created(), r => r._time_created_store, format_date),
-            _user_store = CellFactory.create(this, _book_vm.columns.user_column, nameof(user), r => _order.user(), r => r._user_store),
-            _account_store = CellFactory.create(this, _book_vm.columns.account_column, nameof(account), r => r._order.account(), r => r._account_store),
-            _exchange_store = CellFactory.create(this, _book_vm.columns.exchange_column, nameof(exchange), r => r._security.HasValue ? _security.Value.Exchange : Venue.Unset, r => r._exchange_store),
-            _time_in_force_store = CellFactory.create(this, _book_vm.columns.time_in_force_column, nameof(time_in_force), r => r._order.time_in_force(), r => r._time_in_force_store),
-            _order_type_store = CellFactory.create(this, _book_vm.columns.order_type_column, nameof(order_type), r => r._order.order_type(), r => r._order_type_store),
-            _order_sub_type_store = CellFactory.create(this, _book_vm.columns.sub_type_column, nameof(order_sub_type), r => r._order.sub_type(), r => r._order_sub_type_store, format_sub_type),
-            _side_store = CellFactory.create(this, _book_vm.columns.side_column, nameof(side), r => r._order.side(), r => r._side_store),
-            _last_change_store = CellFactory.create(this, _book_vm.columns.last_change_column, nameof(last_change), r => _order.last_change(), r => r._last_change_store, format_date),
-            _price_store = CellFactory.create(this, _book_vm.columns.price_column, nameof(price), r => _order.price(), r => r._price_store, format_price),
-            _quantity_store = CellFactory.create(this, _book_vm.columns.qty_column, nameof(quantity), r => _order.quantity(), r => r._quantity_store),
-            _filled_quantity_store = CellFactory.create(this, _book_vm.columns.filled_qty_column, nameof(filled_quantity), r => _order.filled(), r => r._filled_quantity_store),
-            _avg_price_store = CellFactory.create(this, _book_vm.columns.avg_price_column, nameof(avg_price), r => _order.avg_price(), r => r._avg_price_store, format_price),
-            _last_price_store = CellFactory.create(this, _book_vm.columns.last_price_column, nameof(last_price), r => _order.last_price(), r => r._last_price_store, format_price),
-            _status_store = CellFactory.create(this, _book_vm.columns.status_column, nameof(status), r => _order.status(), r => r._status_store),
-            _last_msg_store = CellFactory.create(this, _book_vm.columns.last_message_column, nameof(last_msg), r => _order.last_msg(), r => r._last_msg_store),
-         };
-         //
-         string format_price(double v) => _security.HasValue ? _security.Value.ContractSpec.format(v) :
-            Math.Round(v, _unknown_security_price_num_fractional_digits, MidpointRounding.AwayFromZero).ToString(_unknown_security_price_format_string);
-         string format_date(DateTime v) => $"{v:M/d} {v:h:mm:ss.fff tt}";
-         string format_sub_type(OrderSubType v) {
-            switch (v) {
-               case OrderSubType.HedgingLeg: return "H";
-               case OrderSubType.QuotingLeg: return "Q";
-               default: return string.Empty;
-            }
-         }
-         //
+         //animator = new PredicateAnimator<OrderBookRowVM,OrderRecordTuple>(this);
          _visible = backing(nameof(is_visible), true);
          _expanded = backing(nameof(is_expanded), false);
          _depth_level = backing(nameof(depth_level), () => {
@@ -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);
@@ -217,7 +212,31 @@ namespace mbts.spark {
       }
 
       public ObservableCollection<RowAnimation> row_animations { get; } = new ObservableCollection<RowAnimation>();
-      public IReadOnlyList<DIYGrid.IGridCellStore> cell_stores { get; }
+
+      DIYGrid.IGridCellStore[] _cell_stores_backing;
+      public IReadOnlyList<DIYGrid.IGridCellStore> cell_stores => _cell_stores_backing ?? (_cell_stores_backing = new DIYGrid.IGridCellStore[] {
+         _security_store,
+         _security_key_store,
+         _parent_order_key_store,
+         _order_key_store,
+         _symbol_store,
+         _time_created_store,
+         _user_store,
+         _account_store,
+         _exchange_store,
+         _time_in_force_store,
+         _order_type_store,
+         _order_sub_type_store,
+         _side_store,
+         _last_change_store,
+         _price_store,
+         _quantity_store,
+         _filled_quantity_store,
+         _avg_price_store,
+         _last_price_store,
+         _status_store,
+         _last_msg_store
+      });
 
       public static Stack<OrderBookRowVM> lineage(OrderBookRowVM node) {
          var rv = new Stack<OrderBookRowVM>();
@@ -228,11 +247,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 +334,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

@@ -1,4 +1,5 @@
-using bts.api;
+using bts;
+using bts.api;
 using bts.api.core;
 using bts.api.orderman;
 using bts.shell;
@@ -27,10 +28,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 +37,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 += () => {
@@ -71,11 +70,12 @@ namespace mbts.spark {
 
          _dispose = Disposable.Create(
             bus.subscribe(SparkTags.SecuritiesTag).OnDispatcher().Subscribe(on_securities),
-            model.on_orders.OnDispatcher().AddRemoveOnly().Subscribe(on_orders),
+            model.on_orders.OnDispatcher().Subscribe(on_orders),
             grid_settings_manager,
-            columns);
+            columns,
+            order_securities_subject);
 
-         void on_orders(LegacyTableUpdate<OrderRecordImmutable, uint> table) {
+         void on_orders(TableUpdate<OrderRecordImmutable, uint> table) {
             // clear everything (most likely the server restarted)
             if (table.removed.Count() == _order_rows.Count && !table.added.Any()) {
                _selected_rows.Clear();
@@ -86,18 +86,14 @@ namespace mbts.spark {
                _parent_to_children.clear();
                _security_orders.clear();
                missing_security_orders.clear();
+               Log.Ui.warn("OrderBook received reset message.");
                return;
             }
-            var changed_keys = new HashSet<uint>();
             var added_securities = new List<SecurityImmutable>(1);
             var removed_securities = new List<SecurityImmutable>(1);
 
             foreach (var o in table.removed) {
-               // if state has it, then it wasn't actually removed, just changed
-               if (table.state.TryGetValue(o.Key, out OrderRecordImmutable changed_order)) {
-                  changed_keys.Add(o.Key);
-                  _order_rows_cache[o.Key].update_order(changed_order);
-               } else if (_order_rows_cache.TryPopValue(o.Key, out OrderBookRowVM row)) {
+               if (_order_rows_cache.TryPopValue(o.Key, out OrderBookRowVM row)) {
                   row.Dispose();
                   _order_rows.Remove(row);
                   if (_security_orders.removes_key(o.Def.SecurityKey, o.Key)) {
@@ -124,10 +120,10 @@ namespace mbts.spark {
                   missing_security_orders.remove(o.Key);
                }
             }
+            foreach (var ch in table.changes)
+               _order_rows_cache[ch.New.Key].update_order(ch.New);
             var orphans = new Dictionary<uint, OrderBookRowVM>();
             foreach (var o in table.added) {
-               if (changed_keys.Contains(o.Key))
-                  continue;
                SecurityMD? security_md;
                if (security_dict.TryGetValue(o.Def.SecurityKey, out SecurityMD smd)) {
                   security_md = smd;
@@ -170,7 +166,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 +202,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 +239,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 +296,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/Modules/Trading/OrderBook/OrderBookView.xaml.cs

@@ -15,11 +15,9 @@ namespace mbts.spark {
          CVRowTemplateKey = "CVRowTemplateKey";
 
       public const double RowHeight = 17;
-      OrderBookVM _vm;
 
       public OrderBookView(OrderBookVM vm) {
          InitializeComponent();
-         _vm = vm;
          DataContext = vm;
       }
    }

src/client/spark/Orders/OrdersModel.cs

@@ -32,7 +32,7 @@ namespace mbts.spark {
       IDisposable _dispose;
 
       [ImportingConstructor]
-      private OrdersModel(IBus bus, OrderFactory factory, OrderService service, PositionService position_service, SpreaderModel spreader, ErrorReporter error) {
+      private OrdersModel(IBus bus, OrderFactory factory, OrderService service, PositionService position_service, SpreaderModel spreader, ErrorReporter error, AdminModel admin_model) {
          _bus = bus;
          _factory = factory;
          _service = service;
@@ -41,15 +41,15 @@ namespace mbts.spark {
          _error = error;
          var scheduler = ThreadPoolScheduler.Instance;
          var interval = TimeSpan.FromMilliseconds(100);
-         on_orders = _service.subscribei_orders().Naggle(interval, scheduler).Table().Cache();
-         on_positions = _position_service.subscribei_position_data().Naggle(interval, scheduler).Table().Cache();
+         on_orders = _service.subscribei_orders().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
+         on_positions = _position_service.subscribei_position_data().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
          working_orders = 
             on_orders.Scan(
                ImmutableDictionary<uint,OrderRecordImmutable>.Empty, 
                (workers, table) => _update_working_orders(workers, table))
                .Replay(1).RefCount();
          on_order_tuples = OrderRecordTupleSource.create_observable(bus.subscribe(SparkTags.SecuritiesTag), on_orders);
-         on_fills = _service.subscribei_fills().Naggle(interval, scheduler).Table().Cache();
+         on_fills = _service.subscribei_fills().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
          _activity = new OrderActivitySummarizer(bus, on_orders);
          on_fill_tuples = FillTupleSource.create_observable(bus.subscribe(SparkTags.SecuritiesTag), on_orders, on_fills);
          _security_orders_demuxer = new TableUpdateDemuxer<OrderRecordImmutable, uint, uint>(o => o.Key, o => o.security_key());

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/Collections/ObservableAutoWrapper.cs

@@ -6,6 +6,7 @@ using System.Collections.Generic;
 using NUnit.Framework;
 using mbts.collections.extensions;
 using System.Collections;
+using bts.collections;
 
 namespace mbts.ui.utility
 {

src/client/spark/core/GlobalExtensions/GenericCollectionExtensions.cs

@@ -666,44 +666,6 @@ namespace mbts.collections.extensions {
       }
    }
 
-   public static class ListExtensions
-   {
-      #region RemoveWhere
-      public static void RemoveWhere<T>(this IList<T> list, Func<T, bool> where) { foreach (var r in list.Where(where).ToList()) list.Remove(r); }
-
-      public static void RemoveWhere<T>(this IList<T> list, Func<T, bool> where, Action<T> on_remove) {
-         int count = list.Count;
-         for (int i = 0; i < count; i++) {
-            var t = list[i];
-            if (where(t)) {
-               list.RemoveAt(i--);
-               --count;
-               on_remove(t);
-            }
-         }
-      }
-
-      private static Random rng = new Random();
-
-      public static void Shuffle<T>(this IList<T> list) {
-         int n = list.Count;
-         while (n > 1) {
-            n--;
-            int k = rng.Next(n + 1);
-            T value = list[k];
-            list[k] = list[n];
-            list[n] = value;
-         }
-      }
-
-      #endregion
-
-      public static void RemoveAt<T>(this IList<T> list, int index, int count) {
-         for (int i = 0; i < count; i++)
-            list.RemoveAt(index);
-      }
-   }
-
    public static class DictionaryExtensions
    {
       #region GetValueOrDefault

src/client/spark/core/Settings/ISavedWindowFactory.cs

@@ -29,6 +29,7 @@ namespace mbts.ui.settings
       }
 
       public Task<IShellWindow> restore(WindowState window) {
+         Log.Ui.info("Loading window '{0}'".fmt(window.title));
          bool changed;
          int count = 0;
          do {

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

@@ -446,9 +446,6 @@ namespace mbts.ui.shell
       public IEnumerable<string> AllErrors { get { return _errors.Values; } }
       #endregion
 
-      protected Property<T> backing<T>(string name, T default_value, bool validate_now = false, params Func<T, bool>[] validation_steps) =>
-          backing(name, default_value, validation_steps.Select(v => new PropertyValidation<T>(v)).ToLArray(), validate_now);
-
       protected Property<T> backing<T>(string name, T default_value, IEnumerable<PropertyValidation<T>> validation_steps, bool validate_now = false) {
          void run_validation(T val) {
             foreach (var pv in validation_steps) {

src/client/spark/view/Controls/DIYGrid/DIYGrid.cs

@@ -168,7 +168,7 @@ namespace mbts.ui.controls {
                o.Resources[ActiveSelectionHighlightKey] = v;
                o.Resources[InactiveSelectionHighlightKey] = new SolidColorBrush(v.Color) { Opacity = 0.2 }.frozen();
                o.Resources[SelectionStrokeKey] = new Pen(o.SelectionHighlight.Color.to_brush().frozen(), 1).frozen();
-               foreach (var col in o.Columns.OfType<ISelectionHighlightListener>())
+               foreach (var col in o.Columns.empty_if_null().OfType<ISelectionHighlightListener>())
                   col.on_selection_fill(v);
             }, (o, v) => v as SolidColorBrush ?? DefaultSelectionHighlight,
             DefaultSelectionHighlight);
@@ -634,7 +634,7 @@ namespace mbts.ui.controls {
          (o, v) => {
             if (DoubleHelper.IsNaNOrInfinity(v))
                v = 0;
-            if (!o._visible_columns.is_null_or_empty()) {
+            if (!o._visible_columns.is_null_or_empty() && o.Columns != null) {
                double max_font_size = o.Columns.Max(c => c.font_size);
                double min_row_height = o.RowPadding.Top + max_font_size + o.RowPadding.Bottom;
                if (v < min_row_height)

src/client/spark/view/Controls/DIYGrid/Settings/GridSettingsManager.cs

@@ -31,7 +31,7 @@ namespace mbts.ui {
       readonly ObservableCollection<TRow> _rows;
       readonly ICollectionView _rows_view;
       readonly ObservableCollection<ColumnBase> _columns;
-      readonly ObservableAutoWrapper<PredicateAnimator<TRow, TRowData>, TRow> _animators;
+      //readonly ObservableAutoWrapper<PredicateAnimator<TRow, TRowData>, TRow> _animators;
       readonly Property<GridSettings> _grid_settings;
       readonly DelegateProperty<FontFamily> _font_family;
       readonly DelegateProperty<FontWeight> _font_weight;
@@ -59,18 +59,18 @@ namespace mbts.ui {
          _font_size = backing(nameof(font_size), () => grid_settings.font_size, nameof(font_size));
          _font_color = backing(nameof(font_color), () => grid_settings.font_color.rgb_solid_brush().frozen(), nameof(grid_settings));
          _selection_highlight = backing(nameof(selection_highlight), () => grid_settings.selection_highlight.to_brush().frozen(), nameof(grid_settings));
-         _animators = new ObservableAutoWrapper<PredicateAnimator<TRow, TRowData>, TRow>(_rows = rows, create_animator);
+         //_animators = new ObservableAutoWrapper<PredicateAnimator<TRow, TRowData>, TRow>(_rows = rows, create_animator);
          _resources = new List<IDisposable> {
             ObservableAutoWrapper.connect<SortDescriptionCollection, SortDescription, string>(
                rows_view.SortDescriptions, live_sorting_props = ((ICollectionViewLiveShaping)rows_view).LiveSortingProperties, sd => sd.PropertyName),
             (launch_settings_editor = new DialogCommand(create_settings_view))
          };
          //
-         PredicateAnimator<TRow, TRowData> create_animator(TRow r) {
-            var pa = new PredicateAnimator<TRow, TRowData>(r);
-            pa.set_predicate_animations(_animation_defs);
-            return pa;
-         }
+         //PredicateAnimator<TRow, TRowData> create_animator(TRow r) {
+         //   var pa = new PredicateAnimator<TRow, TRowData>(r);
+         //   pa.set_predicate_animations(_animation_defs);
+         //   return pa;
+         //}
          Window create_settings_view() {
             var vm = new GridSettingsVM<TRowData, TPredicateBuilder>(owner_title, _bus, _orders_model, create_predicate_vm, create_column_predicate_vm);
             // need to refresh before launching the settings-view bc users can chanve some grid-column-settings properties directly from the grid (e.g. width, is_collapsed, etc.)
@@ -134,9 +134,9 @@ namespace mbts.ui {
                _animation_defs.add(pa.predicate_builder.create_predicate(_arg), PredicateAnimations.flat_definitions<TRow, TRowData>(pa.row_animations, pa.cell_animations, lookup_column));
             }
          }
-         foreach (var a in _animators) {
-            a.set_predicate_animations(_animation_defs);
-         }
+         //foreach (var a in _animators) {
+         //   a.set_predicate_animations(_animation_defs);
+         //}
          // ** in the unfortunate event that we botch backwards compatability, and the columns don't bear a 1:1 relationship with the column-settings
          // don't add insult to injury by crashing **
          if (grid_settings.columns != null && _columns != null) {

src/client/spark/view/Controls/SparkMenu/SparkMenu.cs

@@ -1,6 +1,7 @@
 using bts.api;
 using bts.spark;
 using System.Collections.Generic;
+using System.Collections.ObjectModel;
 using System.Linq;
 using System.Windows;
 using System.Windows.Controls;
@@ -14,9 +15,10 @@ namespace mbts.ui.controls {
          SetCurrentValue(FocusableProperty, false);
       }
 
-      
+
       public SparkMenu(IEnumerable<SparkCommand> commands, IBus bus) : this() =>
-         ItemsSource = commands.Select(c => new SparkCommandVM(c, bus)).ToList();
+         // ObservableCollection to avoid WPF Binding Leak
+         ItemsSource = new ObservableCollection<SparkCommandVM>(commands.Select(c => new SparkCommandVM(c, bus)));
 
       public bool ShowText {
          get { return (bool)GetValue(ShowTextProperty); }

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/HwndContentHost.cs

@@ -6,6 +6,7 @@ using System.Windows.Interop;
 using System.Runtime.InteropServices;
 using mbts.ui.native;
 using System.Windows;
+using System.Windows.Threading;
 
 namespace mbts.ui.shell
 {
@@ -82,7 +83,14 @@ namespace mbts.ui.shell
          return _hwnd;
       }
 
-      protected override void DestroyWindowCore(HandleRef hwnd) { }
+      protected override void DestroyWindowCore(HandleRef hwnd) {
+         // Improve GC performance by keeping windows out of the Finalizer queue
+         if (hwnd.Wrapper is WPFPanel p) {
+            p.BeginInvoke((Action)(() => {
+               p.Dispose();
+            }));
+         }
+      }
 
       [DllImport("user32.dll", SetLastError=true)]
       static extern IntPtr SetParent(HandleRef hwndChild, HandleRef hwndParent);

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/Shell/Workspaces/WorkspacesModel.cs

@@ -222,16 +222,15 @@ namespace mbts.ui.shell
             }
          }
          if (def.windows != null) {
-            foreach (var window in def.windows) {
-               Log.Ui.info("Loading window '{0}'".fmt(window.title));
-               var new_shell_window = await _window_restore.restore(window);
-               // user can close the launcher from the taskbar while a workspace is loading
-               if (_closing) {
-                  return;
-               } else if (new_shell_window != null) {
-                  await _wm.add(new_shell_window);
+            async Task restore(WindowState window) {
+               var w = await _window_restore.restore(window);
+               if (!_closing) {
+                  await _wm.add(w);
                }
             }
+            var loading = def.windows.Select(restore);
+            await Task.WhenAll(loading);
+            if (_closing) return;
          }
          sw.Stop();
          Log.Ui.info($"Time to load workspace '{name}': {sw.Elapsed}");

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

@@ -16,6 +16,7 @@ using System.Windows.Media.Imaging;
 using System.Windows.Navigation;
 using System.Windows.Shapes;
 using mbts.ui;
+using System.Collections.ObjectModel;
 
 namespace bts.view {
    public partial class SparkWindowView : UserControl {
@@ -36,7 +37,8 @@ namespace bts.view {
             _menu.ShowText = false;
             _grid.Children.Add(_menu);
             if (ContextMenu != null) {
-               ContextMenu.ItemsSource = commands.Select(c => new SparkCommandVM(c, bus));
+               // ObservableCollection to avoid WPF Binding Leak
+               ContextMenu.ItemsSource = new ObservableCollection<SparkCommandVM>(commands.Select(c => new SparkCommandVM(c, bus)));
             }
          } else {
             expand_button.Visibility = Visibility.Collapsed;

src/client/spark_api/api/Update.cs

@@ -129,6 +129,7 @@ namespace bts.api.core {
       public IEnumerable<T> removed { get; }
 
       public TKey key(T t) { return _key(t); }
+      internal Func<T, TKey> key_extractor => _key;
 
 #if DEBUG
       public override string ToString() {
@@ -347,5 +348,72 @@ namespace bts.api.core {
          return source.Select(t => new LegacyTableUpdate<T, TKey>(t));
       }
 
+      public static IObservable<Update<T, TKey>> Filter<T, TKey>(this IObservable<Update<T, TKey>> source, IObservable<Func<T, bool>> predicate) where TKey : IEquatable<TKey>
+      {
+         return Observable.Create<Update<T, TKey>>(observer =>
+         {
+            Update<T, TKey> unfiltered_cache = null, filtered_cache = null;
+            Func<T, bool> pred_cache = null;
+            Log.DataFlow.diag($"Create Filter {typeof(T).Name}");
+
+            return source.CombineLatest(predicate, (upd, pred) => (upd, pred)).Subscribe( tup =>
+            {
+               bool pred_changed = !object.ReferenceEquals(tup.pred, pred_cache) && (pred_cache != null);
+               pred_cache = tup.pred;
+               unfiltered_cache = tup.upd;
+               if (pred_changed && filtered_cache != null)
+               {
+                  //clear out existing
+                  var clear = new Update<T, TKey>(
+                     (IReadOnlyCollection<T>)Array.Empty<T>(), 
+                     Enumerable.Empty<T>(),
+                     Enumerable.Empty<Change<T>>(),
+                     filtered_cache.all, unfiltered_cache.key_extractor);
+                  observer.OnNext(clear);
+
+                  var refilter = unfiltered_cache.all.Where(tup.pred).ToList();
+                  var readd = new Update<T, TKey>(
+                     refilter, refilter, 
+                     Enumerable.Empty<Change<T>>(),
+                     Enumerable.Empty<T>(), unfiltered_cache.key_extractor);
+                  observer.OnNext(readd);
+                  filtered_cache = readd;
+               }
+               else
+               {
+                  IReadOnlyCollection<T> all; IEnumerable<T> added, removed; IEnumerable<Change<T>> changed;
+                  apply_predicate(tup.pred, tup.upd, out all, out added, out changed, out removed);
+                  var next = new Update<T, TKey>(all, added, changed, removed, tup.upd.key_extractor);
+                  observer.OnNext(next);
+                  filtered_cache = next;
+               }
+            }, 
+            () => observer.OnCompleted());
+         });
+      }
+
+      private static void apply_predicate<T, TKey>(Func<T, bool> predicate, Update<T, TKey> cache, 
+                                                   out IReadOnlyCollection<T> all, out IEnumerable<T> added, 
+                                                   out IEnumerable<Change<T>> changed, out IEnumerable<T> removed) where TKey : IEquatable<TKey>
+      {                 
+         var adds = new List<T>();
+         adds.AddRange(cache.added.Where(predicate));
+         var rems = new List<T>();
+         rems.AddRange(cache.removed.Where(predicate));
+         var diffs = new List<Change<T>>();
+         foreach (var change in cache.changes)
+         {
+            if(!predicate(change.Old) && predicate(change.New))
+               adds.Add(change.New);
+            else if(predicate(change.Old) && !predicate(change.New))
+               rems.Add(change.Old);
+            else if(predicate(change.Old) && predicate(change.New))
+               diffs.Add(change);
+         }
+         all = cache.all.Where(predicate).ToList();
+         added = adds;
+         changed = diffs;
+         removed = rems;
+      }
    }
 }

src/client/spark_core/Collections/Extensions/ListExtensions.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace bts.collections {
+   public static class ListExtensions {
+      public static void RemoveWhere<T>(this IList<T> list, Func<T, bool> where) { foreach (var r in list.Where(where).ToList()) list.Remove(r); }
+
+      public static void RemoveWhere<T>(this IList<T> list, Func<T, bool> where, Action<T> on_remove) {
+         int count = list.Count;
+         for (int i = 0; i < count; i++) {
+            var t = list[i];
+            if (where(t)) {
+               list.RemoveAt(i--);
+               --count;
+               on_remove(t);
+            }
+         }
+      }
+
+      private static Random rng = new Random();
+
+      public static void Shuffle<T>(this IList<T> list) {
+         int n = list.Count;
+         while (n > 1) {
+            n--;
+            int k = rng.Next(n + 1);
+            T value = list[k];
+            list[k] = list[n];
+            list[n] = value;
+         }
+      }
+
+      public static void RemoveAt<T>(this IList<T> list, int index, int count) {
+         for (int i = 0; i < count; i++)
+            list.RemoveAt(index);
+      }
+   }
+}

src/client/spark_core/Collections/SortedTree.cs

@@ -27,7 +27,7 @@ namespace bts.collections {
 
       // Insert (return: true) or overwrite (return false) a new value
       // NOTE: you should use AddAll if you are adding multiple values
-      bool Insert(T item) {
+      public bool Insert(T item) {
          var item_id = id(item);
          if (!_index.TryGetValue(item_id, out Node itemnode)) {
             // Item is new
@@ -57,7 +57,7 @@ namespace bts.collections {
             }
          }
       }
-      void Remove(T item) {
+      public void Remove(T item) {
          var item_id = id(item);
          if (_index.TryGetValue(item_id, out Node itemnode)) {
             if (_index.TryGetValue(parent(item), out Node parentnode)) {
@@ -75,7 +75,7 @@ namespace bts.collections {
       }
 
       // Note: This method does not handle replacements correctly. 
-      void AddAll(IEnumerable<T> items) {
+      public void AddAll(IEnumerable<T> items) {
          // When adding multiple items we build the index first to avoid the children before parents problem
          foreach (var item in items) {
             //_index[id(item)] = new Node(item, _comparer);
@@ -103,6 +103,27 @@ namespace bts.collections {
             }
          }
       }
+
+      Node parentmost(T item) {
+         if (_index.TryGetValue(parent(item), out Node pnode)) {
+            return parentmost(pnode.value);
+         }
+         else if (_index.TryGetValue(id(item), out Node mynode)) {
+            return mynode;
+         }
+         return null;
+      }
+
+
+      public IEnumerable<T> EnumerateGroup(T item) {
+         var pnode = parentmost(item);
+         if (pnode != null) {
+            yield return pnode.value;
+            foreach (var child in iterate(pnode.children)) {
+               yield return child;
+            }
+         }
+      }
       public IEnumerator<T> GetEnumerator() {
          return iterate(_tree).GetEnumerator();
       }

src/client/spark_test/SortedTree/SortedTreeTests.cs

@@ -1,6 +1,5 @@
 using bts.collections;
-using mbts.collections.extensions;
-using NUnit.Framework;
+using bts.test;
 using System;
 using System.Collections;
 using System.Collections.Generic;
@@ -9,7 +8,7 @@ using System.Text;
 using System.Threading.Tasks;
 
 namespace bts.tests {
-   [TestFixture]
+   [TestGroup]
    class SortedTreeTests {
 
       class Order {
@@ -92,5 +91,30 @@ namespace bts.tests {
          }
       }
 
+      [Test]
+      public void test_enumerate_group() {
+         var comparer = Comparer<Order>.Create((l, r) => l.name.CompareTo(r.name));
+         var input = orders.ToList();
+         input.Shuffle();
+         var tree = new SortedTree<Order, int>(comparer, id, pid, input);
+         Assert.SequenceEqual(tree, orders);
+
+         var children_of_a = tree.EnumerateGroup(orders.First(o => o.name == 'A'));
+         var also_children_of_a = tree.EnumerateGroup(orders.First(o => o.name == 'C'));
+         Assert.AreEqual(3, children_of_a.Count());
+         Assert.AreEqual(3, also_children_of_a.Count());
+         Assert.SequenceEqual(children_of_a, also_children_of_a);
+
+         var children_of_d = tree.EnumerateGroup(orders.First(o => o.name == 'F'));
+         var also_children_of_d = tree.EnumerateGroup(orders.First(o => o.name == 'E'));
+         Assert.AreEqual(children_of_d.First().name, 'D');
+         Assert.AreEqual(also_children_of_d.First().name, 'D');
+         Assert.SequenceEqual(children_of_d, also_children_of_d);
+         Assert.SequenceEqual(children_of_d.Select(o => o.name), new [] { 'D', 'E', 'F' });
+
+         var children_of_g = tree.EnumerateGroup(orders.First(o => o.name == 'G'));
+         Assert.SequenceEqual(children_of_g.Select(o => o.name), new [] { 'G' });
+      }
+
    }
 }

src/client/spark_test/api/ApiUpdateTests.cs

@@ -6,6 +6,7 @@ using System.Reactive.Linq;
 using System.Threading.Tasks;
 using System.Linq;
 using System.Reactive.Concurrency;
+using System.Reactive.Subjects;
 
 namespace mbts.api.Test {
    [TestGroup]
@@ -104,6 +105,121 @@ namespace mbts.api.Test {
 
       }
 
+      void validate_filter(IObservable<RawUpdate<Item>> raw, IObservable<Update<Item, int>> filtered, Func<Item, bool> pred)
+      {
+         var total = raw.ToEnumerable().Count();
+
+         // Validate that update is right at each point
+         for (var count = 1; count < total; ++count)
+         {
+            var orig = raw.Take(count).reconstitute(id).ToEnumerable().Last();            
+            var next = filtered.Take(count).ToEnumerable().Last();
+            Assert.That(next.all.All(pred));
+            Assert.That(next.added.All(pred));
+            Assert.That(next.removed.All(pred));
+            Assert.That(next.changed.All(pred));            
+         }
+
+         // Do it again with TableUpdate
+         for (var count = 1; count < total; ++count)
+         {
+            var orig = raw.Take(count).reconstitute(id).Table().ToEnumerable().Last();
+            var next = filtered.Take(count).ToEnumerable().Last();
+            Assert.That(next.added.All(pred));
+            Assert.That(next.removed.All(pred));
+            Assert.That(next.changed.All(pred));
+         }
+      }
+
+
+      void validate_filter_sub(IObservable<RawUpdate<Item>> raw, IObservable<Update<Item, int>> filtered, BehaviorSubject<Func<Item, bool>> pred)
+      {
+         raw.reconstitute(x => x.id).CombineLatest(filtered, (r, f) => Tuple.Create(r, f)).Subscribe(tup =>
+         {
+            Assert.That(tup.Item2.all.All(pred.Value));
+            Assert.That(tup.Item2.added.All(pred.Value));
+            Assert.That(!tup.Item2.removed.Any(pred.Value)); 
+            Assert.That(tup.Item2.changed.All(pred.Value));
+
+            Dictionary<int, Item> rawd = tup.Item1.added.ToDictionary(x => x.id);
+            Dictionary<int, Item> fd = tup.Item2.added.ToDictionary(x => x.id);
+            foreach (var kv in rawd)
+            {
+               if(pred.Value(kv.Value))
+                  Assert.That(fd.ContainsKey(kv.Key));
+               else 
+                  Assert.That(!fd.ContainsKey(kv.Key));
+            }
+         });
+      }
+
+
+      [Test]
+      public void test_filter()
+      {
+         var raw = Observable.Create<RawUpdate<Item>>(observer =>
+         {
+            RawUpdate<Item> rup = new RawUpdate<Item>();
+            for (int x = 0; x < 50; ++x)
+            {
+               rup.Clear();
+               var item = _item_pool[x];
+               rup.added_items.Add(item);
+               rup.added_keys.Add(item.id);
+               observer.OnNext(rup);
+            }
+            var rand = new Random();
+            for (int r = 0; r < rand.Next(4, 30); r += rand.Next(1, 2))
+            {
+               rup.Clear();
+               rup.removed_keys.Add(_item_pool[r].id);
+               observer.OnNext(rup);
+            }
+            rup.Clear();
+            for (int c = 31; c < rand.Next(32, 50); c += rand.Next(1, 2))
+            {               
+               var ch = new Item(_item_pool[c].id, _item_pool[c].name + "_change");
+               rup.changed_items.Add(ch);
+               rup.changed_keys.Add(ch.id);               
+            }
+            observer.OnNext(rup);
+
+            observer.OnCompleted(); 
+            return Disposable.empty;
+         });
+
+         Func<Item, bool> evens = (x => x.id % 2 == 0);
+         Func<Item, bool> odds = (x => x.id % 2 == 1);
+         var predeven = Observable.Create<Func<Item, bool>>(observer =>
+         {            
+            observer.OnNext(evens);            
+            observer.OnCompleted();
+            return Disposable.empty;
+         });
+
+         var filtered = raw.reconstitute(x=> x.id).Filter(predeven);
+         validate_filter(raw, filtered, evens);
+
+         var predodd = Observable.Create<Func<Item, bool>>(observer =>
+         {
+            observer.OnNext(odds);
+            observer.OnCompleted();
+            return Disposable.empty;
+         });
+
+         filtered = raw.reconstitute(x => x.id).Filter(predodd);
+         validate_filter(raw, filtered, odds);
+
+
+
+         var sub= new BehaviorSubject<Func<Item, bool>>(evens);
+         filtered = raw.reconstitute(x => x.id).Filter(sub);
+         validate_filter_sub(raw, filtered, sub);
+
+         sub.OnNext(odds);
+         sub.OnCompleted();
+      }      
+
       [Test]
       public async Task test_add() {
          // Add 1 item 10x

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/defaults/logging.conf

@@ -8,6 +8,17 @@ Logging {
    }
 }
 
+Logging {
+    rule: "System.platform == 'windows'"
+    value {
+      console { default_level: "Error" }
+      syslog  { default_level: "Info" }
+      file  { default_level: "Info" }
+      initial_queue_size: 1024
+      max_queue_growth: 100
+    }
+}
+
 Logging
 {
    value

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/jetfix/codec/FIXEncodingStream.h

@@ -18,7 +18,7 @@ class FIXEncodingStream
 {
 public:
    FIXEncodingStream(config::JetFIX::Connection const &cfg,
-                     uint32_t initial_capacity = 100*1024*10)
+                     uint32_t initial_capacity = 1024*10)
       :
       _msg_header_template (),
       _header_template_size (),

src/lib/jetfix/codec/FIXFragment.h

@@ -30,7 +30,7 @@ public:
       _owns_buffer (false)
    {}
 
-   explicit FIXFragment(uint32_t initial_capacity=100*1024)
+   explicit FIXFragment(uint32_t initial_capacity=100)
       :
       _buffer (new FIXBuffer (initial_capacity)),
       _checksum (0),

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/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

testing/manual/admin/user_account_permissions.txt

@@ -0,0 +1,14 @@
+This test verifies that users can only see trades that are in accounts for which they're permissioned
+
+1. Start a client and the admin gui
+
+2. For your user, add permissions to see all accounts in the system in the admin gui
+
+3. Enter some orders in the client for those accounts (you may have to add limits, etc), make some orders fill, cancel some orders
+
+4. Now close the client, and in the admin UI pick another user and remove trading access to the account you just traded into.   Start the client again with the user you just de-permissioned.  You should not be able to see any of the trades you just made
+
+5. Next re-permission your current user for the accounts with trades in them and the should show up in the UI.
+
+6. Start another client with -multiple with the original user and make sure you can still see all trades.
+

tools/runsparktest.cmd

@@ -0,0 +1,4 @@
+@echo off
+pushd %~dp0\..\src\client\sparkcmd
+dotnet watch run test %1 -t
+popd