Revert "Revert "Merge branch 'ilink3_gateway' of bitbucket.org:bluetradesys/spark into ilink3_shard""

This reverts commit c1df312f88.

.vscode/launch.json

@@ -31,9 +31,10 @@
             "request": "launch",
             "program": "/home/bts/spark/build/ninja_debug/server/sparkd",
             "args": [
-                "test",
-                "-a",
-                "[ilink3]"
+               "run"
+                //"test",
+                //"-a",
+                //"[ilink3]"
             ],
             "stopAtEntry": false,
             "cwd": "/home/bts/spark/build/ninja_debug/server/",
@@ -70,7 +71,8 @@
                 "debuggerPath": "/usr/bin/gdb"
             },
             "sourceFileMap": {
-                "/home/bts/spark/": "${workspaceRoot}"
+                "/home/bts/spark/": "${workspaceRoot}",
+                "/home/bts/spark_host/": "${workspaceRoot}"
             }
         },
         {

doc/user/CMakeDeps.cmake

@@ -1 +1 @@
-set(sources /home/eric/spark/doc/user/build.sh;content/_index.md;content/bts_spark/Glossary/_index.md;content/bts_spark/Help/_index.md;content/bts_spark/Help/about_spark.png;content/bts_spark/Help/image39.png;content/bts_spark/Help/image58.png;content/bts_spark/Launcher/_index.md;content/bts_spark/Launcher/image38.png;content/bts_spark/Launcher/image68.png;content/bts_spark/Logging-In-to-the-System/_index.md;content/bts_spark/Logging-In-to-the-System/image3.png;content/bts_spark/Logging-In-to-the-System/image64.png;content/bts_spark/Market/_index.md;content/bts_spark/Market/change_securities.png;content/bts_spark/Market/image11.png;content/bts_spark/Market/image14.png;content/bts_spark/Market/image20.png;content/bts_spark/Market/image21.png;content/bts_spark/Market/image23.png;content/bts_spark/Market/image24.png;content/bts_spark/Market/image34.png;content/bts_spark/Market/image49.png;content/bts_spark/Market/image53.png;content/bts_spark/Market/image66.png;content/bts_spark/Market/image69.png;content/bts_spark/Market/market_view.png;content/bts_spark/Market/settings_icon.png;content/bts_spark/Market/time_and_sales.png;content/bts_spark/My-Windows/_index.md;content/bts_spark/My-Windows/image15.png;content/bts_spark/My-Windows/image16.png;content/bts_spark/My-Windows/image37.png;content/bts_spark/Settings/_index.md;content/bts_spark/Settings/image51.png;content/bts_spark/Settings/image9.png;content/bts_spark/Spreader/_index.md;content/bts_spark/Spreader/image10.png;content/bts_spark/Spreader/image13.png;content/bts_spark/Spreader/image18.png;content/bts_spark/Spreader/image19.png;content/bts_spark/Spreader/image28.png;content/bts_spark/Spreader/image29.png;content/bts_spark/Spreader/image31.png;content/bts_spark/Spreader/image35.png;content/bts_spark/Spreader/image40.png;content/bts_spark/Spreader/image43.png;content/bts_spark/Spreader/image54.png;content/bts_spark/Spreader/image55.png;content/bts_spark/Spreader/image56.png;content/bts_spark/Spreader/image6.png;content/bts_spark/Spreader/image7.png;content/bts_spark/Spreader/image70.png;content/bts_spark/Spreader/image72.png;content/bts_spark/Spreader/image8.png;content/bts_spark/Trading/OrderTypes/index.md;content/bts_spark/Trading/_index.md;content/bts_spark/Trading/algo_studio.png;content/bts_spark/Trading/collapse_all_icon.png;content/bts_spark/Trading/cover_algo_control.png;content/bts_spark/Trading/expand_all_icon.png;content/bts_spark/Trading/image1.png;content/bts_spark/Trading/image17.png;content/bts_spark/Trading/image2.png;content/bts_spark/Trading/image25.png;content/bts_spark/Trading/image27.png;content/bts_spark/Trading/image33.png;content/bts_spark/Trading/image42.png;content/bts_spark/Trading/image44.png;content/bts_spark/Trading/image47.png;content/bts_spark/Trading/image48.png;content/bts_spark/Trading/image5.png;content/bts_spark/Trading/image50.png;content/bts_spark/Trading/image52.png;content/bts_spark/Trading/image57.png;content/bts_spark/Trading/image59.png;content/bts_spark/Trading/image60.png;content/bts_spark/Trading/image61.png;content/bts_spark/Trading/image62.png;content/bts_spark/Trading/image73.png;content/bts_spark/Trading/image74.png;content/bts_spark/Trading/join_algo_control.png;content/bts_spark/Trading/ladder.png;content/bts_spark/Trading/ladder_badges.png;content/bts_spark/Trading/ladder_main_part.png;content/bts_spark/Trading/lean_algo_control.png;content/bts_spark/Trading/mini_orderbook.png;content/bts_spark/Trading/oco_algo_control.png;content/bts_spark/Trading/offset_algo_control.png;content/bts_spark/Trading/pnl_calculation.png;content/bts_spark/Trading/pnl_window.png;content/bts_spark/Trading/recreate_icon.png;content/bts_spark/Trading/settings_icon.png;content/bts_spark/Trading/spark_menu_order_book.png;content/bts_spark/Trading/top_algo_control.png;content/bts_spark/Trading/trigger_algo_control.png;content/bts_spark/_index.md;content/release_notes/_index.md;content/system_requirements/_index.md;static/book.js;static/favicon.ico;static/image.js;templates/index.html;templates/page.html;templates/section.html;templates/shortcodes/back_button.html)
+set(sources /home/bts/spark/doc/user/build.sh;content/_index.md;content/bts_spark/Glossary/_index.md;content/bts_spark/Help/_index.md;content/bts_spark/Help/about_spark.png;content/bts_spark/Help/image39.png;content/bts_spark/Help/image58.png;content/bts_spark/Launcher/_index.md;content/bts_spark/Launcher/image38.png;content/bts_spark/Launcher/image68.png;content/bts_spark/Logging-In-to-the-System/_index.md;content/bts_spark/Logging-In-to-the-System/image3.png;content/bts_spark/Logging-In-to-the-System/image64.png;content/bts_spark/Market/_index.md;content/bts_spark/Market/change_securities.png;content/bts_spark/Market/image11.png;content/bts_spark/Market/image14.png;content/bts_spark/Market/image20.png;content/bts_spark/Market/image21.png;content/bts_spark/Market/image23.png;content/bts_spark/Market/image24.png;content/bts_spark/Market/image34.png;content/bts_spark/Market/image49.png;content/bts_spark/Market/image53.png;content/bts_spark/Market/image66.png;content/bts_spark/Market/image69.png;content/bts_spark/Market/market_view.png;content/bts_spark/Market/settings_icon.png;content/bts_spark/Market/time_and_sales.png;content/bts_spark/My-Windows/_index.md;content/bts_spark/My-Windows/image15.png;content/bts_spark/My-Windows/image16.png;content/bts_spark/My-Windows/image37.png;content/bts_spark/Settings/_index.md;content/bts_spark/Settings/image51.png;content/bts_spark/Settings/image9.png;content/bts_spark/Spreader/_index.md;content/bts_spark/Spreader/image10.png;content/bts_spark/Spreader/image13.png;content/bts_spark/Spreader/image18.png;content/bts_spark/Spreader/image19.png;content/bts_spark/Spreader/image28.png;content/bts_spark/Spreader/image29.png;content/bts_spark/Spreader/image31.png;content/bts_spark/Spreader/image35.png;content/bts_spark/Spreader/image40.png;content/bts_spark/Spreader/image43.png;content/bts_spark/Spreader/image54.png;content/bts_spark/Spreader/image55.png;content/bts_spark/Spreader/image56.png;content/bts_spark/Spreader/image6.png;content/bts_spark/Spreader/image7.png;content/bts_spark/Spreader/image70.png;content/bts_spark/Spreader/image72.png;content/bts_spark/Spreader/image8.png;content/bts_spark/Trading/OrderTypes/index.md;content/bts_spark/Trading/_index.md;content/bts_spark/Trading/algo_studio.png;content/bts_spark/Trading/collapse_all_icon.png;content/bts_spark/Trading/cover_algo_control.png;content/bts_spark/Trading/expand_all_icon.png;content/bts_spark/Trading/image1.png;content/bts_spark/Trading/image17.png;content/bts_spark/Trading/image2.png;content/bts_spark/Trading/image25.png;content/bts_spark/Trading/image27.png;content/bts_spark/Trading/image33.png;content/bts_spark/Trading/image42.png;content/bts_spark/Trading/image44.png;content/bts_spark/Trading/image47.png;content/bts_spark/Trading/image48.png;content/bts_spark/Trading/image5.png;content/bts_spark/Trading/image50.png;content/bts_spark/Trading/image52.png;content/bts_spark/Trading/image57.png;content/bts_spark/Trading/image59.png;content/bts_spark/Trading/image60.png;content/bts_spark/Trading/image61.png;content/bts_spark/Trading/image62.png;content/bts_spark/Trading/image73.png;content/bts_spark/Trading/image74.png;content/bts_spark/Trading/join_algo_control.png;content/bts_spark/Trading/ladder.png;content/bts_spark/Trading/ladder_badges.png;content/bts_spark/Trading/ladder_main_part.png;content/bts_spark/Trading/lean_algo_control.png;content/bts_spark/Trading/mini_orderbook.png;content/bts_spark/Trading/oco_algo_control.png;content/bts_spark/Trading/offset_algo_control.png;content/bts_spark/Trading/pnl_calculation.png;content/bts_spark/Trading/pnl_window.png;content/bts_spark/Trading/recreate_icon.png;content/bts_spark/Trading/settings_icon.png;content/bts_spark/Trading/spark_menu_order_book.png;content/bts_spark/Trading/top_algo_control.png;content/bts_spark/Trading/trigger_algo_control.png;content/bts_spark/_index.md;content/release_notes/_index.md;content/system_requirements/_index.md;static/book.js;static/favicon.ico;static/image.js;templates/index.html;templates/page.html;templates/section.html;templates/shortcodes/back_button.html)

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/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MarketDataService.proto;api/services/MoneyFileService.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/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/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.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/cme_reference_data.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spreader.proto;config/protos/subprocess.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/sod.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/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/Product.cpp;md/SecurityCache.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SecurityMapper.cpp;md/SecurityOracle.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/ice/ICEUtils.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/service/MarketDataServiceServer.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/mbo_tests.cpp;md/test/md2_tests.cpp;md/test/md_test_utils.cpp;md/test/synthetic_book_merge_tests.cpp;md/vendor/OnixsUtils.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;metricsLine 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/service_helpers/AlgoServiceUtils.cpp;api/service_helpers/PositionUtilities.cpp;api/services/AdminService.proto;api/services/AlgoDefinitionService.proto;api/services/AlgoService.proto;api/services/ApiUnitTestService.proto;api/services/AuthService.proto;api/services/BasePositionService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MBOService.proto;api/services/MarginService.proto;api/services/MarketDataAuxiliaryService.proto;api/services/MarketDataRelay.proto;api/services/MarketDataService.proto;api/services/MoneyFileService.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/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/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cme_reference_data.conf;config/defaults/edge.conf;config/defaults/jetfix.conf;config/defaults/keycode.conf;config/defaults/logging.conf;config/defaults/orderman_cfg.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/position_cfg.conf;config/defaults/span.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/cme_reference_data.proto;config/protos/jetfix.proto;config/protos/keycode.proto;config/protos/logging.proto;config/protos/mercury_cfg.proto;config/protos/orderman_cfg.proto;config/protos/orders_fills_cleanup.proto;config/protos/perflog_cfg.proto;config/protos/position_cfg.proto;config/protos/quickbus_cfg.proto;config/protos/span.proto;config/protos/spreader.proto;config/protos/subprocess.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/sod.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/MDPeriodicDumper.cpp;md/Diag/ShmDumper.cpp;md/MBOProcessor.cpp;md/MBOServiceServer.cpp;md/MDClient.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/Product.cpp;md/SecurityCache.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SecurityMapper.cpp;md/SecurityOracle.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/ice/ICEUtils.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/service/MarketDataServiceServer.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/mbo_tests.cpp;md/test/md2_tests.cpp;md/test/md_test_utils.cpp;md/test/synthetic_book_merge_tests.cpp;md/vendor/OnixsUtils.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;metricsLine truncated

src/lib/config/defaults/keycode.conf

@@ -1,12 +1,12 @@
 
 
-#nusers: 1
+#nusers: 2
 #allow concurrent : true
-#expires: 2020-Mar-31 00:00:00   637212096000000000
-#generation_time: 2020-Jan-06 18:56:23.406931   5248825356261457214
+#expires: 2021-Jan-01 00:00:00   637450560000000000
+#generation_time: 2020-Apr-27 21:55:11.126045   5248922231538648354
 EncryptedKeyCode {
    value {
-      keycode: "Ixb/z6hwFQmojpm4GwNEpQmsqpYT9OTHbcKtROIqj38GwGqyVn54JMDB04V048quO+NAI78DKJXsUWgjDH0y5HQlwN6Lq48s/bdwKqR1OTBPr6xoTImtCb+l5b+8fUGlg7t3X8gfMdxohHgcUi9k9n7ZrnNvqcjdYKoP0CaTZT8zHuBQx3f4Lqrz32gq9rJ5skRjH6ZqNblNdlsF+kOAqXCkvI5p0QfyKgtCodr33qsW2cbqY5gUD9KwsnmFVa8/zuFBucsoM7dHB47yRaIeOFFu7xGayAdAHA8SwLyoxE5y+adIX1SYnAiqAu1raZpeXgRE1LNQk3c7uhUjt0coag=="
+      keycode: "jVEAueEfjx2MHzAOgJMSl8z9zwK9sUXp6iKr2+zTvgNrrZevi2I5pK1OCkHhE5IQZuRde5MyL8VpgBmZg/AnNVBo6719LT06MQBExbWKpx+2/LWF/JDv8m2sBUKdfze40vnPIGhXnLl/9mg74gD6uJgRU8uLEq3Hi5Xolqdi+kLKu9Qy4yuxy1hcYzdWPmYHvGcHGlaPH3jMN96m1DIcNYoV+RvoqlIv5zVDDWRWBMIiiwqrxlpwRJBw3R0pVpRVjUYqWw9dpfkIYevpCyvw+O4hqeGKuXCylie6wHiRPIa5Pxoxu3e5K2E1epGG6I7F56hQv+e1YPk1w4vwuEACWA=="
    }
 }
 

src/lib/config/defaults/orderman_cfg.conf

@@ -2,17 +2,28 @@ OrdermanCfg {
    rule: "DeploymentType.type == 'dev'"
    value {
       executor {
-         thread_name: "oman_orders"
-	 poll: false 
-	 hosted_actor: "MDSource"
-	 hosted_actor: "OrderManager"
-	 hosted_actor: "ExecutionGateway"
+         thread_name: "oman0"
+         poll: false
+         hosted_actor: { type: "MDSource" }
+         hosted_actor: { type: "OrderManagerShard" qualifier: "om0" }
+      }
+      executor {
+         thread_name: "oman1"
+         poll: false
+         hosted_actor: { type: "MDSource" }
+         hosted_actor: { type: "OrderManagerShard" qualifier: "catch_all" }
+      }
+      executor {
+         thread_name: "oman_gw"
+         poll: false
+         hosted_actor: { type: "Gateway" qualifier: "gw0" }
       }
       executor {
          thread_name: "oman_slow"
-	 poll: false 
-	 hosted_actor: "*"
+         poll: false
+         hosted_actor: { type : "*" }
       }
+      use_gateway: true
    }
 }
 
@@ -20,15 +31,44 @@ OrdermanCfg {
    value {
       executor {
          thread_name: "oman_orders"
-	 poll: true 
-	 hosted_actor: "MDSource"
-	 hosted_actor: "OrderManager"
-	 hosted_actor: "ExecutionGateway"
+         poll: true
+         hosted_actor: { type: "MDSource" }
+         hosted_actor: { type: "OrderManager" }
+      }
+      executor {
+         thread_name: "oman_gw"
+         poll: true
+         hosted_actor: { type: "Gateway" qualifier: "gw0" }
       }
       executor {
          thread_name: "oman_slow"
-	 poll: false 
-	 hosted_actor: "*"
+         poll: false
+         hosted_actor: { type: "*" }
+      }
+      use_gateway: true
+   }
+}
+
+OrdermanShardCfg {
+   value {
+      shard {
+         qualifier: "om0"
+         user_regex: "^.*(bts|cqg).*$"
+      }
+      shard {
+         qualifier: "catch_all"
+         user_regex: "^((?!(bts|cqg)).)*$"
+         is_unknown_order_catch_all: true
       }
    }
+}
+
+GatewayCfg {
+   value {
+      #note that multiple gws are possible but not well tested as of 4/23/2020
+      gateway {
+         qualifier: "gw0"
+         connection: "*"
+      }    
+   }
 }

src/lib/config/protos/orderman_cfg.proto

@@ -2,12 +2,35 @@ syntax='proto2';
 package bts.config;
 
 message OrdermanCfg {
+  message HostedActor {
+     required string type = 1;
+     optional string qualifier = 2;
+  }
   message Executor {
     required string thread_name = 1;
     optional bool poll = 2 [ default = false];
-    repeated string hosted_actor = 3;
+    //repeated string hosted_actor = 3;
+    repeated HostedActor hosted_actor = 4;
   }
   repeated Executor executor = 1;
   optional uint32 default_iceberg_pause_secs = 2 [default = 0];
   optional string fallback_cust_order_handling_inst = 3 [default = "Y"];
+  optional bool use_gateway = 4 [default = true];
 }
+
+message OrdermanShardCfg {
+   message OrderManagerShard {
+      required string qualifier = 1;
+      required string user_regex = 2;
+      optional bool is_unknown_order_catch_all = 3;
+   }
+   repeated OrderManagerShard shard = 1;
+}
+
+message GatewayCfg {
+   message Gateway {
+      required string qualifier = 1;
+      repeated string connection = 2;
+   }   
+   repeated Gateway gateway = 1;
+}

src/lib/orderman/actors/ActorThread.cpp

@@ -0,0 +1,81 @@
+#include "ActorThread.h"
+#include "lib/utils/thread_manager.h"
+#include "lib/utils/type_name.h"
+#include <boost/algorithm/string/case_conv.hpp>
+#include <boost/algorithm/string/predicate.hpp>
+
+namespace bts {
+namespace orderman {
+
+ActorThread::ActorThread(const config::OrdermanCfg::Executor &cfg, metrics::MetricsContainer &stats)
+   : cfg(cfg), thread(), executor(), loop_metric(), done(false) 
+{
+   std::array<double, 30> buckets;
+   auto i = 0;
+   for (auto &b : buckets)
+      b = (i += 1000);
+
+   loop_metric = std::make_shared<metrics::HistogramMetric<30>>(
+         fmt::format("spark_metrics_{}_thread_loop_ns", cfg.thread_name()),
+         buckets);
+   stats.publish(loop_metric);
+}
+
+ActorThread::~ActorThread() 
+{
+   done = true;
+   thread.join();
+}
+
+void ActorThread::start(trace::PerfLog &log) 
+{
+   if (cfg.poll())
+      thread = threading::make_thread(cfg.thread_name(), [this, &log]() 
+      {
+         for (;;)
+            executor.do_work(log);
+      });
+   else 
+   {
+      thread = threading::make_thread(cfg.thread_name(), [this, &log]() 
+      {
+         const auto inner_iter = 1000;
+         while (!done) 
+         {
+            auto start = Timer::now();
+            for (auto i = 0; i < inner_iter; i++)
+               executor.do_work(log);
+            auto stop = Timer::now();
+            loop_metric->observe(ns_diff(stop, start) / inner_iter);
+            std::this_thread::sleep_for(10ms);
+         }
+      });
+   }
+}
+
+ActorThread* ActorThread::find_thread_for_actor(const std::string& actor_name,
+                                                const std::vector<std::unique_ptr<ActorThread>>& all_executors, 
+                                                const std::string& qualifier) 
+{
+   for (auto const &ex : all_executors)
+   {
+      for (auto const &hosted : ex->cfg.hosted_actor())
+      {
+         if (hosted.type() == "*" || boost::algorithm::to_lower_copy(hosted.type()) ==
+                                     boost::algorithm::to_lower_copy(actor_name))
+         {
+            if(qualifier != "")
+            {
+               if(boost::algorithm::to_lower_copy(qualifier) == boost::algorithm::to_lower_copy(hosted.qualifier()))
+                  return ex.get();
+            }
+            else
+               return ex.get();
+         }
+      }
+   }
+   return nullptr;
+}
+
+
+}}

src/lib/orderman/actors/ActorThread.h

@@ -0,0 +1,31 @@
+#pragma once
+#include "lib/mercury/executor/Executor.h"
+#include "lib/metrics/metrics.h"
+#include "lib/utils/thread_manager.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/jetfix.pb.h"
+#include "lib/config/protos/orderman_cfg.pb.h"
+
+namespace bts {
+namespace orderman {
+
+struct ActorThread 
+{
+   ActorThread(const config::OrdermanCfg::Executor &cfg, metrics::MetricsContainer &stats);
+
+   void start(trace::PerfLog &log);
+   ~ActorThread();
+   
+   static ActorThread* find_thread_for_actor(const std::string&actor_name, 
+                                             const std::vector<std::unique_ptr<ActorThread>>& all_executors,
+                                             const std::string& qualifier = "");
+   
+
+   config::OrdermanCfg::Executor cfg;
+   std::thread thread;
+   mercury::Executor executor;
+   std::shared_ptr<metrics::HistogramMetric<30>> loop_metric;
+   std::atomic<bool> done;
+};
+
+}}

src/lib/orderman/actors/AsyncFactory.cpp

@@ -59,7 +59,8 @@ void AsyncFactory::init() {
             response.reserve(ev.n_buffers);
             while (response.size() < ev.n_buffers)
                response.push_back(std::make_shared<jetfix::OutFIX>(*ev.cfg));
-            this->publish(me->fix_buffer_stream, std::move(response), ev.cfg);
+            this->publish(me->fix_buffer_stream, std::move(response), ev.cfg, 
+                          ev.request_qualifier);
          }
    });
 }

src/lib/orderman/actors/IExecutionGateway.cpp

@@ -2,18 +2,47 @@
 #include "IExecutionGateway.h"
 #include "GLBXExecutionGateway.h"
 #include "ICEExecutionGateway.h"
+#include "lib/orderman/actors/gateway/GatewayProxy.h"
 #include "lib/utils/ValueWrapper.h"
 #include "lib/orderman/actors/GLBXiLink3ExecutionGateway.h"
-
-#include <variant>
+#include "lib/config/config.h"
+#include "lib/config/protos/orderman_cfg.pb.h"
+#include <utility>
 
 namespace bts {
 namespace orderman {
 
-std::unique_ptr<IExecutionGateway>
+std::atomic<bool> ExecutionGatewayFactory::_registration_open = true;
+std::map<std::string, std::shared_ptr<IExecutionGateway>> ExecutionGatewayFactory::_proxies;
+
+void ExecutionGatewayFactory::register_proxy(const config::JetFIX::Connection& conn, std::shared_ptr<IExecutionGateway> prox)
+{
+   bts_assert_msg(_registration_open, "attempt to register gw after construction");
+   auto it = _proxies.insert(std::make_pair(conn.name(), prox));
+   bts_assert_msg(it.second, "duplicate gw proxy");
+}
+
+std::shared_ptr<IExecutionGateway>
 ExecutionGatewayFactory::create_gateway(const config::JetFIX::Connection &conn,
                                         trace::PerfLog &pl,
-                                        md::SecurityOracle o) {
+                                        md::SecurityOracle o)
+{
+   if(config::get<config::OrdermanCfg>().use_gateway())
+   {
+      _registration_open = false;
+      auto it = _proxies.find(conn.name());
+      bts_assert_msg(it != _proxies.end(), "can't find gw proxy.");
+      return it->second;
+   }
+   else
+      return create_local_gateway(conn, pl, o);
+}
+
+std::unique_ptr<IExecutionGateway>
+ExecutionGatewayFactory::create_local_gateway(const config::JetFIX::Connection &conn,
+                                              trace::PerfLog &pl,
+                                              md::SecurityOracle o)
+{
    if (conn.platform() == bts::config::JetFIX_Connection_JetFIXPlatform_ICE ||
        conn.platform() == bts::config::JetFIX_Connection_JetFIXPlatform_SIMICE)
       return std::make_unique<ICEExecutionGateway>(conn, pl, o);
@@ -21,7 +50,6 @@ ExecutionGatewayFactory::create_gateway(const config::JetFIX::Connection &conn,
       return std::make_unique<GLBXiLink3ExecutionGateway>(conn, pl, o);
    else
       return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
-      
 }
 
 } // namespace orderman

src/lib/orderman/actors/IExecutionGateway.h

@@ -10,6 +10,8 @@
 
 #include <memory>
 #include <variant>
+#include <map>
+#include <atomic>
 
 namespace bts {
 namespace trace {struct CorID;class PerfLog;}
@@ -77,8 +79,16 @@ private:
 class ExecutionGatewayFactory
 {
 public:
-   static std::unique_ptr<IExecutionGateway> create_gateway(const config::JetFIX::Connection&, trace::PerfLog&, md::SecurityOracle);
-   
+   static std::shared_ptr<IExecutionGateway>
+      create_gateway(const config::JetFIX::Connection&, trace::PerfLog&, md::SecurityOracle);
+   static std::unique_ptr<IExecutionGateway>
+      create_local_gateway(const config::JetFIX::Connection&, trace::PerfLog&, md::SecurityOracle);   
+
+private:
+   friend class GatewayProxy;
+   static void register_proxy(const config::JetFIX::Connection&, std::shared_ptr<IExecutionGateway>);
+   static std::map<std::string, std::shared_ptr<IExecutionGateway>> _proxies;
+   static std::atomic<bool> _registration_open;
 };
 
 } // namespace orderman

src/lib/orderman/actors/gateway/Gateway.cpp

@@ -0,0 +1,160 @@
+#include "Gateway.h"
+
+#include "GatewayProxy.h"
+#include "lib/mercury/executor/Executor.h"
+#include "lib/mercury/events/Update.h"
+#include "lib/mercury/events/Stream.h"
+#include "lib/orderman/events/NewSecuritiesEvent.h"
+#include "lib/orderman/events/NewAdminDataEvent.h"
+#include "lib/orderman/events/SendFixEvent.h"
+#include "lib/orderman/events/SendOrderMsgsEvent.h"
+#include "lib/orderman/events/GatewayStartEvent.h"
+#include "lib/orderman/events/GatewayStatusChangeEvent.h"
+#include "lib/orderman/events/CancelRejectEvent.h"
+#include "lib/orderman/events/NewExecutionReportEvent.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/orderman/orderman_utils/ExecutionReport.h"
+#include "lib/jetfix/common/OutFIXPtr.h"
+#include "lib/jetfix/common/OutFIX.h"
+#include "lib/utils/bts_assert.h"
+#include "lib/utils/helper_macros.h"
+#include "lib/utils/type_name.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/mercury_cfg.pb.h"
+#include "lib/config/protos/jetfix.pb.h"
+#include <map>
+#include <atomic>
+
+namespace bts {
+namespace orderman {
+   
+void Gateway::spawn_gateways(const std::vector<std::unique_ptr<ActorThread>>& actor_threads,
+                             trace::PerfLog& pl, md::SecurityOracle oracle)
+{
+   if(!config::get<config::OrdermanCfg>().use_gateway())
+      return;
+   bts_assert_msg(config::get<config::OrdermanCfg>().use_gateway(), "invalid gw config");
+   std::map<std::string, config::JetFIX::Connection> connections;
+   for(auto& conn : config::get<config::JetFIX>().connection())
+      connections[conn.name()] = conn;
+   
+   bool spawned_one = false;
+   for(auto& e : config::get<config::OrdermanCfg>().executor())
+   {
+      for(auto& h : e.hosted_actor())
+      {
+         if(h.type() == "Gateway")
+         {
+            auto thread = ActorThread::find_thread_for_actor(h.type(), actor_threads, h.qualifier());
+            for(auto& g : config::get<config::GatewayCfg>().gateway())
+            {
+               if(g.qualifier() == h.qualifier())
+               {
+                  for(auto& cc : g.connection())
+                  {
+                     if(cc == "*")
+                     {
+                        for(auto& allconn : connections)
+                        {
+                           thread->executor.spawn<Gateway>(allconn.second, pl, oracle);
+                           thread->executor.spawn<GatewayProxy>(allconn.second, pl, oracle);
+                           spawned_one = true;
+                        }
+                     }
+                     else
+                     {
+                        auto& conn = connections[cc];
+                        thread->executor.spawn<Gateway>(conn, pl, oracle);
+                        thread->executor.spawn<GatewayProxy>(conn, pl, oracle);
+                        connections.erase(cc);
+                        spawned_one = true;
+                     }
+                  }
+               }
+            }
+         }
+      }
+   }   
+   bts_assert_msg(spawned_one, "no Gateway spawned");
+}
+   
+struct Gateway::PImpl
+{
+   PImpl(const config::JetFIX::Connection& conn, trace::PerfLog& pl, md::SecurityOracle orc)
+       : gw(ExecutionGatewayFactory::create_local_gateway(conn, pl, orc)),
+         gateway_name(GatewayName(conn.name()))
+   {
+   }   
+   std::unique_ptr<IExecutionGateway> gw;
+   GatewayName gateway_name;
+   std::atomic<bool> started;
+   
+   mercury::Stream<NewExecutionReportEvent> new_executionreport_pub;
+   mercury::Stream<GatewayStatusChangeEvent> gatewaystatus_pub;
+   mercury::Stream<CancelRejectEvent> cancelreject_pub;
+};
+
+Gateway::Gateway(const config::JetFIX::Connection& conn, trace::PerfLog& pl, md::SecurityOracle orc)
+   : me(new PImpl(conn, pl, orc))
+{
+}
+
+Gateway::~Gateway() 
+{
+}
+
+void Gateway::init()
+{
+   me->new_executionreport_pub = make_stream<NewExecutionReportEvent>(me->gateway_name);
+   me->gw->on_exe_report() = [this](const auto& er, const auto& t)
+   {
+      publish(me->new_executionreport_pub, &er, t);
+   };
+   me->gatewaystatus_pub = make_stream<GatewayStatusChangeEvent>(me->gateway_name);
+   me->gw->on_gw_status_change() = [this](bool b)
+   {
+      publish(me->gatewaystatus_pub, me->gateway_name, b);
+   };
+   me->cancelreject_pub = make_stream<CancelRejectEvent>(me->gateway_name);
+   me->gw->on_cancel_reject() = [this](const auto& k,const auto& msg)
+   {
+      publish(me->cancelreject_pub, k, msg);
+   };
+   subscribe<SendFixEvent>(me->gateway_name, [this](const mercury::Update<SendFixEvent>& update) //for GWNAME
+   {
+      if(update.events)
+         for (const auto& ev : *update.events)
+            me->gw->send_fix(ev.fix, ev.corid);
+   });
+   subscribe<SendOrderMsgsEvent>(me->gateway_name, [this](const mercury::Update<SendOrderMsgsEvent>& update)
+   {
+      if(update.events)
+         for (const auto& ev : *update.events)
+            me->gw->send_msgs(*ev.msgs, ev.corid);
+   });
+
+   subscribe<NewAdminDataEvent>([this](const auto& admin_update) 
+   {
+      me->gw->on_admin_data(admin_update); 
+   });
+   subscribe<NewSecuritiesEvent>([this](const auto& secs) 
+   {
+      me->gw->on_securities(secs); 
+   });   
+   subscribe<GatewayStartEvent>(me->gateway_name, [this](const auto& ev) 
+   {
+      if (ev.events && !me->started)  //this is kind of a hack because messages go through once at startup with a null events
+      {
+         me->started = true;
+         me->gw->start();
+      }
+   });
+}
+
+void Gateway::on_poll(trace::PerfLog&)
+{
+   me->gw->dispatch_incoming_fix_messages();
+}
+
+
+}}

src/lib/orderman/actors/gateway/Gateway.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "lib/orderman/actors/IExecutionGateway.h"
+#include "lib/jetfix/common/OutFIXPtr.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/utils/bts_no_copy.h"
+#include "lib/mercury/actors/Actor.h"
+#include "lib/mercury/actors/IPollableActor.h"
+#include "lib/tracing/PerfLog.h"
+#include "lib/tracing/bts_trace.h"
+#include "lib/utils/TimeStamp.h"
+#include "lib/orderman/actors/ActorThread.h"
+#include <memory>
+
+namespace bts {
+namespace trace {struct CorID;class PerfLog;}
+namespace orderman {
+
+class Gateway : public mercury::Actor, public mercury::IPollableActor 
+{
+public:
+   static void spawn_gateways(const std::vector<std::unique_ptr<ActorThread>>& actor_threads,
+                              trace::PerfLog&, md::SecurityOracle);
+ public:   
+   Gateway(const config::JetFIX::Connection&, trace::PerfLog&, md::SecurityOracle);
+   virtual ~Gateway();
+
+   void init() override;
+   void on_poll(trace::PerfLog&) override;   
+   
+private:
+   BTS_NO_COPY(Gateway);   
+   struct PImpl;
+   std::unique_ptr<PImpl> me;      
+};
+
+}}

src/lib/orderman/actors/gateway/GatewayProxy.cpp

@@ -0,0 +1,114 @@
+#include "GatewayProxy.h"
+
+#include "lib/mercury/broker/BrokerProxy.h"
+#include "lib/mercury/events/Update.h"
+#include "lib/mercury/events/Stream.h"
+#include "lib/orderman/events/NewSecuritiesEvent.h"
+#include "lib/orderman/events/NewAdminDataEvent.h"
+#include "lib/orderman/events/SendFixEvent.h"
+#include "lib/orderman/events/SendOrderMsgsEvent.h"
+#include "lib/orderman/events/GatewayStatusChangeEvent.h"
+#include "lib/orderman/events/GatewayStartEvent.h"
+#include "lib/orderman/events/CancelRejectEvent.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/jetfix/common/OutFIXPtr.h"
+#include "lib/jetfix/common/OutFIX.h"
+#include "lib/utils/helper_macros.h"
+#include "lib/utils/bts_assert.h"
+#include "lib/config/config.h"
+
+#include <variant>
+#include <tuple>
+
+namespace bts {
+namespace orderman {
+   
+struct GatewayProxy::PImpl
+{
+   PImpl(const config::JetFIX::Connection &cfg, trace::PerfLog &l, md::SecurityOracle o)
+       : cfg(cfg), perf_log(l), sec_oracle(o), gateway_name(cfg.name())
+   {
+   }   
+   const config::JetFIX::Connection& cfg;
+   trace::PerfLog& perf_log;
+   md::SecurityOracle sec_oracle;
+   GatewayName gateway_name;
+
+   mercury::Stream<GatewayStartEvent> start_pub;
+   mercury::Stream<SendFixEvent> send_fix_pub;
+   mercury::Stream<SendOrderMsgsEvent> send_ordermsgs_pub;
+      
+   OnExeReport on_exe_report;
+   OnGwStatusChange on_gw_status_change;
+   OnCancelReject on_cancel_reject;
+};
+
+
+GatewayProxy::GatewayProxy(const config::JetFIX::Connection& conn, trace::PerfLog& pl, md::SecurityOracle md)
+   : me(new PImpl(conn, pl, md))
+{
+   ExecutionGatewayFactory::register_proxy(conn, std::shared_ptr<GatewayProxy>(this));
+}
+
+GatewayProxy::~GatewayProxy() { }
+
+//actor init
+void GatewayProxy::init() 
+{
+   me->start_pub = make_stream<GatewayStartEvent>(me->gateway_name);
+   me->send_fix_pub = make_stream<SendFixEvent>(me->gateway_name);
+   me->send_ordermsgs_pub = make_stream<SendOrderMsgsEvent>(me->gateway_name);   
+}
+
+//start the gateway
+void GatewayProxy::start()
+{
+   publish<GatewayStartEvent>(me->start_pub, me->gateway_name);
+}
+ 
+void GatewayProxy::dispatch_incoming_fix_messages()
+{
+   //called on GW thread and sent back through queue
+}
+   
+void GatewayProxy::on_securities(const mercury::Update<NewSecuritiesEvent>& )
+{
+   //the actual gateway subscribes to these, no-op here
+}
+ 
+void GatewayProxy::on_admin_data(const mercury::Update<NewAdminDataEvent>& )
+{
+   //the actual gateway subscribes to these, no-op here
+}
+   
+void GatewayProxy::send_fix(const jetfix::OutFIXPtr& fixptr, trace::CorID corid)
+{
+   //jetfix::OutFIXPtr copy(new jetfix::OutFIX(*fixptr));  //copy not necessary see OrderManager::reload_fix_buffer
+   publish<SendFixEvent>(me->send_fix_pub, fixptr, corid);
+}
+   
+void GatewayProxy::send_msgs(const std::vector<OrderMsg>& msgs, trace::CorID corid)
+{
+   SendOrderMsgsEvent::MsgsPtr copy(new SendOrderMsgsEvent::Msgs(msgs));
+   publish<SendOrderMsgsEvent>(me->send_ordermsgs_pub, copy, corid);
+}
+   
+OnExeReport& GatewayProxy::on_exe_report() 
+{
+   bts_assert_msg(false, "GatewayProxy");
+   return me->on_exe_report;
+}
+   
+OnGwStatusChange& GatewayProxy::on_gw_status_change()
+{
+   bts_assert_msg(false, "GatewayProxy");
+   return me->on_gw_status_change;
+}
+   
+OnCancelReject& GatewayProxy::on_cancel_reject()
+{
+   bts_assert_msg(false, "GatewayProxy");
+   return me->on_cancel_reject;
+}
+
+}}

src/lib/orderman/actors/gateway/GatewayProxy.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include "lib/orderman/actors/IExecutionGateway.h"
+#include "lib/mercury/actors/Actor.h"
+#include "lib/jetfix/common/OutFIXPtr.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/jetfix.pb.h"
+#include "lib/utils/bts_no_copy.h"
+
+#include <memory>
+
+namespace bts {
+namespace trace {struct CorID;class PerfLog;}
+namespace orderman {
+     
+class GatewayProxy : public mercury::Actor, public IExecutionGateway
+{
+ public:   
+   GatewayProxy(const config::JetFIX::Connection&, trace::PerfLog&, md::SecurityOracle);
+   virtual ~GatewayProxy();
+
+   //Actor
+   virtual void init();
+
+   //IExecutionGateway
+   virtual void start();
+   virtual void dispatch_incoming_fix_messages();
+   virtual void on_securities(const mercury::Update<NewSecuritiesEvent> &);
+   virtual void on_admin_data(const mercury::Update<NewAdminDataEvent> &);
+   virtual void send_fix(const jetfix::OutFIXPtr&, trace::CorID);
+   virtual void send_msgs(std::vector<OrderMsg> const&, trace::CorID);
+   
+   virtual OnExeReport& on_exe_report();
+   virtual OnGwStatusChange& on_gw_status_change();
+   virtual OnCancelReject &on_cancel_reject();
+private:
+   BTS_NO_COPY(GatewayProxy);   
+   
+   struct PImpl;
+   std::unique_ptr<PImpl> me;     
+};
+
+}}

src/lib/orderman/actors/order_manager/OrderManager.cpp

@@ -16,6 +16,8 @@
 #include "lib/orderman/securities/ISecurityDB.h"
 #include "lib/utils/inplace_function.h"
 #include "lib/utils/thread_manager.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/orderman_cfg.pb.h"
 
 #include <algorithm>
 #include <boost/algorithm/string/trim.hpp>
@@ -36,8 +38,11 @@ enum class OrderManager::State {
 
 struct OrderManager::PImpl {
    PImpl(OrderManager &owner, DynamicOrderRegistry const &b, trace::PerfLog &l,
-         md::SecurityOracle o)
-       : state(State::WAIT_ADMIN), services_data(owner, b, l, o) {}
+         md::SecurityOracle o,
+         const std::string& qualifier)
+       : state(State::WAIT_ADMIN), services_data(owner, b, l, o),
+         qualifier(qualifier) 
+       {}
 
    State state;
    std::unordered_map<OrderKey, Order *> orders;
@@ -47,11 +52,12 @@ struct OrderManager::PImpl {
    mercury::Stream<OrderManagerReadyEvent> pub_ready;
    std::thread background_thread;
    std::atomic<bool> shutdown = false;
+   std::string qualifier;
 };
 
 OrderManager::OrderManager(DynamicOrderRegistry const &b, trace::PerfLog &l,
-                           md::SecurityOracle o)
-    : me(new PImpl(*this, b, l, o)) {
+                           md::SecurityOracle o, const std::string& q)
+    : me(new PImpl(*this, b, l, o, q)) {
    auto id_gen = &me->services_data.id_gen;
    auto done = &me->shutdown;
    me->background_thread =
@@ -369,6 +375,8 @@ void OrderManager::on_new_fix_buffer(
     const mercury::Update<CreateFIXBufferReplyEvent> &m) {
    if (m.events)
       for (auto const &ev : *m.events) {
+         if(ev.request_qualifier != me->qualifier)
+            continue;
          auto &fb = free_buffers(ev.cfg->name());
          fb.insert(fb.end(), ev.buffers.begin(), ev.buffers.end());
       }
@@ -382,7 +390,8 @@ void OrderManager::on_timer(const mercury::Update<mercury::TimerEvent> &m) {
    for (auto &osd : me->services_data.sessions) {
       if (fix_buffer_pool_size > osd->free_buffers.size()) {
          auto need = fix_buffer_pool_size - osd->free_buffers.size();
-         publish(me->services_data.create_fix_req_pub, need, osd->cfg);
+         publish(me->services_data.create_fix_req_pub, need, osd->cfg, 
+                 me->qualifier);
       }
    }
 }
@@ -457,20 +466,23 @@ void OrderManager::init_order_services() {
 
       // publishers for execution reports, gateway status and cancel
       // reject evets
-      auto s_ptr = s.get();
-      s->exe_report_pub = make_stream<NewExecutionReportEvent>(gw_name);
-      s->gw->on_exe_report() = [s_ptr, this](auto const &er, auto const &t) {
-         publish(s_ptr->exe_report_pub, &er, t);
-      };
-      s->gw_status_change_pub = make_stream<GatewayStatusChangeEvent>(gw_name);
-      s->gw->on_gw_status_change() = [s_ptr, this, gw_name](bool b) {
-         publish(s_ptr->gw_status_change_pub, gw_name, b);
-      };
-      s->cancel_reject_pub = make_stream<CancelRejectEvent>(gw_name);
-      s->gw->on_cancel_reject() = [s_ptr, this](auto const &k,
-                                                auto const &msg) {
-         publish(s_ptr->cancel_reject_pub, k, msg);
-      };
+      if(!config::get<config::OrdermanCfg>().use_gateway()) //if we are in gateway mode, the gw will publish these events
+      {
+         auto s_ptr = s.get();
+         s->exe_report_pub = make_stream<NewExecutionReportEvent>(gw_name);
+         s->gw->on_exe_report() = [s_ptr, this](auto const &er, auto const &t) {
+            publish(s_ptr->exe_report_pub, &er, t);
+         };
+         s->gw_status_change_pub = make_stream<GatewayStatusChangeEvent>(gw_name);
+         s->gw->on_gw_status_change() = [s_ptr, this, gw_name](bool b) {
+            publish(s_ptr->gw_status_change_pub, gw_name, b);
+         };
+         s->cancel_reject_pub = make_stream<CancelRejectEvent>(gw_name);
+         s->gw->on_cancel_reject() = [s_ptr, this](auto const &k,
+                                                   auto const &msg) {
+            publish(s_ptr->cancel_reject_pub, k, msg);
+         };
+      }
       // handlers are ready so we can start processing fix events
       s->gw->start();
 

src/lib/orderman/actors/order_manager/OrderManager.h

@@ -64,23 +64,24 @@ namespace orderman {
 
 class OrderManager : public mercury::Actor, public mercury::IPollableActor {
  public:
-   OrderManager(DynamicOrderRegistry const &dyn_order_builder, trace::PerfLog &, md::SecurityOracle);
+   OrderManager(DynamicOrderRegistry const &dyn_order_builder, trace::PerfLog &, md::SecurityOracle,
+                const std::string& qualifier);
   
    ~OrderManager();
    void init() override;
 
- private:
+ protected:
    friend class OrderServices;
 
    // external events
    void on_securities(const mercury::Update<NewSecuritiesEvent> &);
    void on_filled_position(const mercury::Update<FilledPositionEvent> &);
-   void on_order_recovery(const mercury::Update<StoredOrderEvent> &);
-   void on_execution_report(const mercury::Update<NewExecutionReportEvent> &);
-   void on_activate_order(const mercury::Update<ActivateOrderEvent> &);
-   void on_update_order(const mercury::Update<UpdateOrderEvent> &);
-   void on_cancel_order(const mercury::Update<CancelOrderEvent> &);
-   void on_detach_order(const mercury::Update<DetachOrderEvent> &);
+   virtual void on_order_recovery(const mercury::Update<StoredOrderEvent> &);
+   virtual void on_execution_report(const mercury::Update<NewExecutionReportEvent> &);
+   virtual void on_activate_order(const mercury::Update<ActivateOrderEvent> &);
+   virtual void on_update_order(const mercury::Update<UpdateOrderEvent> &);
+   virtual void on_cancel_order(const mercury::Update<CancelOrderEvent> &);
+   virtual void on_detach_order(const mercury::Update<DetachOrderEvent> &);
    void on_new_fix_buffer(const mercury::Update<CreateFIXBufferReplyEvent> &);
    void on_gateway_state(const mercury::Update<GatewayStatusChangeEvent> &);
    void on_timer(const mercury::Update<mercury::TimerEvent> &);

src/lib/orderman/actors/order_manager/OrderManagerShard.cpp

@@ -0,0 +1,201 @@
+#include "OrderManagerShard.h"
+
+#include "lib/api/services/OrderService.pb.h"
+#include "lib/md/MDConsumer.h"
+#include "lib/orderman/actors/MDSource.h"
+#include "lib/orderman/orders/Order.h"
+#include "lib/orderman/actors/order_manager/OrderManager.h"
+#include "lib/orderman/order_factory/DynamicOrderRegistry.h"
+#include "lib/mercury/executor/Executor.h"
+#include "lib/mercury/actors/ActorBroker.h"
+#include "lib/mercury/events/Update.h"
+#include "lib/mercury/events/Stream.h"
+#include "lib/orderman/events/AccountPnLUpdateEvent.h"
+#include "lib/orderman/events/FilledPositionEvent.h"
+#include "lib/orderman/events/ExcessLiquidityUpdateEvent.h"
+#include "lib/orderman/events/NewAdminDataEvent.h"
+#include "lib/orderman/events/NewSecuritiesEvent.h"
+#include "lib/orderman/events/OrderManagerReadyEvent.h"
+#include "lib/orderman/events/StoredOrderEvent.h"
+#include "lib/orderman/events/NewExecutionReportEvent.h"
+#include "lib/orderman/events/OrderDefinitionUpdateEvent.h"
+#include "lib/mercury/events/TimerEvent.h"
+#include "lib/utils/bts_assert.h"
+#include "lib/utils/helper_macros.h"
+#include "lib/utils/type_name.h"
+#include "lib/metrics/metrics.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/mercury_cfg.pb.h"
+#include "lib/config/protos/jetfix.pb.h"
+
+#include <regex>
+#include <tuple>
+#include <map>
+
+namespace bts {
+namespace orderman {
+   
+//TODO this creates order manager shards, for now we accept that if they're paired with MDSources, we'll 
+//receive duplicate updates for the MD events.  With 2-3 OM/MD threads this is probably not a huge deal.  
+bool OrderManagerShard::spawn_shards(const std::vector<std::unique_ptr<ActorThread>>& actor_threads,
+                                     const DynamicOrderRegistry& dynorders, trace::PerfLog& pl, md::SecurityOracle md)
+{
+   //OrderManagerShard requires gateway use
+   if(!config::get<config::OrdermanCfg>().use_gateway())
+      return false;
+   bts_assert_msg(config::get<config::OrdermanCfg>().use_gateway(), "invalid OrderManagerShard config");      
+   int catch_all = 0, spawned = 0;
+   for(auto& e : config::get<config::OrdermanCfg>().executor())
+   {
+      for(auto& h : e.hosted_actor())
+      {
+         if(h.type() == "OrderManagerShard")
+         {
+            auto thread = ActorThread::find_thread_for_actor(h.type(), actor_threads, h.qualifier());
+            for(auto& cfg : config::get<config::OrdermanShardCfg>().shard())
+            {               
+               if(cfg.qualifier() == h.qualifier())
+               {                  
+                  thread->executor.spawn<OrderManagerShard>(cfg, dynorders, pl, md);
+                  ++spawned;
+                  if(cfg.is_unknown_order_catch_all())
+                     catch_all++;
+               }
+            }
+         }
+      }
+   }
+   if(spawned)
+      bts_assert_msg(catch_all == 1, "exactly one OrderManagerShard must be used as an unknown order catch all");
+   return spawned > 0;
+}
+     
+struct OrderManagerShard::PImpl
+{
+   PImpl(const config::OrdermanShardCfg::OrderManagerShard& cfg,
+         const DynamicOrderRegistry&, trace::PerfLog&, md::SecurityOracle) 
+      : cfg(cfg), rx(cfg.user_regex()), all_orders()
+   {
+   }
+   config::OrdermanShardCfg::OrderManagerShard cfg;
+   std::regex rx;
+   std::unordered_map<OrderKey, api::orderman::OrderDefinition> all_orders;
+};
+
+OrderManagerShard::OrderManagerShard(const config::OrdermanShardCfg::OrderManagerShard& cfg,
+                                     const DynamicOrderRegistry& dyn, trace::PerfLog& pl, md::SecurityOracle orc)
+   : OrderManager(dyn, pl, orc, cfg.user_regex()), me(new PImpl(cfg, dyn, pl, orc))
+{
+}
+
+OrderManagerShard::~OrderManagerShard() 
+{
+}
+
+template<typename EV>
+void filter_and_dispatch(const mercury::Update<EV>& update, 
+                         std::function<bool(const EV&)> selector,
+                         std::function<void(const mercury::Update<EV>&)> dispatcher)
+{
+   if(update.events)
+   {
+      typename EV::Array filtered; 
+      int count = 0;
+      for(auto& ev : *update.events)
+      {
+         if(selector(ev))
+         {
+            filtered.emplace_back(ev);
+            ++count;
+         }
+      }
+      //have to send through 0 length for on_order_recovery
+      mercury::Update<EV> fu;
+      fu.events = &filtered;
+      dispatcher(fu);
+   }
+}
+
+void OrderManagerShard::init()
+{
+   //this should give us all orders on all OMs
+   subscribe<OrderDefinitionUpdateEvent>([this](const auto& update)
+   {
+      if(update.events)
+      {
+         me->all_orders.reserve(update.events->size());
+         for (const auto &ev : *(update.events))
+            me->all_orders.emplace(ev.key, *ev.def);
+      }
+   });
+   OrderManager::init();
+}
+
+
+void OrderManagerShard::on_execution_report(const mercury::Update<NewExecutionReportEvent>& upd)
+{
+   UNUSED(upd);
+   if(me->cfg.is_unknown_order_catch_all() && upd.events)
+   {
+      for (const auto &ev : *(upd.events)) 
+         if(me->all_orders.find(ev.key()) == me->all_orders.end()) //if we don't know the order at all, try to recover it
+            OrderManager::on_execution_report(upd);
+   }
+}
+
+void OrderManagerShard::on_order_recovery(const mercury::Update<StoredOrderEvent>& update)
+{
+   filter_and_dispatch<StoredOrderEvent>(update, 
+      [this](const auto& ev)
+      {
+         bool match = std::regex_match(ev.order_record.def().user(), me->rx);
+         return match;
+      }, 
+      [this](const auto& fu) 
+      { 
+         OrderManager::on_order_recovery(fu); 
+      }
+   );
+}
+
+void OrderManagerShard::on_activate_order(const mercury::Update<ActivateOrderEvent>& update)
+{
+   filter_and_dispatch<ActivateOrderEvent>(update, 
+      [this](const auto& ev)
+      {
+         bool match = std::regex_match(ev.order->def()->user(), me->rx);
+         return match;
+      }, 
+      [this](const auto& fu) 
+      { 
+         OrderManager::on_activate_order(fu); 
+      }
+   );   
+}
+   
+void OrderManagerShard::on_update_order(const mercury::Update<UpdateOrderEvent>& update)
+{
+   //TODO this is just to avoid error message in logs, might be better upstream
+   filter_and_dispatch<UpdateOrderEvent>(update, 
+      [this](const auto& ev) { return lookup_order(ev.key()) != nullptr; }, 
+      [this](const auto& fu) { OrderManager::on_update_order(fu); }
+   );     
+}
+   
+void OrderManagerShard::on_cancel_order(const mercury::Update<CancelOrderEvent>& update)
+{
+   filter_and_dispatch<CancelOrderEvent>(update, 
+      [this](const auto& ev) { return lookup_order(ev.key()) != nullptr; }, 
+      [this](const auto& fu) { OrderManager::on_cancel_order(fu); }
+   );     
+}
+   
+void OrderManagerShard::on_detach_order(const mercury::Update<DetachOrderEvent>& update)
+{
+   filter_and_dispatch<DetachOrderEvent>(update, 
+      [this](const auto& ev) { return lookup_order(ev.key()) != nullptr; }, 
+      [this](const auto& fu) { OrderManager::on_detach_order(fu); }
+   );       
+}
+
+}}

src/lib/orderman/actors/order_manager/OrderManagerShard.h

@@ -0,0 +1,48 @@
+#pragma once
+
+#include "lib/orderman/actors/order_manager/OrderManager.h"
+#include "lib/md/MDConsumer.h"
+#include "lib/md/SecurityOracle.h"
+#include "lib/utils/bts_no_copy.h"
+#include "lib/mercury/actors/Actor.h"
+#include "lib/mercury/actors/IPollableActor.h"
+#include "lib/tracing/PerfLog.h"
+#include "lib/tracing/bts_trace.h"
+#include "lib/utils/TimeStamp.h"
+#include "lib/orderman/actors/ActorThread.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/jetfix.pb.h"
+#include <memory>
+
+namespace bts {
+namespace trace {struct CorID;class PerfLog;}
+namespace orderman {
+
+class DynamicOrderRegistry;
+
+class OrderManagerShard : public OrderManager
+{
+public:
+   static bool spawn_shards(const std::vector<std::unique_ptr<ActorThread>>& actor_threads,
+                            const DynamicOrderRegistry&, trace::PerfLog&, md::SecurityOracle);
+ public:   
+   OrderManagerShard(const config::OrdermanShardCfg::OrderManagerShard&,
+                     const DynamicOrderRegistry&, trace::PerfLog&, md::SecurityOracle);
+   virtual ~OrderManagerShard();
+   void init() override;
+
+private:
+   void on_order_recovery(const mercury::Update<StoredOrderEvent> &) override;
+   void on_execution_report(const mercury::Update<NewExecutionReportEvent> &) override;
+   void on_activate_order(const mercury::Update<ActivateOrderEvent> &) override;
+   void on_update_order(const mercury::Update<UpdateOrderEvent> &) override;
+   void on_cancel_order(const mercury::Update<CancelOrderEvent> &) override;
+   void on_detach_order(const mercury::Update<DetachOrderEvent> &) override;
+   
+private:
+   BTS_NO_COPY(OrderManagerShard);   
+   struct PImpl;
+   std::unique_ptr<PImpl> me;      
+};
+
+}}

src/lib/orderman/actors/order_manager/OrderServicesData.h

@@ -78,7 +78,7 @@ struct SessionServicesData {
    std::vector<OrderMsg> buffered_order_msgs;
    
    trace::CorID current_trace;
-   std::unique_ptr<IExecutionGateway> gw;
+   std::shared_ptr<IExecutionGateway> gw;
    mercury::Stream<NewExecutionReportEvent> exe_report_pub;
    mercury::Stream<GatewayStatusChangeEvent> gw_status_change_pub;
    mercury::Stream<CancelRejectEvent> cancel_reject_pub;

src/lib/orderman/events/CreateFIXBufferReplyEvent.h

@@ -13,13 +13,14 @@ class CreateFIXBufferReplyEvent
     : public mercury::Event<CreateFIXBufferReplyEvent> {
  public:
    using SynopsisPolicy = mercury::NoSynopsis;
-
    CreateFIXBufferReplyEvent(std::vector<jetfix::OutFIXPtr> &&b,
-                             config::JetFIX::Connection const *cfg)
-       : buffers(std::move(b)), cfg(cfg) {}
+                             config::JetFIX::Connection const *cfg,
+                             const std::string& request_qual)
+       : buffers(std::move(b)), cfg(cfg), request_qualifier(request_qual) {}
 
    std::vector<jetfix::OutFIXPtr> buffers;
    config::JetFIX::Connection const *cfg;
+   std::string request_qualifier;
 };
 }
 }

src/lib/orderman/events/CreateFIXBufferRequestEvent.h

@@ -15,13 +15,16 @@ public:
    using SynopsisPolicy = mercury::NoSynopsis;
    
    CreateFIXBufferRequestEvent  (size_t n_buffers,
-                                 config::JetFIX::Connection const &cfg) :
+                                 config::JetFIX::Connection const &cfg,
+                                 const std::string& request_qual) :
       n_buffers(n_buffers),
-      cfg(&cfg)
+      cfg(&cfg),
+      request_qualifier(request_qual)
    {}
 
    size_t n_buffers; 
    config::JetFIX::Connection const *cfg;
+   std::string request_qualifier;
 };
 
 }}

src/lib/orderman/events/GatewayStartEvent.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <memory>
+#include "lib/mercury/events/Event.h"
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/mercury/synopsis/CacheLast.h"
+
+namespace bts {
+namespace orderman {
+
+class GatewayStartEvent : public mercury::Event<GatewayStartEvent> 
+{
+public:
+   using Tag = GatewayName;
+   using SynopsisPolicy = mercury::CacheLast;
+
+   GatewayStartEvent(const GatewayName &n) : gw_name(n) {}
+   GatewayName gw_name;
+};
+
+}}

src/lib/orderman/events/SendFixEvent.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include "lib/api/services/AdminService.pb.h"
+
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/mercury/common/mercury_common.h"
+#include "lib/mercury/events/Event.h"
+#include "lib/mercury/synopsis/NoSynopsis.h"
+#include "lib/jetfix/common/OutFIXPtr.h"
+#include "lib/tracing/PerfLog.h"
+
+#include <memory>
+
+namespace bts {
+namespace orderman {
+
+class SendFixEvent : public mercury::Event<SendFixEvent>
+{
+public:
+   using Tag = GatewayName;
+   using SynopsisPolicy = mercury::NoSynopsis;
+   jetfix::OutFIXPtr fix;
+   trace::CorID corid;
+   SendFixEvent(const jetfix::OutFIXPtr& f, const trace::CorID& cid) : fix(f), corid(cid) {};
+};
+
+
+}}
+

src/lib/orderman/events/SendOrderMsgsEvent.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "lib/api/services/AdminService.pb.h"
+
+#include "lib/orderman/orderman_utils/orderman_common.h"
+#include "lib/mercury/common/mercury_common.h"
+#include "lib/mercury/events/Event.h"
+#include "lib/mercury/synopsis/NoSynopsis.h"
+#include "lib/orderman/actors/IExecutionGateway.h"
+#include "lib/tracing/PerfLog.h"
+
+#include <memory>
+
+namespace bts {
+namespace orderman {
+
+class SendOrderMsgsEvent : public mercury::Event<SendOrderMsgsEvent>
+{
+public:
+   using Tag = GatewayName;
+   using SynopsisPolicy = mercury::NoSynopsis;
+   using Msgs = std::vector<orderman::OrderMsg>;
+   using MsgsPtr = std::shared_ptr<Msgs>;
+   MsgsPtr msgs;
+   trace::CorID corid;
+   SendOrderMsgsEvent(const MsgsPtr& m, const trace::CorID& cid) : msgs(m), corid(cid) {};
+};
+
+
+}}
+

src/lib/orderman/server/OrderServiceServer.cpp

@@ -11,6 +11,8 @@
 #include "lib/mercury/actors/Timer.h"
 #include "lib/mercury/executor/Executor.h"
 #include "lib/metrics/metrics.h"
+#include "lib/orderman/actors/ActorThread.h"
+#include "lib/orderman/actors/gateway/Gateway.h"
 #include "lib/orderman/actors/AsyncFactory.h"
 #include "lib/orderman/actors/FillStore.h"
 #include "lib/orderman/actors/MDSource.h"
@@ -19,6 +21,7 @@
 #include "lib/orderman/actors/SecuritySrc.h"
 #include "lib/orderman/actors/ServiceBridge.h"
 #include "lib/orderman/actors/order_manager/OrderManager.h"
+#include "lib/orderman/actors/order_manager/OrderManagerShard.h"
 #include "lib/orderman/events/OrderOpResultEvent.h"
 #include "lib/orderman/order_factory/DynamicOrderRegistry.h"
 #include "lib/plugin/SparkServices.h"
@@ -32,65 +35,6 @@
 namespace bts {
 namespace orderman {
 
-struct ActorThread {
-   ActorThread(const config::OrdermanCfg::Executor &cfg,
-               metrics::MetricsContainer &stats)
-       : cfg(cfg), thread(), executor(), loop_metric(), done(false) {
-      std::array<double, 30> buckets;
-      auto i = 0;
-      for (auto &b : buckets)
-         b = (i += 1000);
-
-      loop_metric = std::make_shared<metrics::HistogramMetric<30>>(
-          fmt::format("spark_metrics_{}_thread_loop_ns", cfg.thread_name()),
-          buckets);
-      stats.publish(loop_metric);
-   }
-
-   void start(trace::PerfLog &log) {
-      if (cfg.poll())
-         thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
-            for (;;)
-               executor.do_work(log);
-         });
-      else {
-         thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
-            const auto inner_iter = 1000;
-            while (!done) {
-               auto start = Timer::now();
-               for (auto i = 0; i < inner_iter; i++)
-                  executor.do_work(log);
-               auto stop = Timer::now();
-               loop_metric->observe(ns_diff(stop, start) / inner_iter);
-               std::this_thread::sleep_for(10ms);
-            }
-         });
-      }
-   }
-
-   ~ActorThread() {
-      done = true;
-      thread.join();
-   }
-
-   config::OrdermanCfg::Executor cfg;
-   std::thread thread;
-   mercury::Executor executor;
-   std::shared_ptr<metrics::HistogramMetric<30>> loop_metric;
-   std::atomic<bool> done;
-};
-
-ActorThread *find_thread_for_actor(
-    std::string const &actor_name,
-    std::vector<std::unique_ptr<ActorThread>> const &all_executors) {
-   for (auto const &ex : all_executors)
-      for (auto const &hosted : ex->cfg.hosted_actor())
-         if (hosted == "*" || boost::algorithm::to_lower_copy(hosted) ==
-                                  boost::algorithm::to_lower_copy(actor_name))
-            return ex.get();
-   return nullptr;
-}
-
 struct OrderServiceServer::PImpl {
    PImpl(bts::api::ServiceContext &ctx, md::MDClient &mdc,
          DynamicOrderRegistry const &d, metrics::MetricsContainer &stats,
@@ -141,63 +85,65 @@ OrderServiceServer::OrderServiceServer(bts::api::ServiceContext &ctx,
       auto at = std::make_unique<ActorThread>(ex_cfg, stats);
       all_executors.push_back(&at->executor);
       me->actor_threads.push_back(std::move(at));
-   }
-
+   }   
+   Gateway::spawn_gateways(me->actor_threads, log, mdc.oracle());   
    for (auto &t : me->actor_threads)
       t->executor.init_neighbours(all_executors);
 
    // MDSource
-   auto thread = find_thread_for_actor("MDSource", me->actor_threads);
+   auto thread = ActorThread::find_thread_for_actor("MDSource", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for MDSource. See the "
                   "OrdermanCfg key");
 
    thread->executor.spawn<MDSource>(me->md_consumer, me->stats, log);
 
-   // OrderManager
-   thread = find_thread_for_actor("OrderManager", me->actor_threads);
-   bts_assert_msg(thread,
-                  "Could not find a configured thread for OrderManager. See "
-                  "the OrdermanCfg key");
-   thread->executor.spawn<OrderManager>(d, me->log, mdc.oracle());
+   if(!OrderManagerShard::spawn_shards(me->actor_threads, d, log, mdc.oracle())) {   
+      // OrderManager
+      thread = ActorThread::find_thread_for_actor("OrderManager", me->actor_threads);
+      bts_assert_msg(thread,
+                     "Could not find a configured thread for OrderManager. See "
+                     "the OrdermanCfg key");
+      thread->executor.spawn<OrderManager>(d, me->log, mdc.oracle(), "main_om");
+   }
 
    // Timer
-   thread = find_thread_for_actor("Timer", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("Timer", me->actor_threads);
    bts_assert_msg(
        thread,
        "Could not find a configured thread for Timer. See the OrdermanCfg key");
    thread->executor.spawn<mercury::Timer>();
 
    // AsyncFactory
-   thread = find_thread_for_actor("AsyncFactory", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("AsyncFactory", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for AsyncFactory. See "
                   "the OrdermanCfg key");
    thread->executor.spawn<AsyncFactory>(d, mdc.oracle());
 
    // OrderStore
-   thread = find_thread_for_actor("OrderStore", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("OrderStore", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for OrderStore. See "
                   "the OrdermanCfg key");
    thread->executor.spawn<OrderStore>(mdc.oracle());
 
    // FillStore
-   thread = find_thread_for_actor("FillStore", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("FillStore", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for FilllStore. See "
                   "the OrdermanCfg key");
    thread->executor.spawn<FillStore>(mdc.oracle());
 
    // SecuritySrc
-   thread = find_thread_for_actor("SecuritySrc", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("SecuritySrc", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for SecuritySrc. See "
                   "the OrdermanCfg key");
    thread->executor.spawn<SecuritySrc>(me->md_consumer);
 
    // MDSupplementarySource
-   thread = find_thread_for_actor("MDSupplementarySource", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("MDSupplementarySource", me->actor_threads);
    bts_assert_msg(
        thread,
        "Could not find a configured thread for MDSupplementarySource. See "
@@ -205,7 +151,7 @@ OrderServiceServer::OrderServiceServer(bts::api::ServiceContext &ctx,
    thread->executor.spawn<MDSupplementarySource>(me->md_consumer);
 
    // ServiceBridge
-   thread = find_thread_for_actor("ServiceBridge", me->actor_threads);
+   thread = ActorThread::find_thread_for_actor("ServiceBridge", me->actor_threads);
    bts_assert_msg(thread,
                   "Could not find a configured thread for ServiceBridge. See "
                   "the OrdermanCfg key");

src/server/sparkd/cmds/Run.cpp

@@ -5,6 +5,7 @@
 #include "lib/config/config.h"
 #include "lib/config/protos/Ports.pb.h"
 #include "lib/config/protos/quickbus_cfg.pb.h"
+#include "lib/config/protos/orderman_cfg.pb.h"
 #include "lib/logging/logging.h"
 #include "lib/os/os.h"
 #include "lib/quickbus/quickbus.h"
@@ -44,6 +45,7 @@
 #include "lib/orderman/orders/algo/algo_orders/ScratchAlgo.h"
 
 #include <thread>
+#include <memory>
 
 namespace bts {
 namespace cmd {
@@ -92,8 +94,7 @@ int Run::execute(std::vector<std::string> const &) {
    plugin::init_plugins(plugin_services);
    // finished initializing order plugins
 
-   orderman::OrderServiceServer order_service_server(
-       sc, mdclient, dyn_order_registry, stats, perf_log);
+   orderman::OrderServiceServer order_service(sc, mdclient, dyn_order_registry, stats, perf_log);
    support::ErrorReportServer error_report_service(sc);
    admin::AdminServiceServer admin_service(sc);
    md::auxiliary::MarketDataAuxiliaryServiceServer md_aux_server(sc);

src/web/CMakeDeps.cmake

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

src/web/api/build.sh

@@ -61,7 +61,7 @@ $PROTOC --plugin=protoc-gen-service=../../lib/api/service_compiler/plugin.py \
 
 # run typescript compiler to convert typescript to javascript
 echo -e \* ${ORANGE} tsc ${NOCOLOR}
-npx --no-install tsc -p tsconfig.$MODE.json
+npx --no-install tsc -p tsconfig.$MODE.json --outDir $OUT/api
 
 # cleanup
 
@@ -75,7 +75,7 @@ ts2fable() {
     wait
 }
 
-echo -e \* ${ORANGE} ts2fable ${NOCOLOR}
+echo -e \* ${ORANGE} ts2fable ${NOCOLOR} $OUT
 ts2fable $OUT/api
 # ts2fable $OUT/api/gen
 

tools/setup_builds.sh

@@ -1,8 +1,12 @@
 #/bin/bash
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 SRC_ROOT=$DIR/../
-mkdir -p $SRC_ROOT/build/
-BUILD_ROOT=`readlink -f $SRC_ROOT/build`
+#mkdir -p $SRC_ROOT/build/
+#BUILD_ROOT=`readlink -f $SRC_ROOT/build`
+cd $SRC_ROOT
+mkdir -p ./build/
+cd ./build
+BUILD_ROOT=`pwd`
 
 rm -rf $BUILD_ROOT/clang_debug 
 mkdir $BUILD_ROOT/clang_debug