diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 723eda07ea..3ebe28e9dd 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -606,8 +606,11 @@ def initialize(name) # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis @queued = update_interval > 0.0 if @queued - EphemeralStoreQueued.instance.set_update_interval(update_interval) - StoreQueued.instance.set_update_interval(update_interval) + db_shards = [0] + @interface.tlm_target_names.map { |target_name| Store.db_shard_for_target(target_name, scope: @scope) } + db_shards.uniq.each do |db_shard| + EphemeralStoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) + StoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) + end end @interface_thread_sleeper = Sleeper.new diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 120fc6f05a..e72e15c3dd 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -43,6 +43,7 @@ from openc3.utilities.json import JsonDecoder, JsonEncoder from openc3.utilities.logger import Logger from openc3.utilities.sleeper import Sleeper +from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.thread_manager import ThreadManager from openc3.utilities.time import from_nsec_from_epoch @@ -692,8 +693,12 @@ def __init__(self, name): # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis self.queued = update_interval > 0.0 if self.queued: - EphemeralStoreQueued.instance().set_update_interval(update_interval) - StoreQueued.instance().set_update_interval(update_interval) + db_shards = {0} + for target_name in self.interface.tlm_target_names: + db_shards.add(Store.db_shard_for_target(target_name, scope=self.scope)) + for db_shard in db_shards: + EphemeralStoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) + StoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) if self.interface_or_router == "INTERFACE": self.handler_thread = InterfaceCmdHandlerThread( diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index a5d4dbe0be..7eb919c6df 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -28,6 +28,7 @@ from openc3.topics.interface_topic import InterfaceTopic from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic from openc3.topics.topic import Topic +from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.time import from_nsec_from_epoch from test.test_helper import * @@ -887,7 +888,7 @@ def test_queues_writes_by_default_with_default_update_interval(self): im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") self.addCleanup(im.shutdown) - self.assertEqual(im.queued, True) + self.assertTrue(im.queued) self.assertEqual(StoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) self.assertEqual(EphemeralStoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) self.assertEqual(InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL, 0.1) @@ -909,7 +910,20 @@ def test_update_interval_zero_disables_queued_writes(self): im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") self.addCleanup(im.shutdown) - self.assertEqual(im.queued, False) + self.assertFalse(im.queued) im.shutdown() time.sleep(0.1) # Allow threads to exit + + def test_applies_default_update_interval_to_target_shards(self): + with ( + patch.object(Store, "db_shard_for_target", return_value=1), + patch.object(EphemeralStoreQueued, "instance") as ephemeral_instance, + patch.object(StoreQueued, "instance") as store_instance, + ): + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + for instance in (ephemeral_instance, store_instance): + instance.assert_any_call(db_shard=0) + instance.assert_any_call(db_shard=1) + instance.return_value.set_update_interval.assert_called_with(0.1) diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 8b198958cf..15aff06fb0 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -192,6 +192,22 @@ class ApiTest sleep 0.1 # Allow threads to exit end + it "applies the default update interval to target shards" do + InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: []).update + allow(Store).to receive(:db_shard_for_target).and_return(1) + [EphemeralStoreQueued, StoreQueued].each do |store_class| + [0, 1].each do |db_shard| + queued_store = double("queued store") + allow(store_class).to receive(:instance).with(db_shard: db_shard).and_return(queued_store) + expect(queued_store).to receive(:set_update_interval).with(0.1) + end + end + + im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + im.shutdown + sleep 0.1 # Allow threads to exit + end + it "disables queued writes when UPDATE_INTERVAL is 0" do InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: [["UPDATE_INTERVAL", "0"]]).update