Allow renaming logged channels for replay
If we change an application to use a new channel for an existing
message, we need a way to support replaying old logs through it.
RenameLoggedChannel() provides an API to map the original channel to
the new one, updating the logged configuration as needed.
Since global maps for the new channel need not follow the same patterns
as the old one, any relevant maps must be specified by the user.
Change-Id: I68e9d2fe16bceaa60972a3f762f955e583c80255
Signed-off-by: Sanjay Narayanan <sanjay.narayanan@bluerivertech.com>
diff --git a/aos/events/logging/log_reader.cc b/aos/events/logging/log_reader.cc
index fa5e5d5..0af69c7 100644
--- a/aos/events/logging/log_reader.cc
+++ b/aos/events/logging/log_reader.cc
@@ -1331,45 +1331,8 @@
std::string_view add_prefix,
std::string_view new_type,
RemapConflict conflict_handling) {
- if (replay_channels_ != nullptr) {
- CHECK(std::find(replay_channels_->begin(), replay_channels_->end(),
- std::make_pair(std::string{name}, std::string{type})) !=
- replay_channels_->end())
- << "Attempted to remap channel " << name << " " << type
- << " which is not included in the replay channels passed to LogReader.";
- }
-
- for (size_t ii = 0; ii < logged_configuration()->channels()->size(); ++ii) {
- const Channel *const channel = logged_configuration()->channels()->Get(ii);
- if (channel->name()->str() == name &&
- channel->type()->string_view() == type) {
- CHECK_EQ(0u, remapped_channels_.count(ii))
- << "Already remapped channel "
- << configuration::CleanedChannelToString(channel);
- RemappedChannel remapped_channel;
- remapped_channel.remapped_name =
- std::string(add_prefix) + std::string(name);
- remapped_channel.new_type = new_type;
- const std::string_view remapped_type =
- remapped_channel.new_type.empty() ? type : remapped_channel.new_type;
- CheckAndHandleRemapConflict(
- remapped_channel.remapped_name, remapped_type,
- remapped_configuration_, conflict_handling,
- [this, &remapped_channel, remapped_type, add_prefix,
- conflict_handling]() {
- RemapLoggedChannel(remapped_channel.remapped_name, remapped_type,
- add_prefix, "", conflict_handling);
- });
- remapped_channels_[ii] = std::move(remapped_channel);
- VLOG(1) << "Remapping channel "
- << configuration::CleanedChannelToString(channel)
- << " to have name " << remapped_channels_[ii].remapped_name;
- MakeRemappedConfig();
- return;
- }
- }
- LOG(FATAL) << "Unabled to locate channel with name " << name << " and type "
- << type;
+ RemapLoggedChannel(name, type, nullptr, add_prefix, new_type,
+ conflict_handling);
}
void LogReader::RemapLoggedChannel(std::string_view name, std::string_view type,
@@ -1377,7 +1340,16 @@
std::string_view add_prefix,
std::string_view new_type,
RemapConflict conflict_handling) {
- VLOG(1) << "Node is " << aos::FlatbufferToJson(node);
+ if (node != nullptr) {
+ VLOG(1) << "Node is " << aos::FlatbufferToJson(node);
+ }
+ if (replay_channels_ != nullptr) {
+ CHECK(std::find(replay_channels_->begin(), replay_channels_->end(),
+ std::make_pair(std::string{name}, std::string{type})) !=
+ replay_channels_->end())
+ << "Attempted to remap channel " << name << " " << type
+ << " which is not included in the replay channels passed to LogReader.";
+ }
const Channel *remapped_channel =
configuration::GetChannel(logged_configuration(), name, type, "", node);
CHECK(remapped_channel != nullptr) << ": Failed to find {\"name\": \"" << name
@@ -1408,6 +1380,7 @@
maps_.emplace_back(std::move(new_map));
}
+ // Then remap the logged channel to the prefixed channel.
const size_t channel_index =
configuration::ChannelIndex(logged_configuration(), remapped_channel);
CHECK_EQ(0u, remapped_channels_.count(channel_index))
@@ -1432,6 +1405,51 @@
MakeRemappedConfig();
}
+void LogReader::RenameLoggedChannel(const std::string_view name,
+ const std::string_view type,
+ const std::string_view new_name,
+ const std::vector<MapT> &add_maps) {
+ RenameLoggedChannel(name, type, nullptr, new_name, add_maps);
+}
+
+void LogReader::RenameLoggedChannel(const std::string_view name,
+ const std::string_view type,
+ const Node *const node,
+ const std::string_view new_name,
+ const std::vector<MapT> &add_maps) {
+ if (node != nullptr) {
+ VLOG(1) << "Node is " << aos::FlatbufferToJson(node);
+ }
+ // First find the channel and rename it.
+ const Channel *remapped_channel =
+ configuration::GetChannel(logged_configuration(), name, type, "", node);
+ CHECK(remapped_channel != nullptr) << ": Failed to find {\"name\": \"" << name
+ << "\", \"type\": \"" << type << "\"}";
+ VLOG(1) << "Original {\"name\": \"" << name << "\", \"type\": \"" << type
+ << "\"}";
+ VLOG(1) << "Remapped "
+ << aos::configuration::StrippedChannelToString(remapped_channel);
+
+ const size_t channel_index =
+ configuration::ChannelIndex(logged_configuration(), remapped_channel);
+ CHECK_EQ(0u, remapped_channels_.count(channel_index))
+ << "Already remapped channel "
+ << configuration::CleanedChannelToString(remapped_channel);
+
+ RemappedChannel remapped_channel_struct;
+ remapped_channel_struct.remapped_name = new_name;
+ remapped_channel_struct.new_type.clear();
+ remapped_channels_[channel_index] = std::move(remapped_channel_struct);
+
+ // Then add any provided maps.
+ for (const MapT &map : add_maps) {
+ maps_.push_back(map);
+ }
+
+ // Finally rewrite the config.
+ MakeRemappedConfig();
+}
+
void LogReader::MakeRemappedConfig() {
for (std::unique_ptr<State> &state : states_) {
if (state) {
@@ -1589,20 +1607,26 @@
// Now create the new maps. These are second so they take effect first.
for (const MapT &map : maps_) {
+ CHECK(!map.match->name.empty());
const flatbuffers::Offset<flatbuffers::String> match_name_offset =
fbb.CreateString(map.match->name);
- const flatbuffers::Offset<flatbuffers::String> match_type_offset =
- fbb.CreateString(map.match->type);
- const flatbuffers::Offset<flatbuffers::String> rename_name_offset =
- fbb.CreateString(map.rename->name);
+ flatbuffers::Offset<flatbuffers::String> match_type_offset;
+ if (!map.match->type.empty()) {
+ match_type_offset = fbb.CreateString(map.match->type);
+ }
flatbuffers::Offset<flatbuffers::String> match_source_node_offset;
if (!map.match->source_node.empty()) {
match_source_node_offset = fbb.CreateString(map.match->source_node);
}
+ CHECK(!map.rename->name.empty());
+ const flatbuffers::Offset<flatbuffers::String> rename_name_offset =
+ fbb.CreateString(map.rename->name);
Channel::Builder match_builder(fbb);
match_builder.add_name(match_name_offset);
- match_builder.add_type(match_type_offset);
- if (!map.match->source_node.empty()) {
+ if (!match_type_offset.IsNull()) {
+ match_builder.add_type(match_type_offset);
+ }
+ if (!match_source_node_offset.IsNull()) {
match_builder.add_source_node(match_source_node_offset);
}
const flatbuffers::Offset<Channel> match_offset = match_builder.Finish();
diff --git a/aos/events/logging/log_reader.h b/aos/events/logging/log_reader.h
index d4936f1..6a0fcea 100644
--- a/aos/events/logging/log_reader.h
+++ b/aos/events/logging/log_reader.h
@@ -32,8 +32,7 @@
class EventNotifier;
// Vector of pair of name and type of the channel
-using ReplayChannels =
- std::vector<std::pair<std::string, std::string>>;
+using ReplayChannels = std::vector<std::pair<std::string, std::string>>;
// Vector of channel indices
using ReplayChannelIndices = std::vector<size_t>;
@@ -198,7 +197,6 @@
RemapLoggedChannel(name, T::GetFullyQualifiedName(), add_prefix, new_type,
conflict_handling);
}
-
// Remaps the provided channel, though this respects node mappings, and
// preserves them too. This makes it so if /aos -> /pi1/aos on one node,
// /original/aos -> /original/pi1/aos on the same node after renaming, just
@@ -221,11 +219,40 @@
new_type, conflict_handling);
}
+ // Similar to RemapLoggedChannel(), but lets you specify a name for the new
+ // channel without constraints. This is useful when an application has been
+ // updated to use new channels but you want to support replaying old logs. By
+ // default, this will not add any maps for the new channel. Use add_maps to
+ // specify any maps you'd like added.
+ void RenameLoggedChannel(std::string_view name, std::string_view type,
+ std::string_view new_name,
+ const std::vector<MapT> &add_maps = {});
+ template <typename T>
+ void RenameLoggedChannel(std::string_view name, std::string_view new_name,
+ const std::vector<MapT> &add_maps = {}) {
+ RenameLoggedChannel(name, T::GetFullyQualifiedName(), new_name, add_maps);
+ }
+ // The following overloads are more suitable for multi-node configurations,
+ // and let you rename a channel on a specific node.
+ void RenameLoggedChannel(std::string_view name, std::string_view type,
+ const Node *node, std::string_view new_name,
+ const std::vector<MapT> &add_maps = {});
+ template <typename T>
+ void RenameLoggedChannel(std::string_view name, const Node *node,
+ std::string_view new_name,
+ const std::vector<MapT> &add_maps = {}) {
+ RenameLoggedChannel(name, T::GetFullyQualifiedName(), node, new_name,
+ add_maps);
+ }
+
template <typename T>
bool HasChannel(std::string_view name, const Node *node = nullptr) {
- return configuration::GetChannel(logged_configuration(), name,
- T::GetFullyQualifiedName(), "", node,
- true) != nullptr;
+ return HasChannel(name, T::GetFullyQualifiedName(), node);
+ }
+ bool HasChannel(std::string_view name, std::string_view type,
+ const Node *node) {
+ return configuration::GetChannel(logged_configuration(), name, type, "",
+ node, true) != nullptr;
}
template <typename T>
@@ -235,6 +262,14 @@
RemapLoggedChannel<T>(name, node);
}
}
+ template <typename T>
+ void MaybeRenameLoggedChannel(std::string_view name, const Node *node,
+ std::string_view new_name,
+ const std::vector<MapT> &add_maps = {}) {
+ if (HasChannel<T>(name, node)) {
+ RenameLoggedChannel<T>(name, node, new_name, add_maps);
+ }
+ }
// Returns true if the channel exists on the node and was logged.
template <typename T>
diff --git a/aos/events/logging/logger_test.cc b/aos/events/logging/logger_test.cc
index fcbd77b..7214a36 100644
--- a/aos/events/logging/logger_test.cc
+++ b/aos/events/logging/logger_test.cc
@@ -96,9 +96,6 @@
// passing in a separate config.
LogReader reader(logfile, &config_.message());
- // Confirm that we can remap logged channels to point to new buses.
- reader.RemapLoggedChannel<aos::examples::Ping>("/test", "/original");
-
// This sends out the fetched messages and advances time to the start of the
// log file.
reader.Register();
@@ -111,8 +108,8 @@
int ping_count = 10;
int pong_count = 10;
- // Confirm that the ping value matches in the remapped channel location.
- test_event_loop->MakeWatcher("/original/test",
+ // Confirm that the ping value matches.
+ test_event_loop->MakeWatcher("/test",
[&ping_count](const examples::Ping &ping) {
EXPECT_EQ(ping.value(), ping_count + 1);
++ping_count;
@@ -128,6 +125,7 @@
reader.event_loop_factory()->RunFor(std::chrono::seconds(100));
EXPECT_EQ(ping_count, 2010);
+ EXPECT_EQ(pong_count, ping_count);
}
// Tests calling StartLogging twice.
diff --git a/aos/events/logging/multinode_logger_test.cc b/aos/events/logging/multinode_logger_test.cc
index f1784bf..18337be 100644
--- a/aos/events/logging/multinode_logger_test.cc
+++ b/aos/events/logging/multinode_logger_test.cc
@@ -166,10 +166,12 @@
UnorderedElementsAre(
std::make_tuple("/pi1/aos",
"aos.message_bridge.ServerStatistics", 1),
- std::make_tuple("/test", "aos.examples.Ping", 1)))
+ std::make_tuple("/test", "aos.examples.Ping", 1),
+ std::make_tuple("/pi1/aos", "aos.examples.Ping", 1)))
<< " : " << logfiles_[2];
{
std::vector<std::tuple<std::string, std::string, int>> channel_counts = {
+ std::make_tuple("/pi1/aos", "aos.examples.Ping", 10),
std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 1),
std::make_tuple("/pi1/aos", "aos.message_bridge.ClientStatistics",
1)};
@@ -185,12 +187,13 @@
}
{
std::vector<std::tuple<std::string, std::string, int>> channel_counts = {
+ std::make_tuple("/pi1/aos", "aos.examples.Ping", 1990),
std::make_tuple("/pi1/aos", "aos.message_bridge.Timestamp", 199),
std::make_tuple("/pi1/aos", "aos.message_bridge.ServerStatistics",
20),
std::make_tuple("/pi1/aos", "aos.message_bridge.ClientStatistics",
199),
- std::make_tuple("/pi1/aos", "aos.timing.Report", 40),
+ std::make_tuple("/pi1/aos", "aos.timing.Report", 60),
std::make_tuple("/test", "aos.examples.Ping", 2000)};
if (!std::get<0>(GetParam()).shared) {
channel_counts.push_back(
@@ -237,8 +240,10 @@
// Timing reports and pongs.
EXPECT_THAT(CountChannelsData(config, logfiles_[7]),
- UnorderedElementsAre(std::make_tuple(
- "/pi2/aos", "aos.message_bridge.ServerStatistics", 1)))
+ UnorderedElementsAre(
+ std::make_tuple("/pi2/aos", "aos.examples.Ping", 1),
+ std::make_tuple("/pi2/aos",
+ "aos.message_bridge.ServerStatistics", 1)))
<< " : " << logfiles_[7];
EXPECT_THAT(
CountChannelsData(config, logfiles_[8]),
@@ -247,12 +252,13 @@
EXPECT_THAT(
CountChannelsData(config, logfiles_[9]),
UnorderedElementsAre(
+ std::make_tuple("/pi2/aos", "aos.examples.Ping", 2000),
std::make_tuple("/pi2/aos", "aos.message_bridge.Timestamp", 200),
std::make_tuple("/pi2/aos", "aos.message_bridge.ServerStatistics",
20),
std::make_tuple("/pi2/aos", "aos.message_bridge.ClientStatistics",
200),
- std::make_tuple("/pi2/aos", "aos.timing.Report", 40),
+ std::make_tuple("/pi2/aos", "aos.timing.Report", 60),
std::make_tuple("/test", "aos.examples.Pong", 2000)))
<< " : " << logfiles_[9];
// And ping timestamps.
@@ -946,7 +952,7 @@
VerifyParts(sorted_parts);
}
-// Tests that if we remap a remapped channel, it shows up correctly.
+// Tests that if we remap a logged channel, it shows up correctly.
TEST_P(MultinodeLoggerTest, RemapLoggedChannel) {
time_converter_.StartEqual();
{
@@ -1025,6 +1031,117 @@
reader.Deregister();
}
+// Tests that if we rename a logged channel, it shows up correctly.
+TEST_P(MultinodeLoggerTest, RenameLoggedChannel) {
+ std::vector<std::string> actual_filenames;
+ time_converter_.StartEqual();
+ {
+ LoggerState pi1_logger = MakeLogger(pi1_);
+ LoggerState pi2_logger = MakeLogger(pi2_);
+
+ event_loop_factory_.RunFor(chrono::milliseconds(95));
+
+ StartLogger(&pi1_logger);
+ StartLogger(&pi2_logger);
+
+ event_loop_factory_.RunFor(chrono::milliseconds(20000));
+
+ pi1_logger.AppendAllFilenames(&actual_filenames);
+ pi2_logger.AppendAllFilenames(&actual_filenames);
+ }
+
+ LogReader reader(SortParts(actual_filenames));
+
+ // Rename just on pi2. Add some global maps just to verify they get added in
+ // the config and used correctly.
+ std::vector<MapT> maps;
+ {
+ MapT map;
+ map.match = std::make_unique<ChannelT>();
+ map.match->name = "/foo*";
+ map.match->source_node = "pi1";
+ map.rename = std::make_unique<ChannelT>();
+ map.rename->name = "/pi1/foo";
+ maps.emplace_back(std::move(map));
+ }
+ {
+ MapT map;
+ map.match = std::make_unique<ChannelT>();
+ map.match->name = "/foo*";
+ map.match->source_node = "pi2";
+ map.rename = std::make_unique<ChannelT>();
+ map.rename->name = "/pi2/foo";
+ maps.emplace_back(std::move(map));
+ }
+ {
+ MapT map;
+ map.match = std::make_unique<ChannelT>();
+ map.match->name = "/foo";
+ map.match->type = "aos.examples.Ping";
+ map.rename = std::make_unique<ChannelT>();
+ map.rename->name = "/foo/renamed";
+ maps.emplace_back(std::move(map));
+ }
+ reader.RenameLoggedChannel<aos::examples::Ping>(
+ "/aos", configuration::GetNode(reader.configuration(), "pi2"),
+ "/pi2/foo/renamed", maps);
+
+ SimulatedEventLoopFactory log_reader_factory(reader.configuration());
+ log_reader_factory.set_send_delay(chrono::microseconds(0));
+
+ std::vector<const Channel *> remapped_channels = reader.RemappedChannels();
+ // Note: An extra channel gets remapped automatically due to a timestamp
+ // channel being LOCAL_LOGGER'd.
+ const bool shared = std::get<0>(GetParam()).shared;
+ ASSERT_EQ(remapped_channels.size(), shared ? 1u : 2u);
+ EXPECT_EQ(remapped_channels[shared ? 0 : 1]->name()->string_view(),
+ "/pi2/foo/renamed");
+ EXPECT_EQ(remapped_channels[shared ? 0 : 1]->type()->string_view(),
+ "aos.examples.Ping");
+ if (!shared) {
+ EXPECT_EQ(remapped_channels[0]->name()->string_view(),
+ "/original/pi1/aos/remote_timestamps/pi2/pi1/aos/"
+ "aos-message_bridge-Timestamp");
+ EXPECT_EQ(remapped_channels[0]->type()->string_view(),
+ "aos.message_bridge.RemoteMessage");
+ }
+
+ reader.Register(&log_reader_factory);
+
+ const Node *pi1 =
+ configuration::GetNode(log_reader_factory.configuration(), "pi1");
+ const Node *pi2 =
+ configuration::GetNode(log_reader_factory.configuration(), "pi2");
+
+ // Confirm we can read the data on the renamed channel, just for pi2. Nothing
+ // else should have moved.
+ std::unique_ptr<EventLoop> pi2_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi2);
+ pi2_event_loop->SkipTimingReport();
+ std::unique_ptr<EventLoop> full_pi2_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi2);
+ full_pi2_event_loop->SkipTimingReport();
+ std::unique_ptr<EventLoop> pi1_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi1);
+ pi1_event_loop->SkipTimingReport();
+
+ MessageCounter<aos::examples::Ping> pi2_ping(pi2_event_loop.get(), "/aos");
+ MessageCounter<aos::examples::Ping> pi2_renamed_ping(pi2_event_loop.get(),
+ "/foo");
+ MessageCounter<aos::examples::Ping> full_pi2_renamed_ping(
+ full_pi2_event_loop.get(), "/pi2/foo/renamed");
+ MessageCounter<aos::examples::Ping> pi1_ping(pi1_event_loop.get(), "/aos");
+
+ log_reader_factory.Run();
+
+ EXPECT_EQ(pi2_ping.count(), 0u);
+ EXPECT_NE(pi2_renamed_ping.count(), 0u);
+ EXPECT_NE(full_pi2_renamed_ping.count(), 0u);
+ EXPECT_NE(pi1_ping.count(), 0u);
+
+ reader.Deregister();
+}
+
// Tests that we can remap a forwarded channel as well.
TEST_P(MultinodeLoggerTest, RemapForwardedLoggedChannel) {
time_converter_.StartEqual();
@@ -1102,6 +1219,109 @@
reader.Deregister();
}
+// Tests that we can rename a forwarded channel as well.
+TEST_P(MultinodeLoggerTest, RenameForwardedLoggedChannel) {
+ std::vector<std::string> actual_filenames;
+ time_converter_.StartEqual();
+ {
+ LoggerState pi1_logger = MakeLogger(pi1_);
+ LoggerState pi2_logger = MakeLogger(pi2_);
+
+ event_loop_factory_.RunFor(chrono::milliseconds(95));
+
+ StartLogger(&pi1_logger);
+ StartLogger(&pi2_logger);
+
+ event_loop_factory_.RunFor(chrono::milliseconds(20000));
+
+ pi1_logger.AppendAllFilenames(&actual_filenames);
+ pi2_logger.AppendAllFilenames(&actual_filenames);
+ }
+
+ LogReader reader(SortParts(actual_filenames));
+
+ std::vector<MapT> maps;
+ {
+ MapT map;
+ map.match = std::make_unique<ChannelT>();
+ map.match->name = "/production*";
+ map.match->source_node = "pi1";
+ map.rename = std::make_unique<ChannelT>();
+ map.rename->name = "/pi1/production";
+ maps.emplace_back(std::move(map));
+ }
+ {
+ MapT map;
+ map.match = std::make_unique<ChannelT>();
+ map.match->name = "/production*";
+ map.match->source_node = "pi2";
+ map.rename = std::make_unique<ChannelT>();
+ map.rename->name = "/pi2/production";
+ maps.emplace_back(std::move(map));
+ }
+ reader.RenameLoggedChannel<aos::examples::Ping>(
+ "/test", configuration::GetNode(reader.configuration(), "pi1"),
+ "/pi1/production", maps);
+
+ SimulatedEventLoopFactory log_reader_factory(reader.configuration());
+ log_reader_factory.set_send_delay(chrono::microseconds(0));
+
+ reader.Register(&log_reader_factory);
+
+ const Node *pi1 =
+ configuration::GetNode(log_reader_factory.configuration(), "pi1");
+ const Node *pi2 =
+ configuration::GetNode(log_reader_factory.configuration(), "pi2");
+
+ // Confirm we can read the data on the renamed channel, on both the source
+ // node and the remote node. In case of split timestamp channels, confirm that
+ // we receive the timestamp messages on the renamed channel as well.
+ std::unique_ptr<EventLoop> pi1_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi1);
+ pi1_event_loop->SkipTimingReport();
+ std::unique_ptr<EventLoop> full_pi1_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi1);
+ full_pi1_event_loop->SkipTimingReport();
+ std::unique_ptr<EventLoop> pi2_event_loop =
+ log_reader_factory.MakeEventLoop("test", pi2);
+ pi2_event_loop->SkipTimingReport();
+
+ MessageCounter<examples::Ping> pi1_ping(pi1_event_loop.get(), "/test");
+ MessageCounter<examples::Ping> pi2_ping(pi2_event_loop.get(), "/test");
+ MessageCounter<examples::Ping> pi1_renamed_ping(pi1_event_loop.get(),
+ "/pi1/production");
+ MessageCounter<examples::Ping> pi2_renamed_ping(pi2_event_loop.get(),
+ "/pi1/production");
+
+ std::unique_ptr<MessageCounter<message_bridge::RemoteMessage>>
+ pi1_renamed_ping_timestamp;
+ std::unique_ptr<MessageCounter<message_bridge::RemoteMessage>>
+ pi1_ping_timestamp;
+ if (!shared()) {
+ pi1_renamed_ping_timestamp =
+ std::make_unique<MessageCounter<message_bridge::RemoteMessage>>(
+ pi1_event_loop.get(),
+ "/pi1/aos/remote_timestamps/pi2/pi1/production/aos-examples-Ping");
+ pi1_ping_timestamp =
+ std::make_unique<MessageCounter<message_bridge::RemoteMessage>>(
+ pi1_event_loop.get(),
+ "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping");
+ }
+
+ log_reader_factory.Run();
+
+ EXPECT_EQ(pi1_ping.count(), 0u);
+ EXPECT_EQ(pi2_ping.count(), 0u);
+ EXPECT_NE(pi1_renamed_ping.count(), 0u);
+ EXPECT_NE(pi2_renamed_ping.count(), 0u);
+ if (!shared()) {
+ EXPECT_NE(pi1_renamed_ping_timestamp->count(), 0u);
+ EXPECT_EQ(pi1_ping_timestamp->count(), 0u);
+ }
+
+ reader.Deregister();
+}
+
// Tests that we observe all the same events in log replay (for a given node)
// whether we just register an event loop for that node or if we register a full
// event loop factory.
diff --git a/aos/events/logging/multinode_logger_test_lib.cc b/aos/events/logging/multinode_logger_test_lib.cc
index 3ffb091..22822e7 100644
--- a/aos/events/logging/multinode_logger_test_lib.cc
+++ b/aos/events/logging/multinode_logger_test_lib.cc
@@ -115,8 +115,14 @@
LOG(INFO) << "Logging data to " << logfiles_[0] << ", " << logfiles_[1]
<< " and " << logfiles_[2];
- pi1_->OnStartup([this]() { pi1_->AlwaysStart<Ping>("ping"); });
- pi2_->OnStartup([this]() { pi2_->AlwaysStart<Pong>("pong"); });
+ pi1_->OnStartup([this]() {
+ pi1_->AlwaysStart<Ping>("ping");
+ pi1_->AlwaysStart<Ping>("ping_local", "/aos");
+ });
+ pi2_->OnStartup([this]() {
+ pi2_->AlwaysStart<Pong>("pong");
+ pi2_->AlwaysStart<Ping>("ping_local", "/aos");
+ });
}
bool MultinodeLoggerTest::shared() const {
diff --git a/aos/events/logging/multinode_logger_test_lib.h b/aos/events/logging/multinode_logger_test_lib.h
index f3f322e..40b5933 100644
--- a/aos/events/logging/multinode_logger_test_lib.h
+++ b/aos/events/logging/multinode_logger_test_lib.h
@@ -56,13 +56,13 @@
};
constexpr std::string_view kCombinedConfigSha1() {
- return "5d73fe35bacaa59d24f8f0c1a806fe10b783b0fcc80809ee30a9db824e82538b";
+ return "c8cd3762e42a4e19b2155f63ccec97d1627a2fbd34d3da3ea6541128ca22b899";
}
constexpr std::string_view kSplitConfigSha1() {
- return "f25e8f6f90d61f41c41517e652300566228b077e44cd86f1af2af4a9bed31ad4";
+ return "0ee6360b3e82a46f3f8b241661934abac53957d494a81ed1938899c220334954";
}
constexpr std::string_view kReloggedSplitConfigSha1() {
- return "f1fabd629bdf8735c3d81bc791d7a454e8e636951c26cba6426545cbc97f911f";
+ return "cc31e1a644dd7bf65d72247aea3e09b3474753e01921f3b6272f8233f288a16b";
}
LoggerState MakeLoggerState(NodeEventLoopFactory *node,
diff --git a/aos/events/logging/multinode_pingpong_combined.json b/aos/events/logging/multinode_pingpong_combined.json
index 6686360..4864674 100644
--- a/aos/events/logging/multinode_pingpong_combined.json
+++ b/aos/events/logging/multinode_pingpong_combined.json
@@ -112,6 +112,18 @@
"num_senders": 2,
"source_node": "pi2"
},
+ {
+ "name": "/pi1/aos",
+ "type": "aos.examples.Ping",
+ "source_node": "pi1",
+ "frequency": 150
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.examples.Ping",
+ "source_node": "pi2",
+ "frequency": 150
+ },
/* Forwarded to pi2 */
{
"name": "/test",
diff --git a/aos/events/logging/multinode_pingpong_split.json b/aos/events/logging/multinode_pingpong_split.json
index f5bd9cb..4cd6248 100644
--- a/aos/events/logging/multinode_pingpong_split.json
+++ b/aos/events/logging/multinode_pingpong_split.json
@@ -35,6 +35,18 @@
},
{
"name": "/pi1/aos",
+ "type": "aos.examples.Ping",
+ "source_node": "pi1",
+ "frequency": 150
+ },
+ {
+ "name": "/pi2/aos",
+ "type": "aos.examples.Ping",
+ "source_node": "pi2",
+ "frequency": 150
+ },
+ {
+ "name": "/pi1/aos",
"type": "aos.message_bridge.ServerStatistics",
"logger": "LOCAL_LOGGER",
"source_node": "pi1"
diff --git a/aos/events/ping_lib.cc b/aos/events/ping_lib.cc
index f9958ff..e80fd75 100644
--- a/aos/events/ping_lib.cc
+++ b/aos/events/ping_lib.cc
@@ -12,14 +12,16 @@
namespace chrono = std::chrono;
-Ping::Ping(EventLoop *event_loop)
+Ping::Ping(EventLoop *event_loop, std::string_view channel_name)
: event_loop_(event_loop),
- sender_(event_loop_->MakeSender<examples::Ping>("/test")) {
+ sender_(event_loop_->MakeSender<examples::Ping>(channel_name)) {
timer_handle_ = event_loop_->AddTimer([this]() { SendPing(); });
timer_handle_->set_name("ping");
- event_loop_->MakeWatcher(
- "/test", [this](const examples::Pong &pong) { HandlePong(pong); });
+ if (event_loop_->HasChannel<examples::Pong>(channel_name)) {
+ event_loop_->MakeWatcher(
+ channel_name, [this](const examples::Pong &pong) { HandlePong(pong); });
+ }
event_loop_->OnRun([this]() {
timer_handle_->Setup(event_loop_->monotonic_now(),
diff --git a/aos/events/ping_lib.h b/aos/events/ping_lib.h
index 0b83d52..fec6c45 100644
--- a/aos/events/ping_lib.h
+++ b/aos/events/ping_lib.h
@@ -2,6 +2,7 @@
#define AOS_EVENTS_PING_LIB_H_
#include <chrono>
+#include <string_view>
#include "aos/events/event_loop.h"
#include "aos/events/ping_generated.h"
@@ -12,7 +13,7 @@
// Class which sends out a Ping message every X ms, and times the response.
class Ping {
public:
- Ping(EventLoop *event_loop);
+ Ping(EventLoop *event_loop, std::string_view channel_name = "/test");
void set_quiet(bool quiet) { quiet_ = quiet; }