diff --git a/ntcore/src/main/java/edu/wpi/first/networktables/PubSubOption.java b/ntcore/src/main/java/edu/wpi/first/networktables/PubSubOption.java index 912d46c138..7cf072474c 100644 --- a/ntcore/src/main/java/edu/wpi/first/networktables/PubSubOption.java +++ b/ntcore/src/main/java/edu/wpi/first/networktables/PubSubOption.java @@ -11,6 +11,7 @@ public class PubSubOption { private static final int kTopicsOnly = 3; private static final int kPollStorage = 4; private static final int kKeepDuplicates = 5; + private static final int kLocalRemote = 6; PubSubOption(int type, double value) { m_type = type; @@ -64,8 +65,8 @@ public class PubSubOption { /** * Polling storage for subscription. Specifies the maximum number of updates NetworkTables should - * store between calls to the subscriber's poll() function. Defaults to 1 if sendAll is false, 20 - * if sendAll is true. + * store between calls to the subscriber's readQueue() function. Defaults to 1 if sendAll is + * false, 20 if sendAll is true. * * @param depth number of entries to save for polling. * @return option @@ -74,6 +75,36 @@ public class PubSubOption { return new PubSubOption(kPollStorage, depth); } + /** + * If only local value updates should be queued for readQueue(). See also remoteOnly() and + * allUpdates(). Default is allUpdates. Only has an effect on subscriptions. + * + * @return option + */ + public static PubSubOption localOnly() { + return new PubSubOption(kLocalRemote, 1.0); + } + + /** + * If only remote value updates should be queued for readQueue(). See also localOnly() and + * allUpdates(). Default is allUpdates. Only has an effect on subscriptions. + * + * @return option + */ + public static PubSubOption remoteOnly() { + return new PubSubOption(kLocalRemote, 2.0); + } + + /** + * If both local and remote value updates should be queued for readQueue(). See also localOnly() + * and remoteOnly(). Default is allUpdates. Only has an effect on subscriptions. + * + * @return option + */ + public static PubSubOption allUpdates() { + return new PubSubOption(kLocalRemote, 0.0); + } + final int m_type; final double m_value; } diff --git a/ntcore/src/main/native/cpp/LocalStorage.cpp b/ntcore/src/main/native/cpp/LocalStorage.cpp index b02531de64..75806bfbaa 100644 --- a/ntcore/src/main/native/cpp/LocalStorage.cpp +++ b/ntcore/src/main/native/cpp/LocalStorage.cpp @@ -492,9 +492,12 @@ bool LSImpl::SetValue(TopicData* topic, const Value& value, void LSImpl::NotifyValue(TopicData* topic, unsigned int eventFlags, bool isDuplicate) { + bool isNetwork = (eventFlags & NT_EVENT_VALUE_REMOTE) != 0; for (auto&& subscriber : topic->localSubscribers) { if (subscriber->active && - (subscriber->config.keepDuplicates || !isDuplicate)) { + (subscriber->config.keepDuplicates || !isDuplicate) && + ((isNetwork && subscriber->config.fromRemote) || + (!isNetwork && subscriber->config.fromLocal))) { subscriber->pollStorage.emplace_back(topic->lastValue); subscriber->handle.Set(); if (!subscriber->valueListeners.empty()) { diff --git a/ntcore/src/main/native/cpp/PubSubOptions.cpp b/ntcore/src/main/native/cpp/PubSubOptions.cpp index 0af0d4d4df..27954f9228 100644 --- a/ntcore/src/main/native/cpp/PubSubOptions.cpp +++ b/ntcore/src/main/native/cpp/PubSubOptions.cpp @@ -29,6 +29,25 @@ nt::PubSubOptions::PubSubOptions(std::span options) { case NT_PUBSUB_POLLSTORAGE: pollStorageSize = static_cast(option.value); break; + case NT_PUBSUB_LOCALREMOTE: + switch (static_cast(option.value)) { + case 0: + case 3: + fromLocal = true; + fromRemote = true; + break; + case 1: + fromLocal = true; + fromRemote = false; + break; + case 2: + fromLocal = false; + fromRemote = true; + break; + default: + break; + } + break; default: break; } diff --git a/ntcore/src/main/native/cpp/PubSubOptions.h b/ntcore/src/main/native/cpp/PubSubOptions.h index ab3db4d7e1..40e3194831 100644 --- a/ntcore/src/main/native/cpp/PubSubOptions.h +++ b/ntcore/src/main/native/cpp/PubSubOptions.h @@ -22,6 +22,8 @@ class PubSubOptions { bool topicsOnly = false; bool prefixMatch = false; bool keepDuplicates = false; + bool fromRemote = true; + bool fromLocal = true; }; } // namespace nt diff --git a/ntcore/src/main/native/include/ntcore_c.h b/ntcore/src/main/native/include/ntcore_c.h index a40349ee00..e9b8255d7e 100644 --- a/ntcore/src/main/native/include/ntcore_c.h +++ b/ntcore/src/main/native/include/ntcore_c.h @@ -95,6 +95,7 @@ enum NT_PubSubOptionType { NT_PUBSUB_TOPICSONLY, /* only send topic changes, no value changes */ NT_PUBSUB_POLLSTORAGE, /* polling storage for subscription */ NT_PUBSUB_KEEPDUPLICATES, /* preserve duplicate values */ + NT_PUBSUB_LOCALREMOTE, /* local, remote, or any value changes */ }; /** Event notification flags. */ @@ -287,7 +288,8 @@ struct NT_PubSubOption { /** * Option value. 1 (true) or 0 (false) for immediate and logging options, - * time between updates, in seconds, for periodic option. + * time between updates, in seconds, for periodic option. For local/remote + * option, 1=local only, 2=remote only, 0 or 3=both local and remote. */ double value; }; diff --git a/ntcore/src/main/native/include/ntcore_cpp.h b/ntcore/src/main/native/include/ntcore_cpp.h index 1b1e4717cf..c5b01698a7 100644 --- a/ntcore/src/main/native/include/ntcore_cpp.h +++ b/ntcore/src/main/native/include/ntcore_cpp.h @@ -314,7 +314,7 @@ class PubSubOption { /** * Polling storage for subscription. Specifies the maximum number of updates - * NetworkTables should store between calls to the subscriber's poll() + * NetworkTables should store between calls to the subscriber's ReadQueue() * function. Defaults to 1 if SendAll is false, 20 if SendAll is true. * * @param depth number of entries to save for polling. @@ -324,6 +324,39 @@ class PubSubOption { return PubSubOption{NT_PUBSUB_POLLSTORAGE, static_cast(depth)}; } + /** + * If only local value updates should be queued for ReadQueue(). See also + * RemoteOnly() and AllUpdates(). Default is AllUpdates. Only has an effect on + * subscriptions. + * + * @return option + */ + static constexpr PubSubOption LocalOnly() { + return PubSubOption{NT_PUBSUB_LOCALREMOTE, 1.0}; + } + + /** + * If only remote value updates should be queued for ReadQueue(). See also + * LocalOnly() and AllUpdates(). Default is AllUpdates. Only has an effect on + * subscriptions. + * + * @return option + */ + static constexpr PubSubOption RemoteOnly() { + return PubSubOption{NT_PUBSUB_LOCALREMOTE, 2.0}; + } + + /** + * If both local and remote value updates should be queued for ReadQueue(). + * See also LocalOnly() and RemoteOnly(). Default is AllUpdates. Only has an + * effect on subscriptions. + * + * @return option + */ + static constexpr PubSubOption AllUpdates() { + return PubSubOption{NT_PUBSUB_LOCALREMOTE, 0.0}; + } + NT_PubSubOptionType type; double value; }; diff --git a/ntcore/src/test/native/cpp/LocalStorageTest.cpp b/ntcore/src/test/native/cpp/LocalStorageTest.cpp index 411d1cbabb..cb4d22b325 100644 --- a/ntcore/src/test/native/cpp/LocalStorageTest.cpp +++ b/ntcore/src/test/native/cpp/LocalStorageTest.cpp @@ -807,4 +807,36 @@ TEST_F(LocalStorageTest, NetworkDuplicateDetect) { storage.SetEntryValue(pub, Value::MakeDouble(1.0, 80)); } +TEST_F(LocalStorageTest, ReadQueueLocalRemote) { + EXPECT_CALL(network, Subscribe(_, _, _)).Times(3); + EXPECT_CALL(network, Publish(_, _, _, _, _, _)).Times(1); + + auto subBoth = storage.Subscribe(fooTopic, NT_DOUBLE, "double", + {{PubSubOption::AllUpdates()}}); + auto subLocal = storage.Subscribe(fooTopic, NT_DOUBLE, "double", + {{PubSubOption::LocalOnly()}}); + auto subRemote = storage.Subscribe(fooTopic, NT_DOUBLE, "double", + {{PubSubOption::RemoteOnly()}}); + auto pub = storage.Publish(fooTopic, NT_DOUBLE, "double", {}, {}); + auto remoteTopic = + storage.NetworkAnnounce("foo", "double", wpi::json::object(), 0); + + // local set + EXPECT_CALL(network, SetValue(_, _)); + storage.SetEntryValue(pub, Value::MakeDouble(1.0, 50)); + EXPECT_THAT(storage.ReadQueueDouble(subBoth), + ElementsAre(TSEq(1.0, 50))); + EXPECT_THAT(storage.ReadQueueDouble(subLocal), + ElementsAre(TSEq(1.0, 50))); + EXPECT_THAT(storage.ReadQueueDouble(subRemote), IsEmpty()); + + // network set + storage.NetworkSetValue(remoteTopic, Value::MakeDouble(2.0, 60)); + EXPECT_THAT(storage.ReadQueueDouble(subBoth), + ElementsAre(TSEq(2.0, 60))); + EXPECT_THAT(storage.ReadQueueDouble(subRemote), + ElementsAre(TSEq(2.0, 60))); + EXPECT_THAT(storage.ReadQueueDouble(subLocal), IsEmpty()); +} + } // namespace nt