Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions openc3/lib/openc3/microservices/interface_microservice.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 7 additions & 2 deletions openc3/python/openc3/microservices/interface_microservice.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
18 changes: 16 additions & 2 deletions openc3/python/test/microservices/test_interface_microservice.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 *
Expand Down Expand Up @@ -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)
Expand All @@ -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)
16 changes: 16 additions & 0 deletions openc3/spec/microservices/interface_microservice_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading