text: #include <chrono>
  #include <thread>
  
  #include "absl/strings/str_cat.h"
  #include "aos/events/ping_generated.h"
  #include "aos/events/pong_generated.h"
  #include "aos/ipc_lib/event.h"
  #include "aos/network/message_bridge_client_lib.h"
  #include "aos/network/message_bridge_server_lib.h"
  #include "aos/network/team_number.h"
  #include "aos/testing/path.h"
  #include "aos/util/file.h"
  #include "gtest/gtest.h"
  
  DECLARE_string(boot_uuid);
  
  namespace aos {
  void SetShmBase(const std::string_view base);
  
  namespace message_bridge {
  namespace testing {
  
  using aos::testing::ArtifactPath;
  
  namespace chrono = std::chrono;
  
  std::string ShmBase(const std::string_view node) {
    const char *tmpdir_c_str = getenv("TEST_TMPDIR");
    if (tmpdir_c_str != nullptr) {
      return absl::StrCat(tmpdir_c_str, "/", node);
    } else {
      return absl::StrCat("/dev/shm/", node);
    }
  }
  
  void DoSetShmBase(const std::string_view node) {
    aos::SetShmBase(ShmBase(node));
  }
  
  // Parameters to run all the tests with.
  struct Param {
    // The config file to use.
    std::string config;
    // If true, the RemoteMessage channel should be shared between all the remote
    // channels.  If false, there will be 1 RemoteMessage channel per remote
    // channel.
    bool shared;
  };
  
  class MessageBridgeParameterizedTest
      : public ::testing::TestWithParam<struct Param> {
   public:
    MessageBridgeParameterizedTest()
        : config(aos::configuration::ReadConfig(
              ArtifactPath(absl::StrCat("aos/network/", GetParam().config)))),
          pi1_boot_uuid_(UUID::Random()),
          pi2_boot_uuid_(UUID::Random()) {
      util::UnlinkRecursive(ShmBase("pi1"));
      util::UnlinkRecursive(ShmBase("pi2"));
    }
  
    bool shared() const { return GetParam().shared; }
  
    void OnPi1() {
      DoSetShmBase("pi1");
      FLAGS_override_hostname = "raspberrypi";
      FLAGS_boot_uuid = pi1_boot_uuid_.ToString();
    }
  
    void OnPi2() {
      DoSetShmBase("pi2");
      FLAGS_override_hostname = "raspberrypi2";
      FLAGS_boot_uuid = pi2_boot_uuid_.ToString();
    }
  
    void MakePi1Server() {
      OnPi1();
      FLAGS_application_name = "pi1_message_bridge_server";
      pi1_server_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
      pi1_server_event_loop->SetRuntimeRealtimePriority(1);
      pi1_message_bridge_server =
          std::make_unique<MessageBridgeServer>(pi1_server_event_loop.get());
    }
  
    void RunPi1Server(chrono::nanoseconds duration) {
      // Setup a shutdown callback.
      aos::TimerHandler *const quit = pi1_server_event_loop->AddTimer(
          [this]() { pi1_server_event_loop->Exit(); });
      pi1_server_event_loop->OnRun([this, quit, duration]() {
        // Stop between timestamps, not exactly on them.
        quit->Setup(pi1_server_event_loop->monotonic_now() + duration);
      });
  
      pi1_server_event_loop->Run();
    }
  
    void StartPi1Server() {
      pi1_server_thread = std::thread([this]() {
        LOG(INFO) << "Started pi1_message_bridge_server";
        pi1_server_event_loop->Run();
      });
    }
  
    void StopPi1Server() {
      if (pi1_server_thread.joinable()) {
        pi1_server_event_loop->Exit();
        pi1_server_thread.join();
        pi1_server_thread = std::thread();
      }
      pi1_message_bridge_server.reset();
      pi1_server_event_loop.reset();
    }
  
    void MakePi1Client() {
      OnPi1();
      FLAGS_application_name = "pi1_message_bridge_client";
      pi1_client_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
      pi1_client_event_loop->SetRuntimeRealtimePriority(1);
      pi1_message_bridge_client =
          std::make_unique<MessageBridgeClient>(pi1_client_event_loop.get());
    }
  
    void StartPi1Client() {
      pi1_client_thread = std::thread([this]() {
        LOG(INFO) << "Started pi1_message_bridge_client";
        pi1_client_event_loop->Run();
      });
    }
  
    void StopPi1Client() {
      pi1_client_event_loop->Exit();
      pi1_client_thread.join();
      pi1_client_thread = std::thread();
      pi1_message_bridge_client.reset();
      pi1_client_event_loop.reset();
    }
  
    void MakePi1Test() {
      OnPi1();
      FLAGS_application_name = "test1";
      pi1_test_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
  
      pi1_test_event_loop->MakeWatcher(
          "/pi1/aos", [](const ServerStatistics &stats) {
            VLOG(1) << "/pi1/aos ServerStatistics " << FlatbufferToJson(&stats);
          });
  
      pi1_test_event_loop->MakeWatcher(
          "/pi1/aos", [](const ClientStatistics &stats) {
            VLOG(1) << "/pi1/aos ClientStatistics " << FlatbufferToJson(&stats);
          });
  
      pi1_test_event_loop->MakeWatcher(
          "/pi1/aos", [](const Timestamp &timestamp) {
            VLOG(1) << "/pi1/aos Timestamp " << FlatbufferToJson(&timestamp);
          });
      pi1_test_event_loop->MakeWatcher(
          "/pi2/aos", [this](const Timestamp &timestamp) {
            VLOG(1) << "/pi2/aos Timestamp " << FlatbufferToJson(&timestamp);
            EXPECT_EQ(pi1_test_event_loop->context().source_boot_uuid,
                      pi2_boot_uuid_);
          });
    }
  
    void StartPi1Test() {
      pi1_test_thread = std::thread([this]() {
        LOG(INFO) << "Started pi1_test";
        pi1_test_event_loop->Run();
      });
    }
  
    void StopPi1Test() {
      pi1_test_event_loop->Exit();
      pi1_test_thread.join();
    }
  
    void MakePi2Server() {
      OnPi2();
      FLAGS_application_name = "pi2_message_bridge_server";
      pi2_server_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
      pi2_server_event_loop->SetRuntimeRealtimePriority(1);
      pi2_message_bridge_server =
          std::make_unique<MessageBridgeServer>(pi2_server_event_loop.get());
    }
  
    void RunPi2Server(chrono::nanoseconds duration) {
      // Setup a shutdown callback.
      aos::TimerHandler *const quit = pi2_server_event_loop->AddTimer(
          [this]() { pi2_server_event_loop->Exit(); });
      pi2_server_event_loop->OnRun([this, quit, duration]() {
        // Stop between timestamps, not exactly on them.
        quit->Setup(pi2_server_event_loop->monotonic_now() + duration);
      });
  
      pi2_server_event_loop->Run();
    }
  
    void StartPi2Server() {
      pi2_server_thread = std::thread([this]() {
        LOG(INFO) << "Started pi2_message_bridge_server";
        pi2_server_event_loop->Run();
      });
    }
  
    void StopPi2Server() {
      if (pi2_server_thread.joinable()) {
        pi2_server_event_loop->Exit();
        pi2_server_thread.join();
        pi2_server_thread = std::thread();
      }
      pi2_message_bridge_server.reset();
      pi2_server_event_loop.reset();
    }
  
    void MakePi2Client() {
      OnPi2();
      FLAGS_application_name = "pi2_message_bridge_client";
      pi2_client_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
      pi2_client_event_loop->SetRuntimeRealtimePriority(1);
      pi2_message_bridge_client =
          std::make_unique<MessageBridgeClient>(pi2_client_event_loop.get());
    }
  
    void RunPi2Client(chrono::nanoseconds duration) {
      // Run for 5 seconds to make sure we have time to estimate the offset.
      aos::TimerHandler *const quit = pi2_client_event_loop->AddTimer(
          [this]() { pi2_client_event_loop->Exit(); });
      pi2_client_event_loop->OnRun([this, quit, duration]() {
        // Stop between timestamps, not exactly on them.
        quit->Setup(pi2_client_event_loop->monotonic_now() + duration);
      });
  
      // And go!
      pi2_client_event_loop->Run();
    }
  
    void StartPi2Client() {
      pi2_client_thread = std::thread([this]() {
        LOG(INFO) << "Started pi2_message_bridge_client";
        pi2_client_event_loop->Run();
      });
    }
  
    void StopPi2Client() {
      if (pi2_client_thread.joinable()) {
        pi2_client_event_loop->Exit();
        pi2_client_thread.join();
        pi2_client_thread = std::thread();
      }
      pi2_message_bridge_client.reset();
      pi2_client_event_loop.reset();
    }
  
    void MakePi2Test() {
      OnPi2();
      FLAGS_application_name = "test2";
      pi2_test_event_loop =
          std::make_unique<aos::ShmEventLoop>(&config.message());
  
      pi2_test_event_loop->MakeWatcher(
          "/pi2/aos", [](const ServerStatistics &stats) {
            VLOG(1) << "/pi2/aos ServerStatistics " << FlatbufferToJson(&stats);
          });
  
      pi2_test_event_loop->MakeWatcher(
          "/pi2/aos", [](const ClientStatistics &stats) {
            VLOG(1) << "/pi2/aos ClientStatistics " << FlatbufferToJson(&stats);
          });
  
      pi2_test_event_loop->MakeWatcher(
          "/pi1/aos", [this](const Timestamp &timestamp) {
            VLOG(1) << "/pi1/aos Timestamp " << FlatbufferToJson(&timestamp);
            EXPECT_EQ(pi2_test_event_loop->context().source_boot_uuid,
                      pi1_boot_uuid_);
          });
      pi2_test_event_loop->MakeWatcher(
          "/pi2/aos", [](const Timestamp &timestamp) {
            VLOG(1) << "/pi2/aos Timestamp " << FlatbufferToJson(&timestamp);
          });
    }
  
    void StartPi2Test() {
      pi2_test_thread = std::thread([this]() {
        LOG(INFO) << "Started pi2_message_bridge_test";
        pi2_test_event_loop->Run();
      });
    }
  
    void StopPi2Test() {
      pi2_test_event_loop->Exit();
      pi2_test_thread.join();
    }
  
    aos::FlatbufferDetachedBuffer<aos::Configuration> config;
    const UUID pi1_boot_uuid_;
    const UUID pi2_boot_uuid_;
  
    std::unique_ptr<aos::ShmEventLoop> pi1_server_event_loop;
    std::unique_ptr<MessageBridgeServer> pi1_message_bridge_server;
    std::thread pi1_server_thread;
  
    std::unique_ptr<aos::ShmEventLoop> pi1_client_event_loop;
    std::unique_ptr<MessageBridgeClient> pi1_message_bridge_client;
    std::thread pi1_client_thread;
  
    std::unique_ptr<aos::ShmEventLoop> pi1_test_event_loop;
    std::thread pi1_test_thread;
  
    std::unique_ptr<aos::ShmEventLoop> pi2_server_event_loop;
    std::unique_ptr<MessageBridgeServer> pi2_message_bridge_server;
    std::thread pi2_server_thread;
  
    std::unique_ptr<aos::ShmEventLoop> pi2_client_event_loop;
    std::unique_ptr<MessageBridgeClient> pi2_message_bridge_client;
    std::thread pi2_client_thread;
  
    std::unique_ptr<aos::ShmEventLoop> pi2_test_event_loop;
    std::thread pi2_test_thread;
  };
  
  // Test that we can send a ping message over sctp and receive it.
  TEST_P(MessageBridgeParameterizedTest, PingPong) {
    // This is rather annoying to set up.  We need to start up a client and
    // server, on the same node, but get them to think that they are on different
    // nodes.
    //
    // We then get to wait until they are connected.
    //
    // After they are connected, we send a Ping message.
    //
    // On the other end, we receive a Pong message.
    //
    // But, we need the client to not post directly to "/test" like it would in a
    // real system, otherwise we will re-send the ping message... So, use an
    // application specific map to have the client post somewhere else.
    //
    // To top this all off, each of these needs to be done with a ShmEventLoop,
    // which needs to run in a separate thread...  And it is really hard to get
    // everything started up reliably.  So just be super generous on timeouts and
    // hope for the best.  We can be more generous in the future if we need to.
    //
    // We are faking the application names by passing in --application_name=foo
    OnPi1();
    // Force ourselves to be "raspberrypi" and allocate everything.
  
    MakePi1Server();
    MakePi1Client();
  
    // And build the app which sends the pings.
    FLAGS_application_name = "ping";
    aos::ShmEventLoop ping_event_loop(&config.message());
    aos::Sender<examples::Ping> ping_sender =
        ping_event_loop.MakeSender<examples::Ping>("/test");
  
    aos::ShmEventLoop pi1_test_event_loop(&config.message());
    aos::Fetcher<RemoteMessage> message_header_fetcher1 =
        pi1_test_event_loop.MakeFetcher<RemoteMessage>(
            shared() ? "/pi1/aos/remote_timestamps/pi2"
                     : "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping");
  
    // Fetchers for confirming the remote timestamps made it.
    aos::Fetcher<examples::Ping> ping_on_pi1_fetcher =
        ping_event_loop.MakeFetcher<examples::Ping>("/test");
    aos::Fetcher<Timestamp> pi1_on_pi1_timestamp_fetcher =
        ping_event_loop.MakeFetcher<Timestamp>("/aos");
  
    // Now do it for "raspberrypi2", the client.
    OnPi2();
  
    MakePi2Client();
    MakePi2Server();
  
    // And build the app which sends the pongs.
    FLAGS_application_name = "pong";
    aos::ShmEventLoop pong_event_loop(&config.message());
  
    // And build the app for testing.
    FLAGS_application_name = "test";
    aos::ShmEventLoop test_event_loop(&config.message());
  
    aos::Fetcher<ClientStatistics> client_statistics_fetcher =
        test_event_loop.MakeFetcher<ClientStatistics>("/aos");
    aos::Fetcher<RemoteMessage> message_header_fetcher2 =
        test_event_loop.MakeFetcher<RemoteMessage>(
            shared() ? "/pi2/aos/remote_timestamps/pi1"
                     : "/pi2/aos/remote_timestamps/pi1/pi2/aos/"
                       "aos-message_bridge-Timestamp");
  
    // Event loop for fetching data delivered to pi2 from pi1 to match up
    // messages.
    aos::ShmEventLoop delivered_messages_event_loop(&config.message());
    aos::Fetcher<Timestamp> pi1_on_pi2_timestamp_fetcher =
        delivered_messages_event_loop.MakeFetcher<Timestamp>("/pi1/aos");
    aos::Fetcher<examples::Ping> ping_on_pi2_fetcher =
        delivered_messages_event_loop.MakeFetcher<examples::Ping>("/test");
    EXPECT_FALSE(ping_on_pi2_fetcher.Fetch());
    EXPECT_FALSE(pi1_on_pi2_timestamp_fetcher.Fetch());
  
    // Count the pongs.
    int pong_count = 0;
    pong_event_loop.MakeWatcher("/test", [&pong_count, &pong_event_loop,
                                          this](const examples::Ping &ping) {
      EXPECT_EQ(pong_event_loop.context().source_boot_uuid, pi1_boot_uuid_);
      ++pong_count;
      VLOG(1) << "Got ping back " << FlatbufferToJson(&ping);
    });
  
    FLAGS_override_hostname = "";
  
    // Wait until we are connected, then send.
    int ping_count = 0;
    int pi1_server_statistics_count = 0;
    ping_event_loop.MakeWatcher("/pi1/aos", [this, &ping_count, &ping_sender,
                                             &pi1_server_statistics_count](
                                                const ServerStatistics &stats) {
      VLOG(1) << "/pi1/aos ServerStatistics " << FlatbufferToJson(&stats);
  
      ASSERT_TRUE(stats.has_connections());
      EXPECT_EQ(stats.connections()->size(), 1);
  
      bool connected = false;
      for (const ServerConnection *connection : *stats.connections()) {
        // Confirm that we are estimating the server time offset correctly. It
        // should be about 0 since we are on the same machine here.
        if (connection->has_monotonic_offset()) {
          EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
                    chrono::milliseconds(1));
          EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
                    chrono::milliseconds(-1));
          ++pi1_server_statistics_count;
        }
  
        if (connection->node()->name()->string_view() ==
            pi2_client_event_loop->node()->name()->string_view()) {
          if (connection->state() == State::CONNECTED) {
            EXPECT_TRUE(connection->has_boot_uuid());
            EXPECT_EQ(connection->connection_count(), 1u);
            EXPECT_LT(monotonic_clock::time_point(chrono::nanoseconds(
                          connection->connected_since_time())),
                      monotonic_clock::now());
            connected = true;
          } else {
            EXPECT_FALSE(connection->has_connection_count());
            EXPECT_FALSE(connection->has_connected_since_time());
          }
        }
      }
  
      if (connected) {
        VLOG(1) << "Connected!  Sent ping.";
        auto builder = ping_sender.MakeBuilder();
        examples::Ping::Builder ping_builder =
            builder.MakeBuilder<examples::Ping>();
        ping_builder.add_value(ping_count + 971);
        EXPECT_EQ(builder.Send(ping_builder.Finish()),
                  RawSender::Error::kOk);
        ++ping_count;
      }
    });
  
    // Confirm both client and server statistics messages have decent offsets in
    // them.
    int pi2_server_statistics_count = 0;
    pong_event_loop.MakeWatcher("/pi2/aos", [&pi2_server_statistics_count](
                                                const ServerStatistics &stats) {
      VLOG(1) << "/pi2/aos ServerStatistics " << FlatbufferToJson(&stats);
      for (const ServerConnection *connection : *stats.connections()) {
        if (connection->has_monotonic_offset()) {
          ++pi2_server_statistics_count;
          // Confirm that we are estimating the server time offset correctly. It
          // should be about 0 since we are on the same machine here.
          EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
                    chrono::milliseconds(1));
          EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
                    chrono::milliseconds(-1));
          EXPECT_TRUE(connection->has_boot_uuid());
        }
  
        if (connection->state() == State::CONNECTED) {
          EXPECT_EQ(connection->connection_count(), 1u);
          EXPECT_LT(monotonic_clock::time_point(
                        chrono::nanoseconds(connection->connected_since_time())),
                    monotonic_clock::now());
        } else {
          EXPECT_FALSE(connection->has_connection_count());
          EXPECT_FALSE(connection->has_connected_since_time());
        }
      }
    });
  
    int pi1_client_statistics_count = 0;
    int pi1_connected_client_statistics_count = 0;
    ping_event_loop.MakeWatcher(
        "/pi1/aos",
        [&pi1_client_statistics_count,
         &pi1_connected_client_statistics_count](const ClientStatistics &stats) {
          VLOG(1) << "/pi1/aos ClientStatistics " << FlatbufferToJson(&stats);
  
          for (const ClientConnection *connection : *stats.connections()) {
            if (connection->has_monotonic_offset()) {
              ++pi1_client_statistics_count;
              // It takes at least 10 microseconds to send a message between the
              // client and server.  The min (filtered) time shouldn't be over 10
              // milliseconds on localhost.  This might have to bump up if this is
              // proving flaky.
              EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
                        chrono::milliseconds(10))
                  << " " << connection->monotonic_offset()
                  << "ns vs 10000ns on iteration " << pi1_client_statistics_count;
              EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
                        chrono::microseconds(10))
                  << " " << connection->monotonic_offset()
                  << "ns vs 10000ns on iteration " << pi1_client_statistics_count;
            }
            if (connection->state() == State::CONNECTED) {
              EXPECT_EQ(connection->connection_count(), 1u);
              EXPECT_LT(monotonic_clock::time_point(chrono::nanoseconds(
                            connection->connected_since_time())),
                        monotonic_clock::now());
              // The first Connected message may not have a UUID in it since no
              // data has flown.  That's fine.
              if (pi1_connected_client_statistics_count > 0) {
                EXPECT_TRUE(connection->has_boot_uuid())
                    << ": " << aos::FlatbufferToJson(connection);
              }
              ++pi1_connected_client_statistics_count;
            } else {
              EXPECT_FALSE(connection->has_connection_count());
              EXPECT_FALSE(connection->has_connected_since_time());
            }
          }
        });
  
    int pi2_client_statistics_count = 0;
    int pi2_connected_client_statistics_count = 0;
    pong_event_loop.MakeWatcher(
        "/pi2/aos",
        [&pi2_client_statistics_count,
         &pi2_connected_client_statistics_count](const ClientStatistics &stats) {
          VLOG(1) << "/pi2/aos ClientStatistics " << FlatbufferToJson(&stats);
  
          for (const ClientConnection *connection : *stats.connections()) {
            if (connection->has_monotonic_offset()) {
              ++pi2_client_statistics_count;
              EXPECT_LT(chrono::nanoseconds(connection->monotonic_offset()),
                        chrono::milliseconds(10))
                  << ": got " << aos::FlatbufferToJson(connection);
              EXPECT_GT(chrono::nanoseconds(connection->monotonic_offset()),
                        chrono::microseconds(10))
                  << ": got " << aos::FlatbufferToJson(connection);
            }
            if (connection->state() == State::CONNECTED) {
              EXPECT_EQ(connection->connection_count(), 1u);
              EXPECT_LT(monotonic_clock::time_point(chrono::nanoseconds(
                            connection->connected_since_time())),
                        monotonic_clock::now());
              if (pi2_connected_client_statistics_count > 0) {
                EXPECT_TRUE(connection->has_boot_uuid());
              }
              ++pi2_connected_client_statistics_count;
            } else {
              EXPECT_FALSE(connection->has_connection_count());
              EXPECT_FALSE(connection->has_connected_since_time());
            }
          }
        });
  
    ping_event_loop.MakeWatcher("/pi1/aos", [](const Timestamp &timestamp) {
      EXPECT_TRUE(timestamp.has_offsets());
      VLOG(1) << "/pi1/aos Timestamp " << FlatbufferToJson(&timestamp);
    });
    pong_event_loop.MakeWatcher("/pi2/aos", [](const Timestamp &timestamp) {
      EXPECT_TRUE(timestamp.has_offsets());
      VLOG(1) << "/pi2/aos Timestamp " << FlatbufferToJson(&timestamp);
    });
  
    // Run for 5 seconds to make sure we have time to estimate the offset.
    aos::TimerHandler *quit = ping_event_loop.AddTimer(
        [&ping_event_loop]() { ping_event_loop.Exit(); });
    ping_event_loop.OnRun([quit, &ping_event_loop]() {
      // Stop between timestamps, not exactly on them.
      quit->Setup(ping_event_loop.monotonic_now() + chrono::milliseconds(5050));
    });
  
    // Find the channel index for both the /pi1/aos Timestamp channel and Ping
    // channel.
    const size_t pi1_timestamp_channel = configuration::ChannelIndex(
        pong_event_loop.configuration(), pi1_on_pi2_timestamp_fetcher.channel());
    const size_t ping_timestamp_channel =
        configuration::ChannelIndex(delivered_messages_event_loop.configuration(),
                                    ping_on_pi2_fetcher.channel());
  
    for (const Channel *channel : *ping_event_loop.configuration()->channels()) {
      VLOG(1) << "Channel "
              << configuration::ChannelIndex(ping_event_loop.configuration(),
                                             channel)
              << " " << configuration::CleanedChannelToString(channel);
    }
  
    // For each remote timestamp we get back, confirm that it is either a ping
    // message, or a timestamp we sent out.  Also confirm that the timestamps are
    // correct.
    for (std::pair<int, std::string> channel :
         shared()
             ? std::vector<std::pair<
                   int, std::string>>{{-1, "/pi1/aos/remote_timestamps/pi2"}}
             : std::vector<std::pair<int, std::string>>{
                   {pi1_timestamp_channel,
                    "/pi1/aos/remote_timestamps/pi2/pi1/aos/"
                    "aos-message_bridge-Timestamp"},
                   {ping_timestamp_channel,
                    "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping"}}) {
      ping_event_loop.MakeWatcher(
          channel.second,
          [pi1_timestamp_channel, ping_timestamp_channel, &ping_on_pi2_fetcher,
           &ping_on_pi1_fetcher, &pi1_on_pi2_timestamp_fetcher,
           &pi1_on_pi1_timestamp_fetcher,
           channel_index = channel.first](const RemoteMessage &header) {
            VLOG(1) << "/pi1/aos/remote_timestamps/pi2 RemoteMessage "
                    << aos::FlatbufferToJson(&header);
  
            EXPECT_TRUE(header.has_boot_uuid());
            if (channel_index != -1) {
              ASSERT_EQ(channel_index, header.channel_index());
            }
  
            const aos::monotonic_clock::time_point header_monotonic_sent_time(
                chrono::nanoseconds(header.monotonic_sent_time()));
            const aos::realtime_clock::time_point header_realtime_sent_time(
                chrono::nanoseconds(header.realtime_sent_time()));
            const aos::monotonic_clock::time_point header_monotonic_remote_time(
                chrono::nanoseconds(header.monotonic_remote_time()));
            const aos::realtime_clock::time_point header_realtime_remote_time(
                chrono::nanoseconds(header.realtime_remote_time()));
  
            const Context *pi1_context = nullptr;
            const Context *pi2_context = nullptr;
  
            if (header.channel_index() == pi1_timestamp_channel) {
              // Find the forwarded message.
              while (pi1_on_pi2_timestamp_fetcher.context().monotonic_event_time <
                     header_monotonic_sent_time) {
                ASSERT_TRUE(pi1_on_pi2_timestamp_fetcher.FetchNext());
              }
  
              // And the source message.
              while (pi1_on_pi1_timestamp_fetcher.context().monotonic_event_time <
                     header_monotonic_remote_time) {
                ASSERT_TRUE(pi1_on_pi1_timestamp_fetcher.FetchNext());
              }
  
              pi1_context = &pi1_on_pi1_timestamp_fetcher.context();
              pi2_context = &pi1_on_pi2_timestamp_fetcher.context();
            } else if (header.channel_index() == ping_timestamp_channel) {
              // Find the forwarded message.
              while (ping_on_pi2_fetcher.context().monotonic_event_time <
                     header_monotonic_sent_time) {
                ASSERT_TRUE(ping_on_pi2_fetcher.FetchNext());
              }
  
              // And the source message.
              while (ping_on_pi1_fetcher.context().monotonic_event_time <
                     header_monotonic_remote_time) {
                ASSERT_TRUE(ping_on_pi1_fetcher.FetchNext());
              }
  
              pi1_context = &ping_on_pi1_fetcher.context();
              pi2_context = &ping_on_pi2_fetcher.context();
            } else {
              LOG(FATAL) << "Unknown channel";
            }
  
            // Confirm the forwarded message has matching timestamps to the
            // timestamps we got back.
            EXPECT_EQ(pi2_context->queue_index, header.queue_index());
            EXPECT_EQ(pi2_context->monotonic_event_time,
                      header_monotonic_sent_time);
            EXPECT_EQ(pi2_context->realtime_event_time,
                      header_realtime_sent_time);
            EXPECT_EQ(pi2_context->realtime_remote_time,
                      header_realtime_remote_time);
            EXPECT_EQ(pi2_context->monotonic_remote_time,
                      header_monotonic_remote_time);
  
            // Confirm the forwarded message also matches the source message.
            EXPECT_EQ(pi1_context->queue_index, header.queue_index());
            EXPECT_EQ(pi1_context->monotonic_event_time,
                      header_monotonic_remote_time);
            EXPECT_EQ(pi1_context->realtime_event_time,
                      header_realtime_remote_time);
          });
    }
  
    // Start everything up.  Pong is the only thing we don't know how to wait
    // on, so start it first.
    std::thread pong_thread([&pong_event_loop]() { pong_event_loop.Run(); });
  
    StartPi1Server();
    StartPi1Client();
    StartPi2Client();
    StartPi2Server();
  
    // And go!
    ping_event_loop.Run();
  
    // Shut everyone else down
    StopPi1Server();
    StopPi1Client();
    StopPi2Client();
    StopPi2Server();
    pong_event_loop.Exit();
    pong_thread.join();
  
    // Make sure we sent something.
    EXPECT_GE(ping_count, 1);
    // And got something back.
    EXPECT_GE(pong_count, 1);
  
    // Confirm that we are estimating a monotonic offset on the client.
    ASSERT_TRUE(client_statistics_fetcher.Fetch());
  
    EXPECT_EQ(client_statistics_fetcher->connections()->size(), 1u);
    EXPECT_EQ(client_statistics_fetcher->connections()
                  ->Get(0)
                  ->node()
                  ->name()
                  ->string_view(),
              "pi1");
  
    // Make sure the offset in one direction is less than a second.
    EXPECT_GT(
        client_statistics_fetcher->connections()->Get(0)->monotonic_offset(), 0)
        << aos::FlatbufferToJson(client_statistics_fetcher.get());
    EXPECT_LT(
        client_statistics_fetcher->connections()->Get(0)->monotonic_offset(),
        1000000000)
        << aos::FlatbufferToJson(client_statistics_fetcher.get());
  
    EXPECT_GE(pi1_server_statistics_count, 2);
    EXPECT_GE(pi2_server_statistics_count, 2);
    EXPECT_GE(pi1_client_statistics_count, 2);
    EXPECT_GE(pi2_client_statistics_count, 2);
  
    // Confirm we got timestamps back!
    EXPECT_TRUE(message_header_fetcher1.Fetch());
    EXPECT_TRUE(message_header_fetcher2.Fetch());
  }
  
  // Test that the client disconnecting triggers the server offsets on both sides
  // to clear.
  TEST_P(MessageBridgeParameterizedTest, ClientRestart) {
    // This is rather annoying to set up.  We need to start up a client and
    // server, on the same node, but get them to think that they are on different
    // nodes.
    //
    // We need the client to not post directly to "/test" like it would in a
    // real system, otherwise we will re-send the ping message... So, use an
    // application specific map to have the client post somewhere else.
    //
    // To top this all off, each of these needs to be done with a ShmEventLoop,
    // which needs to run in a separate thread...  And it is really hard to get
    // everything started up reliably.  So just be super generous on timeouts and
    // hope for the best.  We can be more generous in the future if we need to.
    //
    // We are faking the application names by passing in --application_name=foo
    OnPi1();
  
    MakePi1Server();
    MakePi1Client();
  
    // And build the app for testing.
    MakePi1Test();
    aos::Fetcher<ServerStatistics> pi1_server_statistics_fetcher =
        pi1_test_event_loop->MakeFetcher<ServerStatistics>("/pi1/aos");
  
    // Now do it for "raspberrypi2", the client.
    OnPi2();
    MakePi2Server();
  
    // And build the app for testing.
    MakePi2Test();
    aos::Fetcher<ServerStatistics> pi2_server_statistics_fetcher =
        pi2_test_event_loop->MakeFetcher<ServerStatistics>("/pi2/aos");
  
    // Wait until we are connected, then send.
  
    StartPi1Test();
    StartPi2Test();
    StartPi1Server();
    StartPi1Client();
    StartPi2Server();
  
    {
      MakePi2Client();
  
      RunPi2Client(chrono::milliseconds(3050));
  
      // Now confirm we are synchronized.
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
  
      const ServerConnection *const pi1_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ServerConnection *const pi2_connection =
          pi2_server_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_connection->state(), State::CONNECTED);
      EXPECT_EQ(pi1_connection->connection_count(), 1u);
      EXPECT_TRUE(pi1_connection->has_connected_since_time());
      EXPECT_TRUE(pi1_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi1_connection->has_boot_uuid());
  
      EXPECT_EQ(pi2_connection->state(), State::CONNECTED);
      EXPECT_EQ(pi2_connection->connection_count(), 1u);
      EXPECT_TRUE(pi2_connection->has_connected_since_time());
      EXPECT_TRUE(pi2_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi2_connection->has_boot_uuid());
  
      StopPi2Client();
    }
  
    std::this_thread::sleep_for(SctpClientConnection::kReconnectTimeout +
                                std::chrono::seconds(1));
  
    {
      // Now confirm we are un-synchronized.
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
      const ServerConnection *const pi1_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ServerConnection *const pi2_connection =
          pi2_server_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_connection->state(), State::DISCONNECTED);
      EXPECT_EQ(pi1_connection->connection_count(), 1u);
      EXPECT_FALSE(pi1_connection->has_connected_since_time());
      EXPECT_FALSE(pi1_connection->has_monotonic_offset());
      EXPECT_FALSE(pi1_connection->has_boot_uuid());
      EXPECT_EQ(pi2_connection->state(), State::CONNECTED);
      EXPECT_FALSE(pi2_connection->has_monotonic_offset());
      EXPECT_TRUE(pi2_connection->has_boot_uuid());
      EXPECT_EQ(pi2_connection->connection_count(), 1u);
      EXPECT_TRUE(pi2_connection->has_connected_since_time());
    }
  
    {
      MakePi2Client();
      // And go!
      RunPi2Client(chrono::milliseconds(3050));
  
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
  
      // Now confirm we are synchronized again.
      const ServerConnection *const pi1_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ServerConnection *const pi2_connection =
          pi2_server_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_connection->state(), State::CONNECTED);
      EXPECT_EQ(pi1_connection->connection_count(), 2u);
      EXPECT_TRUE(pi1_connection->has_connected_since_time());
      EXPECT_TRUE(pi1_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(1))
          << ": " << FlatbufferToJson(pi1_connection);
      EXPECT_GT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(-1))
          << ": " << FlatbufferToJson(pi1_connection);
      EXPECT_TRUE(pi1_connection->has_boot_uuid());
  
      EXPECT_EQ(pi2_connection->state(), State::CONNECTED);
      EXPECT_EQ(pi2_connection->connection_count(), 1u);
      EXPECT_TRUE(pi2_connection->has_connected_since_time());
      EXPECT_TRUE(pi2_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(1))
          << ": " << FlatbufferToJson(pi2_connection);
      EXPECT_GT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(-1))
          << ": " << FlatbufferToJson(pi2_connection);
      EXPECT_TRUE(pi2_connection->has_boot_uuid());
  
      StopPi2Client();
    }
  
    // Shut everyone else down
    StopPi1Server();
    StopPi1Client();
    StopPi2Server();
    StopPi1Test();
    StopPi2Test();
  }
  
  // Test that the server disconnecting triggers the server offsets on the other
  // side to clear, along with the other client.
  TEST_P(MessageBridgeParameterizedTest, ServerRestart) {
    // This is rather annoying to set up.  We need to start up a client and
    // server, on the same node, but get them to think that they are on different
    // nodes.
    //
    // We need the client to not post directly to "/test" like it would in a
    // real system, otherwise we will re-send the ping message... So, use an
    // application specific map to have the client post somewhere else.
    //
    // To top this all off, each of these needs to be done with a ShmEventLoop,
    // which needs to run in a separate thread...  And it is really hard to get
    // everything started up reliably.  So just be super generous on timeouts and
    // hope for the best.  We can be more generous in the future if we need to.
    //
    // We are faking the application names by passing in --application_name=foo
    // Force ourselves to be "raspberrypi" and allocate everything.
    OnPi1();
    MakePi1Server();
    MakePi1Client();
  
    // And build the app for testing.
    MakePi1Test();
    aos::Fetcher<ServerStatistics> pi1_server_statistics_fetcher =
        pi1_test_event_loop->MakeFetcher<ServerStatistics>("/pi1/aos");
    aos::Fetcher<ClientStatistics> pi1_client_statistics_fetcher =
        pi1_test_event_loop->MakeFetcher<ClientStatistics>("/pi1/aos");
  
    // Now do it for "raspberrypi2", the client.
    OnPi2();
    MakePi2Client();
  
    // And build the app for testing.
    MakePi2Test();
    aos::Fetcher<ServerStatistics> pi2_server_statistics_fetcher =
        pi2_test_event_loop->MakeFetcher<ServerStatistics>("/pi2/aos");
  
    // Start everything up.  Pong is the only thing we don't know how to wait on,
    // so start it first.
    StartPi1Test();
    StartPi2Test();
    StartPi1Server();
    StartPi1Client();
    StartPi2Client();
  
    // Confirm both client and server statistics messages have decent offsets in
    // them.
  
    {
      MakePi2Server();
  
      RunPi2Server(chrono::milliseconds(3050));
  
      // Now confirm we are synchronized.
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
  
      const ServerConnection *const pi1_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ServerConnection *const pi2_connection =
          pi2_server_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_connection->state(), State::CONNECTED);
      EXPECT_TRUE(pi1_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi1_connection->has_boot_uuid());
      EXPECT_TRUE(pi1_connection->has_connected_since_time());
      EXPECT_EQ(pi1_connection->connection_count(), 1u);
  
      EXPECT_EQ(pi2_connection->state(), State::CONNECTED);
      EXPECT_TRUE(pi2_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi2_connection->has_boot_uuid());
      EXPECT_TRUE(pi2_connection->has_connected_since_time());
      EXPECT_EQ(pi2_connection->connection_count(), 1u);
  
      StopPi2Server();
    }
  
    std::this_thread::sleep_for(std::chrono::seconds(2));
  
    {
      // And confirm we are unsynchronized.
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi1_client_statistics_fetcher.Fetch());
  
      const ServerConnection *const pi1_server_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ClientConnection *const pi1_client_connection =
          pi1_client_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_server_connection->state(), State::CONNECTED);
      EXPECT_FALSE(pi1_server_connection->has_monotonic_offset());
      EXPECT_TRUE(pi1_server_connection->has_connected_since_time());
      EXPECT_EQ(pi1_server_connection->connection_count(), 1u);
  
      EXPECT_TRUE(pi1_server_connection->has_boot_uuid());
      EXPECT_EQ(pi1_client_connection->state(), State::DISCONNECTED);
      EXPECT_FALSE(pi1_client_connection->has_monotonic_offset());
      EXPECT_FALSE(pi1_client_connection->has_connected_since_time());
      EXPECT_EQ(pi1_client_connection->connection_count(), 1u);
      EXPECT_FALSE(pi1_client_connection->has_boot_uuid());
    }
  
    {
      MakePi2Server();
  
      RunPi2Server(chrono::milliseconds(3050));
  
      // And confirm we are synchronized again.
      EXPECT_TRUE(pi1_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi2_server_statistics_fetcher.Fetch());
      EXPECT_TRUE(pi1_client_statistics_fetcher.Fetch());
  
      const ServerConnection *const pi1_connection =
          pi1_server_statistics_fetcher->connections()->Get(0);
      const ServerConnection *const pi2_connection =
          pi2_server_statistics_fetcher->connections()->Get(0);
      const ClientConnection *const pi1_client_connection =
          pi1_client_statistics_fetcher->connections()->Get(0);
  
      EXPECT_EQ(pi1_connection->state(), State::CONNECTED);
      EXPECT_TRUE(pi1_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi1_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi1_connection->has_boot_uuid());
  
      EXPECT_EQ(pi1_client_connection->state(), State::CONNECTED);
      EXPECT_TRUE(pi1_client_connection->has_connected_since_time());
      EXPECT_EQ(pi1_client_connection->connection_count(), 2u);
      EXPECT_TRUE(pi1_client_connection->has_boot_uuid());
  
      EXPECT_EQ(pi2_connection->state(), State::CONNECTED);
      EXPECT_TRUE(pi2_connection->has_monotonic_offset());
      EXPECT_LT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(1));
      EXPECT_GT(chrono::nanoseconds(pi2_connection->monotonic_offset()),
                chrono::milliseconds(-1));
      EXPECT_TRUE(pi2_connection->has_boot_uuid());
  
      StopPi2Server();
    }
  
    // Shut everyone else down
    StopPi1Server();
    StopPi1Client();
    StopPi2Client();
    StopPi1Test();
    StopPi2Test();
  }
  
  // TODO(austin): The above test confirms that the external state does the right
  // thing, but doesn't confirm that the internal state does.  We either need to
  // expose a way to check the state in a thread-safe way, or need a way to jump
  // time for one node to do that.
  
  void SendPing(aos::Sender<examples::Ping> *sender, int value) {
    aos::Sender<examples::Ping>::Builder builder = sender->MakeBuilder();
    examples::Ping::Builder ping_builder = builder.MakeBuilder<examples::Ping>();
    ping_builder.add_value(value);
    builder.CheckOk(builder.Send(ping_builder.Finish()));
  }
  
  // Tests that when a message is sent before the bridge starts up, but is
  // configured as reliable, we forward it.  Confirm this survives a client reset.
  TEST_P(MessageBridgeParameterizedTest, ReliableSentBeforeClientStartup) {
    OnPi1();
  
    FLAGS_application_name = "sender";
    aos::ShmEventLoop send_event_loop(&config.message());
    aos::Sender<examples::Ping> ping_sender =
        send_event_loop.MakeSender<examples::Ping>("/test");
    SendPing(&ping_sender, 1);
    aos::Sender<examples::Ping> unreliable_ping_sender =
        send_event_loop.MakeSender<examples::Ping>("/unreliable");
    SendPing(&unreliable_ping_sender, 1);
  
    MakePi1Server();
    MakePi1Client();
  
    FLAGS_application_name = "pi1_timestamp";
    aos::ShmEventLoop pi1_remote_timestamp_event_loop(&config.message());
  
    // Now do it for "raspberrypi2", the client.
    OnPi2();
  
    MakePi2Server();
  
    aos::ShmEventLoop receive_event_loop(&config.message());
    aos::Fetcher<examples::Ping> ping_fetcher =
        receive_event_loop.MakeFetcher<examples::Ping>("/test");
    aos::Fetcher<examples::Ping> unreliable_ping_fetcher =
        receive_event_loop.MakeFetcher<examples::Ping>("/unreliable");
    aos::Fetcher<ClientStatistics> pi2_client_statistics_fetcher =
        receive_event_loop.MakeFetcher<ClientStatistics>("/pi2/aos");
  
    const size_t ping_channel_index = configuration::ChannelIndex(
        receive_event_loop.configuration(), ping_fetcher.channel());
  
    std::atomic<int> ping_timestamp_count{0};
    const std::string channel_name =
        shared() ? "/pi1/aos/remote_timestamps/pi2"
                 : "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping";
    pi1_remote_timestamp_event_loop.MakeWatcher(
        channel_name, [this, channel_name, ping_channel_index,
                       &ping_timestamp_count](const RemoteMessage &header) {
          VLOG(1) << channel_name << " RemoteMessage "
                  << aos::FlatbufferToJson(&header);
          EXPECT_TRUE(header.has_boot_uuid());
          if (shared() && header.channel_index() != ping_channel_index) {
            return;
          }
          CHECK_EQ(header.channel_index(), ping_channel_index);
          ++ping_timestamp_count;
        });
  
    // Before everything starts up, confirm there is no message.
    EXPECT_FALSE(ping_fetcher.Fetch());
    EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
  
    // Spin up the persistant pieces.
    StartPi1Server();
    StartPi1Client();
    StartPi2Server();
  
    // Event used to wait for the timestamp counting thread to start.
    aos::Event event;
    std::thread pi1_remote_timestamp_thread(
        [&pi1_remote_timestamp_event_loop, &event]() {
          pi1_remote_timestamp_event_loop.OnRun([&event]() { event.Set(); });
          pi1_remote_timestamp_event_loop.Run();
        });
  
    event.Wait();
  
    {
      // Now, spin up a client for 2 seconds.
      MakePi2Client();
  
      RunPi2Client(chrono::milliseconds(2050));
  
      // Confirm there is no detected duplicate packet.
      EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->duplicate_packets(),
                0u);
  
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->partial_deliveries(),
                0u);
  
      EXPECT_TRUE(ping_fetcher.Fetch());
      EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
      EXPECT_EQ(ping_timestamp_count, 1);
  
      StopPi2Client();
    }
  
    {
      // Now, spin up a client for 2 seconds.
      MakePi2Client();
  
      RunPi2Client(chrono::milliseconds(5050));
  
      // Confirm we detect the duplicate packet correctly.
      EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->duplicate_packets(),
                1u);
  
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->partial_deliveries(),
                0u);
  
      EXPECT_EQ(ping_timestamp_count, 1);
      EXPECT_FALSE(ping_fetcher.Fetch());
      EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
  
      StopPi2Client();
    }
  
    // Shut everyone else down
    StopPi1Client();
    StopPi2Server();
    pi1_remote_timestamp_event_loop.Exit();
    pi1_remote_timestamp_thread.join();
    StopPi1Server();
  }
  
  // Tests that when a message is sent before the bridge starts up, but is
  // configured as reliable, we forward it.  Confirm this works across server
  // resets.
  TEST_P(MessageBridgeParameterizedTest, ReliableSentBeforeServerStartup) {
    // Now do it for "raspberrypi2", the client.
    OnPi2();
  
    MakePi2Server();
    MakePi2Client();
  
    aos::ShmEventLoop receive_event_loop(&config.message());
    aos::Fetcher<examples::Ping> ping_fetcher =
        receive_event_loop.MakeFetcher<examples::Ping>("/test");
    aos::Fetcher<examples::Ping> unreliable_ping_fetcher =
        receive_event_loop.MakeFetcher<examples::Ping>("/unreliable");
    aos::Fetcher<ClientStatistics> pi2_client_statistics_fetcher =
        receive_event_loop.MakeFetcher<ClientStatistics>("/pi2/aos");
  
    // Force ourselves to be "raspberrypi" and allocate everything.
    OnPi1();
  
    FLAGS_application_name = "sender";
    aos::ShmEventLoop send_event_loop(&config.message());
    aos::Sender<examples::Ping> ping_sender =
        send_event_loop.MakeSender<examples::Ping>("/test");
    {
      aos::Sender<examples::Ping>::Builder builder = ping_sender.MakeBuilder();
      examples::Ping::Builder ping_builder =
          builder.MakeBuilder<examples::Ping>();
      ping_builder.add_value(1);
      builder.CheckOk(builder.Send(ping_builder.Finish()));
    }
  
    MakePi1Client();
  
    FLAGS_application_name = "pi1_timestamp";
    aos::ShmEventLoop pi1_remote_timestamp_event_loop(&config.message());
  
    const size_t ping_channel_index = configuration::ChannelIndex(
        receive_event_loop.configuration(), ping_fetcher.channel());
  
    std::atomic<int> ping_timestamp_count{0};
    const std::string channel_name =
        shared() ? "/pi1/aos/remote_timestamps/pi2"
                 : "/pi1/aos/remote_timestamps/pi2/test/aos-examples-Ping";
    pi1_remote_timestamp_event_loop.MakeWatcher(
        channel_name, [this, channel_name, ping_channel_index,
                       &ping_timestamp_count](const RemoteMessage &header) {
          VLOG(1) << channel_name << " RemoteMessage "
                  << aos::FlatbufferToJson(&header);
          EXPECT_TRUE(header.has_boot_uuid());
          if (shared() && header.channel_index() != ping_channel_index) {
            return;
          }
          CHECK_EQ(header.channel_index(), ping_channel_index);
          ++ping_timestamp_count;
        });
  
    // Before everything starts up, confirm there is no message.
    EXPECT_FALSE(ping_fetcher.Fetch());
    EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
  
    // Spin up the persistant pieces.
    StartPi1Client();
    StartPi2Server();
    StartPi2Client();
  
    // Event used to wait for the timestamp counting thread to start.
    aos::Event event;
    std::thread pi1_remote_timestamp_thread(
        [&pi1_remote_timestamp_event_loop, &event]() {
          pi1_remote_timestamp_event_loop.OnRun([&event]() { event.Set(); });
          pi1_remote_timestamp_event_loop.Run();
        });
  
    event.Wait();
  
    {
      // Now, spin up a server for 2 seconds.
      MakePi1Server();
  
      RunPi1Server(chrono::milliseconds(2050));
  
      // Confirm there is no detected duplicate packet.
      EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->duplicate_packets(),
                0u);
  
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->partial_deliveries(),
                0u);
  
      EXPECT_TRUE(ping_fetcher.Fetch());
      EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
      EXPECT_EQ(ping_timestamp_count, 1);
      LOG(INFO) << "Shutting down first pi1 MessageBridgeServer";
  
      StopPi1Server();
    }
  
    {
      // Now, spin up a second server for 2 seconds.
      MakePi1Server();
  
      RunPi1Server(chrono::milliseconds(2050));
  
      // Confirm we detect the duplicate packet correctly.
      EXPECT_TRUE(pi2_client_statistics_fetcher.Fetch());
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->duplicate_packets(),
                1u);
  
  
      EXPECT_EQ(pi2_client_statistics_fetcher->connections()
                    ->Get(0)
                    ->partial_deliveries(),
                0u);
  
      EXPECT_EQ(ping_timestamp_count, 1);
      EXPECT_FALSE(ping_fetcher.Fetch());
      EXPECT_FALSE(unreliable_ping_fetcher.Fetch());
  
      StopPi1Server();
    }
  
    // Shut everyone else down
    StopPi1Client();
    StopPi2Server();
    StopPi2Client();
    pi1_remote_timestamp_event_loop.Exit();
    pi1_remote_timestamp_thread.join();
  }
  
  INSTANTIATE_TEST_SUITE_P(
      MessageBridgeTests, MessageBridgeParameterizedTest,
      ::testing::Values(
          Param{"message_bridge_test_combined_timestamps_common_config.json",
                true},
          Param{"message_bridge_test_common_config.json", false}));
  
  }  // namespace testing
  }  // namespace message_bridge
  }  // namespace aos
  