[ntcore] Remove pImpl from implementation (#5480)

Also change Timestamped into a template.

ntcore/src/main/native/cpp/net/ClientImpl.cpp

@@ -10,7 +10,6 @@
 #include <variant>
 
 #include <fmt/format.h>
-#include <wpi/DenseMap.h>
 #include <wpi/Logger.h>
 #include <wpi/raw_ostream.h>
 #include <wpi/timestamp.h>
@@ -19,9 +18,7 @@
 #include "Log.h"
 #include "Message.h"
 #include "NetworkInterface.h"
-#include "PubSubOptions.h"
 #include "WireConnection.h"
-#include "WireDecoder.h"
 #include "WireEncoder.h"
 #include "networktables/NetworkTableValue.h"
 
@@ -34,79 +31,7 @@ static constexpr uint32_t kMinPeriodMs = 5;
 // transmission before we close the connection
 static constexpr uint32_t kWireMaxNotReadyUs = 1000000;
 
-namespace {
-
-struct PublisherData {
-  NT_Publisher handle;
-  PubSubOptionsImpl options;
-  // in options as double, but copy here as integer; rounded to the nearest
-  // 10 ms
-  uint32_t periodMs;
-  uint64_t nextSendMs{0};
-  std::vector<Value> outValues;  // outgoing values
-};
-
-class CImpl : public ServerMessageHandler {
- public:
-  CImpl(uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
-        std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
-            timeSyncUpdated,
-        std::function<void(uint32_t repeatMs)> setPeriodic);
-
-  void ProcessIncomingBinary(uint64_t curTimeMs, std::span<const uint8_t> data);
-  void HandleLocal(std::vector<ClientMessage>&& msgs);
-  bool SendControl(uint64_t curTimeMs);
-  void SendValues(uint64_t curTimeMs, bool flush);
-  void SendInitialValues();
-  bool CheckNetworkReady(uint64_t curTimeMs);
-
-  // ServerMessageHandler interface
-  void ServerAnnounce(std::string_view name, int64_t id,
-                      std::string_view typeStr, const wpi::json& properties,
-                      std::optional<int64_t> pubuid) final;
-  void ServerUnannounce(std::string_view name, int64_t id) final;
-  void ServerPropertiesUpdate(std::string_view name, const wpi::json& update,
-                              bool ack) final;
-
-  void Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
-               std::string_view name, std::string_view typeStr,
-               const wpi::json& properties, const PubSubOptionsImpl& options);
-  bool Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle);
-  void SetValue(NT_Publisher pubHandle, const Value& value);
-
-  int m_inst;
-  WireConnection& m_wire;
-  wpi::Logger& m_logger;
-  LocalInterface* m_local{nullptr};
-  std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
-      m_timeSyncUpdated;
-  std::function<void(uint32_t repeatMs)> m_setPeriodic;
-
-  // indexed by publisher index
-  std::vector<std::unique_ptr<PublisherData>> m_publishers;
-
-  // indexed by server-provided topic id
-  wpi::DenseMap<int64_t, NT_Topic> m_topicMap;
-
-  // timestamp handling
-  static constexpr uint32_t kPingIntervalMs = 3000;
-  uint64_t m_nextPingTimeMs{0};
-  uint64_t m_pongTimeMs{0};
-  uint32_t m_rtt2Us{UINT32_MAX};
-  bool m_haveTimeOffset{false};
-  int64_t m_serverTimeOffsetUs{0};
-
-  // periodic sweep handling
-  uint32_t m_periodMs{kPingIntervalMs + 10};
-  uint64_t m_lastSendMs{0};
-
-  // outgoing queue
-  std::vector<ClientMessage> m_outgoing;
-};
-
-}  // namespace
-
-CImpl::CImpl(
+ClientImpl::ClientImpl(
     uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
     std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
         timeSyncUpdated,
@@ -126,8 +51,8 @@ CImpl::CImpl(
   m_setPeriodic(m_periodMs);
 }
 
-void CImpl::ProcessIncomingBinary(uint64_t curTimeMs,
-                                  std::span<const uint8_t> data) {
+void ClientImpl::ProcessIncomingBinary(uint64_t curTimeMs,
+                                       std::span<const uint8_t> data) {
   for (;;) {
     if (data.empty()) {
       break;
@@ -138,7 +63,7 @@ void CImpl::ProcessIncomingBinary(uint64_t curTimeMs,
     Value value;
     std::string error;
     if (!WireDecodeBinary(&data, &id, &value, &error, -m_serverTimeOffsetUs)) {
-      ERROR("binary decode error: {}", error);
+      ERR("binary decode error: {}", error);
       break;  // FIXME
     }
     DEBUG4("BinaryMessage({})", id);
@@ -146,8 +71,8 @@ void CImpl::ProcessIncomingBinary(uint64_t curTimeMs,
     // handle RTT ping response
     if (id == -1) {
       if (!value.IsInteger()) {
-        WARNING("RTT ping response with non-integer type {}",
-                static_cast<int>(value.type()));
+        WARN("RTT ping response with non-integer type {}",
+             static_cast<int>(value.type()));
         continue;
       }
       DEBUG4("RTT ping response time {} value {}", value.time(),
@@ -168,7 +93,7 @@ void CImpl::ProcessIncomingBinary(uint64_t curTimeMs,
     // otherwise it's a value message, get the local topic handle for it
     auto topicIt = m_topicMap.find(id);
     if (topicIt == m_topicMap.end()) {
-      WARNING("received unknown id {}", id);
+      WARN("received unknown id {}", id);
       continue;
     }
 
@@ -179,7 +104,7 @@ void CImpl::ProcessIncomingBinary(uint64_t curTimeMs,
   }
 }
 
-void CImpl::HandleLocal(std::vector<ClientMessage>&& msgs) {
+void ClientImpl::HandleLocal(std::vector<ClientMessage>&& msgs) {
   DEBUG4("HandleLocal()");
   for (auto&& elem : msgs) {
     // common case is value
@@ -200,7 +125,7 @@ void CImpl::HandleLocal(std::vector<ClientMessage>&& msgs) {
   }
 }
 
-bool CImpl::SendControl(uint64_t curTimeMs) {
+bool ClientImpl::DoSendControl(uint64_t curTimeMs) {
   DEBUG4("SendControl({})", curTimeMs);
 
   // rate limit sends
@@ -246,7 +171,7 @@ bool CImpl::SendControl(uint64_t curTimeMs) {
   return true;
 }
 
-void CImpl::SendValues(uint64_t curTimeMs, bool flush) {
+void ClientImpl::DoSendValues(uint64_t curTimeMs, bool flush) {
   DEBUG4("SendValues({})", curTimeMs);
 
   // can't send value updates until we have a RTT
@@ -255,7 +180,7 @@ void CImpl::SendValues(uint64_t curTimeMs, bool flush) {
   }
 
   // ensure all control messages are sent ahead of value updates
-  if (!SendControl(curTimeMs)) {
+  if (!DoSendControl(curTimeMs)) {
     return;
   }
 
@@ -291,11 +216,11 @@ void CImpl::SendValues(uint64_t curTimeMs, bool flush) {
   }
 }
 
-void CImpl::SendInitialValues() {
+void ClientImpl::SendInitialValues() {
   DEBUG4("SendInitialValues()");
 
   // ensure all control messages are sent ahead of value updates
-  if (!SendControl(0)) {
+  if (!DoSendControl(0)) {
     return;
   }
 
@@ -321,7 +246,7 @@ void CImpl::SendInitialValues() {
   }
 }
 
-bool CImpl::CheckNetworkReady(uint64_t curTimeMs) {
+bool ClientImpl::CheckNetworkReady(uint64_t curTimeMs) {
   if (!m_wire.Ready()) {
     uint64_t lastFlushTime = m_wire.GetLastFlushTime();
     uint64_t now = wpi::Now();
@@ -333,10 +258,10 @@ bool CImpl::CheckNetworkReady(uint64_t curTimeMs) {
   return true;
 }
 
-void CImpl::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
-                    std::string_view name, std::string_view typeStr,
-                    const wpi::json& properties,
-                    const PubSubOptionsImpl& options) {
+void ClientImpl::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
+                         std::string_view name, std::string_view typeStr,
+                         const wpi::json& properties,
+                         const PubSubOptionsImpl& options) {
   unsigned int index = Handle{pubHandle}.GetIndex();
   if (index >= m_publishers.size()) {
     m_publishers.resize(index + 1);
@@ -360,7 +285,7 @@ void CImpl::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
   m_setPeriodic(m_periodMs);
 }
 
-bool CImpl::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) {
+bool ClientImpl::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) {
   unsigned int index = Handle{pubHandle}.GetIndex();
   if (index >= m_publishers.size()) {
     return false;
@@ -400,7 +325,7 @@ bool CImpl::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) {
   return doSend;
 }
 
-void CImpl::SetValue(NT_Publisher pubHandle, const Value& value) {
+void ClientImpl::SetValue(NT_Publisher pubHandle, const Value& value) {
   DEBUG4("SetValue({}, time={}, server_time={}, st_off={})", pubHandle,
          value.time(), value.server_time(), m_serverTimeOffsetUs);
   unsigned int index = Handle{pubHandle}.GetIndex();
@@ -415,10 +340,10 @@ void CImpl::SetValue(NT_Publisher pubHandle, const Value& value) {
   }
 }
 
-void CImpl::ServerAnnounce(std::string_view name, int64_t id,
-                           std::string_view typeStr,
-                           const wpi::json& properties,
-                           std::optional<int64_t> pubuid) {
+void ClientImpl::ServerAnnounce(std::string_view name, int64_t id,
+                                std::string_view typeStr,
+                                const wpi::json& properties,
+                                std::optional<int64_t> pubuid) {
   DEBUG4("ServerAnnounce({}, {}, {})", name, id, typeStr);
   assert(m_local);
   NT_Publisher pubHandle{0};
@@ -429,76 +354,38 @@ void CImpl::ServerAnnounce(std::string_view name, int64_t id,
       m_local->NetworkAnnounce(name, typeStr, properties, pubHandle);
 }
 
-void CImpl::ServerUnannounce(std::string_view name, int64_t id) {
+void ClientImpl::ServerUnannounce(std::string_view name, int64_t id) {
   DEBUG4("ServerUnannounce({}, {})", name, id);
   assert(m_local);
   m_local->NetworkUnannounce(name);
   m_topicMap.erase(id);
 }
 
-void CImpl::ServerPropertiesUpdate(std::string_view name,
-                                   const wpi::json& update, bool ack) {
+void ClientImpl::ServerPropertiesUpdate(std::string_view name,
+                                        const wpi::json& update, bool ack) {
   DEBUG4("ServerProperties({}, {}, {})", name, update.dump(), ack);
   assert(m_local);
   m_local->NetworkPropertiesUpdate(name, update, ack);
 }
 
-class ClientImpl::Impl final : public CImpl {
- public:
-  Impl(uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
-       std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
-           timeSyncUpdated,
-       std::function<void(uint32_t repeatMs)> setPeriodic)
-      : CImpl{curTimeMs,
-              inst,
-              wire,
-              logger,
-              std::move(timeSyncUpdated),
-              std::move(setPeriodic)} {}
-};
-
-ClientImpl::ClientImpl(
-    uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
-    std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
-        timeSyncUpdated,
-    std::function<void(uint32_t repeatMs)> setPeriodic)
-    : m_impl{std::make_unique<Impl>(curTimeMs, inst, wire, logger,
-                                    std::move(timeSyncUpdated),
-                                    std::move(setPeriodic))} {}
-
-ClientImpl::~ClientImpl() = default;
-
 void ClientImpl::ProcessIncomingText(std::string_view data) {
-  if (!m_impl->m_local) {
+  if (!m_local) {
     return;
   }
-  WireDecodeText(data, *m_impl, m_impl->m_logger);
-}
-
-void ClientImpl::ProcessIncomingBinary(uint64_t curTimeMs,
-                                       std::span<const uint8_t> data) {
-  m_impl->ProcessIncomingBinary(curTimeMs, data);
-}
-
-void ClientImpl::HandleLocal(std::vector<ClientMessage>&& msgs) {
-  m_impl->HandleLocal(std::move(msgs));
+  WireDecodeText(data, *this, m_logger);
 }
 
 void ClientImpl::SendControl(uint64_t curTimeMs) {
-  m_impl->SendControl(curTimeMs);
-  m_impl->m_wire.Flush();
+  DoSendControl(curTimeMs);
+  m_wire.Flush();
 }
 
 void ClientImpl::SendValues(uint64_t curTimeMs, bool flush) {
-  m_impl->SendValues(curTimeMs, flush);
-  m_impl->m_wire.Flush();
-}
-
-void ClientImpl::SetLocal(LocalInterface* local) {
-  m_impl->m_local = local;
+  DoSendValues(curTimeMs, flush);
+  m_wire.Flush();
 }
 
 void ClientImpl::SendInitial() {
-  m_impl->SendInitialValues();
-  m_impl->m_wire.Flush();
+  SendInitialValues();
+  m_wire.Flush();
 }

ntcore/src/main/native/cpp/net/ClientImpl.h

@@ -13,8 +13,12 @@
 #include <string_view>
 #include <vector>
 
+#include <wpi/DenseMap.h>
+
 #include "NetworkInterface.h"
+#include "PubSubOptions.h"
 #include "WireConnection.h"
+#include "WireDecoder.h"
 
 namespace wpi {
 class Logger;
@@ -30,14 +34,13 @@ namespace nt::net {
 struct ClientMessage;
 class WireConnection;
 
-class ClientImpl {
+class ClientImpl final : private ServerMessageHandler {
  public:
   ClientImpl(
       uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
       std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
           timeSyncUpdated,
       std::function<void(uint32_t repeatMs)> setPeriodic);
-  ~ClientImpl();
 
   void ProcessIncomingText(std::string_view data);
   void ProcessIncomingBinary(uint64_t curTimeMs, std::span<const uint8_t> data);
@@ -46,12 +49,67 @@ class ClientImpl {
   void SendControl(uint64_t curTimeMs);
   void SendValues(uint64_t curTimeMs, bool flush);
 
-  void SetLocal(LocalInterface* local);
+  void SetLocal(LocalInterface* local) { m_local = local; }
   void SendInitial();
 
  private:
-  class Impl;
-  std::unique_ptr<Impl> m_impl;
+  struct PublisherData {
+    NT_Publisher handle;
+    PubSubOptionsImpl options;
+    // in options as double, but copy here as integer; rounded to the nearest
+    // 10 ms
+    uint32_t periodMs;
+    uint64_t nextSendMs{0};
+    std::vector<Value> outValues;  // outgoing values
+  };
+
+  bool DoSendControl(uint64_t curTimeMs);
+  void DoSendValues(uint64_t curTimeMs, bool flush);
+  void SendInitialValues();
+  bool CheckNetworkReady(uint64_t curTimeMs);
+
+  // ServerMessageHandler interface
+  void ServerAnnounce(std::string_view name, int64_t id,
+                      std::string_view typeStr, const wpi::json& properties,
+                      std::optional<int64_t> pubuid) final;
+  void ServerUnannounce(std::string_view name, int64_t id) final;
+  void ServerPropertiesUpdate(std::string_view name, const wpi::json& update,
+                              bool ack) final;
+
+  void Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
+               std::string_view name, std::string_view typeStr,
+               const wpi::json& properties, const PubSubOptionsImpl& options);
+  bool Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle);
+  void SetValue(NT_Publisher pubHandle, const Value& value);
+
+  int m_inst;
+  WireConnection& m_wire;
+  wpi::Logger& m_logger;
+  LocalInterface* m_local{nullptr};
+  std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
+      m_timeSyncUpdated;
+  std::function<void(uint32_t repeatMs)> m_setPeriodic;
+
+  // indexed by publisher index
+  std::vector<std::unique_ptr<PublisherData>> m_publishers;
+
+  // indexed by server-provided topic id
+  wpi::DenseMap<int64_t, NT_Topic> m_topicMap;
+
+  // timestamp handling
+  static constexpr uint32_t kPingIntervalMs = 3000;
+  uint64_t m_nextPingTimeMs{0};
+  uint64_t m_pongTimeMs{0};
+  uint32_t m_rtt2Us{UINT32_MAX};
+  bool m_haveTimeOffset{false};
+  int64_t m_serverTimeOffsetUs{0};
+
+  // periodic sweep handling
+  uint32_t m_periodMs{kPingIntervalMs + 10};
+  uint64_t m_lastSendMs{0};
+
+  // outgoing queue
+  std::vector<ClientMessage> m_outgoing;
 };
 
 }  // namespace nt::net

ntcore/src/main/native/cpp/net/ServerImpl.h

@@ -6,6 +6,7 @@
 
 #include <stdint.h>
 
+#include <cmath>
 #include <functional>
 #include <memory>
 #include <span>
@@ -14,11 +15,28 @@
 #include <utility>
 #include <vector>
 
+#include <wpi/DenseMap.h>
+#include <wpi/StringMap.h>
+#include <wpi/UidVector.h>
+#include <wpi/json.h>
+
+#include "Message.h"
 #include "NetworkInterface.h"
+#include "PubSubOptions.h"
+#include "VectorSet.h"
+#include "WireConnection.h"
+#include "WireDecoder.h"
+#include "WireEncoder.h"
+#include "net3/Message3.h"
+#include "net3/SequenceNumber.h"
 #include "net3/WireConnection3.h"
+#include "net3/WireDecoder3.h"
 
 namespace wpi {
 class Logger;
+template <typename T>
+class SmallVectorImpl;
+class raw_ostream;
 }  // namespace wpi
 
 namespace nt::net3 {
@@ -38,7 +56,6 @@ class ServerImpl final {
       std::function<void(std::string_view name, uint16_t proto)>;
 
   explicit ServerImpl(wpi::Logger& logger);
-  ~ServerImpl();
 
   void SendControl(uint64_t curTimeMs);
   void SendValues(int clientId, uint64_t curTimeMs);
@@ -69,8 +86,357 @@ class ServerImpl final {
   std::string LoadPersistent(std::string_view in);
 
  private:
-  class Impl;
-  std::unique_ptr<Impl> m_impl;
+  static constexpr uint32_t kMinPeriodMs = 5;
+
+  struct PublisherData;
+  struct SubscriberData;
+  struct TopicData;
+
+  class ClientData {
+   public:
+    ClientData(std::string_view name, std::string_view connInfo, bool local,
+               ServerImpl::SetPeriodicFunc setPeriodic, ServerImpl& server,
+               int id, wpi::Logger& logger)
+        : m_name{name},
+          m_connInfo{connInfo},
+          m_local{local},
+          m_setPeriodic{std::move(setPeriodic)},
+          m_server{server},
+          m_id{id},
+          m_logger{logger} {}
+    virtual ~ClientData() = default;
+
+    virtual void ProcessIncomingText(std::string_view data) = 0;
+    virtual void ProcessIncomingBinary(std::span<const uint8_t> data) = 0;
+
+    enum SendMode { kSendDisabled = 0, kSendAll, kSendNormal, kSendImmNoFlush };
+
+    virtual void SendValue(TopicData* topic, const Value& value,
+                           SendMode mode) = 0;
+    virtual void SendAnnounce(TopicData* topic,
+                              std::optional<int64_t> pubuid) = 0;
+    virtual void SendUnannounce(TopicData* topic) = 0;
+    virtual void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
+                                      bool ack) = 0;
+    virtual void SendOutgoing(uint64_t curTimeMs) = 0;
+    virtual void Flush() = 0;
+
+    void UpdateMetaClientPub();
+    void UpdateMetaClientSub();
+
+    std::span<SubscriberData*> GetSubscribers(
+        std::string_view name, bool special,
+        wpi::SmallVectorImpl<SubscriberData*>& buf);
+
+    std::string_view GetName() const { return m_name; }
+    int GetId() const { return m_id; }
+
+   protected:
+    std::string m_name;
+    std::string m_connInfo;
+    bool m_local;  // local to machine
+    ServerImpl::SetPeriodicFunc m_setPeriodic;
+    // TODO: make this per-topic?
+    uint32_t m_periodMs{UINT32_MAX};
+    uint64_t m_lastSendMs{0};
+    ServerImpl& m_server;
+    int m_id;
+
+    wpi::Logger& m_logger;
+
+    wpi::DenseMap<int64_t, std::unique_ptr<PublisherData>> m_publishers;
+    wpi::DenseMap<int64_t, std::unique_ptr<SubscriberData>> m_subscribers;
+
+   public:
+    // meta topics
+    TopicData* m_metaPub = nullptr;
+    TopicData* m_metaSub = nullptr;
+  };
+
+  class ClientData4Base : public ClientData, protected ClientMessageHandler {
+   public:
+    ClientData4Base(std::string_view name, std::string_view connInfo,
+                    bool local, ServerImpl::SetPeriodicFunc setPeriodic,
+                    ServerImpl& server, int id, wpi::Logger& logger)
+        : ClientData{name, connInfo, local, setPeriodic, server, id, logger} {}
+
+   protected:
+    // ClientMessageHandler interface
+    void ClientPublish(int64_t pubuid, std::string_view name,
+                       std::string_view typeStr,
+                       const wpi::json& properties) final;
+    void ClientUnpublish(int64_t pubuid) final;
+    void ClientSetProperties(std::string_view name,
+                             const wpi::json& update) final;
+    void ClientSubscribe(int64_t subuid,
+                         std::span<const std::string> topicNames,
+                         const PubSubOptionsImpl& options) final;
+    void ClientUnsubscribe(int64_t subuid) final;
+
+    void ClientSetValue(int64_t pubuid, const Value& value);
+
+    wpi::DenseMap<TopicData*, bool> m_announceSent;
+  };
+
+  class ClientDataLocal final : public ClientData4Base {
+   public:
+    ClientDataLocal(ServerImpl& server, int id, wpi::Logger& logger)
+        : ClientData4Base{"", "", true, [](uint32_t) {}, server, id, logger} {}
+
+    void ProcessIncomingText(std::string_view data) final {}
+    void ProcessIncomingBinary(std::span<const uint8_t> data) final {}
+
+    void SendValue(TopicData* topic, const Value& value, SendMode mode) final;
+    void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
+    void SendUnannounce(TopicData* topic) final;
+    void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
+                              bool ack) final;
+    void SendOutgoing(uint64_t curTimeMs) final {}
+    void Flush() final {}
+
+    void HandleLocal(std::span<const ClientMessage> msgs);
+  };
+
+  class ClientData4 final : public ClientData4Base {
+   public:
+    ClientData4(std::string_view name, std::string_view connInfo, bool local,
+                WireConnection& wire, ServerImpl::SetPeriodicFunc setPeriodic,
+                ServerImpl& server, int id, wpi::Logger& logger)
+        : ClientData4Base{name,   connInfo, local, setPeriodic,
+                          server, id,       logger},
+          m_wire{wire} {}
+
+    void ProcessIncomingText(std::string_view data) final;
+    void ProcessIncomingBinary(std::span<const uint8_t> data) final;
+
+    void SendValue(TopicData* topic, const Value& value, SendMode mode) final;
+    void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
+    void SendUnannounce(TopicData* topic) final;
+    void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
+                              bool ack) final;
+    void SendOutgoing(uint64_t curTimeMs) final;
+
+    void Flush() final;
+
+   public:
+    WireConnection& m_wire;
+
+   private:
+    std::vector<ServerMessage> m_outgoing;
+    wpi::DenseMap<NT_Topic, size_t> m_outgoingValueMap;
+
+    bool WriteBinary(int64_t id, int64_t time, const Value& value) {
+      return WireEncodeBinary(SendBinary().Add(), id, time, value);
+    }
+
+    TextWriter& SendText() {
+      m_outBinary.reset();  // ensure proper interleaving of text and binary
+      if (!m_outText) {
+        m_outText = m_wire.SendText();
+      }
+      return *m_outText;
+    }
+
+    BinaryWriter& SendBinary() {
+      m_outText.reset();  // ensure proper interleaving of text and binary
+      if (!m_outBinary) {
+        m_outBinary = m_wire.SendBinary();
+      }
+      return *m_outBinary;
+    }
+
+    // valid when we are actively writing to this client
+    std::optional<TextWriter> m_outText;
+    std::optional<BinaryWriter> m_outBinary;
+  };
+
+  class ClientData3 final : public ClientData, private net3::MessageHandler3 {
+   public:
+    ClientData3(std::string_view connInfo, bool local,
+                net3::WireConnection3& wire,
+                ServerImpl::Connected3Func connected,
+                ServerImpl::SetPeriodicFunc setPeriodic, ServerImpl& server,
+                int id, wpi::Logger& logger)
+        : ClientData{"", connInfo, local, setPeriodic, server, id, logger},
+          m_connected{std::move(connected)},
+          m_wire{wire},
+          m_decoder{*this} {}
+
+    void ProcessIncomingText(std::string_view data) final {}
+    void ProcessIncomingBinary(std::span<const uint8_t> data) final;
+
+    void SendValue(TopicData* topic, const Value& value, SendMode mode) final;
+    void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
+    void SendUnannounce(TopicData* topic) final;
+    void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
+                              bool ack) final;
+    void SendOutgoing(uint64_t curTimeMs) final;
+
+    void Flush() final { m_wire.Flush(); }
+
+   private:
+    // MessageHandler3 interface
+    void KeepAlive() final;
+    void ServerHelloDone() final;
+    void ClientHelloDone() final;
+    void ClearEntries() final;
+    void ProtoUnsup(unsigned int proto_rev) final;
+    void ClientHello(std::string_view self_id, unsigned int proto_rev) final;
+    void ServerHello(unsigned int flags, std::string_view self_id) final;
+    void EntryAssign(std::string_view name, unsigned int id,
+                     unsigned int seq_num, const Value& value,
+                     unsigned int flags) final;
+    void EntryUpdate(unsigned int id, unsigned int seq_num,
+                     const Value& value) final;
+    void FlagsUpdate(unsigned int id, unsigned int flags) final;
+    void EntryDelete(unsigned int id) final;
+    void ExecuteRpc(unsigned int id, unsigned int uid,
+                    std::span<const uint8_t> params) final {}
+    void RpcResponse(unsigned int id, unsigned int uid,
+                     std::span<const uint8_t> result) final {}
+
+    ServerImpl::Connected3Func m_connected;
+    net3::WireConnection3& m_wire;
+
+    enum State { kStateInitial, kStateServerHelloComplete, kStateRunning };
+    State m_state{kStateInitial};
+    net3::WireDecoder3 m_decoder;
+
+    std::vector<net3::Message3> m_outgoing;
+    wpi::DenseMap<NT_Topic, size_t> m_outgoingValueMap;
+    int64_t m_nextPubUid{1};
+
+    struct TopicData3 {
+      explicit TopicData3(TopicData* topic) { UpdateFlags(topic); }
+
+      unsigned int flags{0};
+      net3::SequenceNumber seqNum;
+      bool sentAssign{false};
+      bool published{false};
+      int64_t pubuid{0};
+
+      bool UpdateFlags(TopicData* topic);
+    };
+    wpi::DenseMap<TopicData*, TopicData3> m_topics3;
+    TopicData3* GetTopic3(TopicData* topic) {
+      return &m_topics3.try_emplace(topic, topic).first->second;
+    }
+  };
+
+  struct TopicData {
+    TopicData(std::string_view name, std::string_view typeStr)
+        : name{name}, typeStr{typeStr} {}
+    TopicData(std::string_view name, std::string_view typeStr,
+              wpi::json properties)
+        : name{name}, typeStr{typeStr}, properties(std::move(properties)) {
+      RefreshProperties();
+    }
+
+    bool IsPublished() const {
+      return persistent || retained || !publishers.empty();
+    }
+
+    // returns true if properties changed
+    bool SetProperties(const wpi::json& update);
+    void RefreshProperties();
+    bool SetFlags(unsigned int flags_);
+
+    std::string name;
+    unsigned int id;
+    Value lastValue;
+    ClientData* lastValueClient = nullptr;
+    std::string typeStr;
+    wpi::json properties = wpi::json::object();
+    bool persistent{false};
+    bool retained{false};
+    bool special{false};
+    NT_Topic localHandle{0};
+
+    VectorSet<PublisherData*> publishers;
+    VectorSet<SubscriberData*> subscribers;
+
+    // meta topics
+    TopicData* metaPub = nullptr;
+    TopicData* metaSub = nullptr;
+  };
+
+  struct PublisherData {
+    PublisherData(ClientData* client, TopicData* topic, int64_t pubuid)
+        : client{client}, topic{topic}, pubuid{pubuid} {}
+
+    ClientData* client;
+    TopicData* topic;
+    int64_t pubuid;
+  };
+
+  struct SubscriberData {
+    SubscriberData(ClientData* client, std::span<const std::string> topicNames,
+                   int64_t subuid, const PubSubOptionsImpl& options)
+        : client{client},
+          topicNames{topicNames.begin(), topicNames.end()},
+          subuid{subuid},
+          options{options},
+          periodMs(std::lround(options.periodicMs / 10.0) * 10) {
+      if (periodMs < kMinPeriodMs) {
+        periodMs = kMinPeriodMs;
+      }
+    }
+
+    void Update(std::span<const std::string> topicNames_,
+                const PubSubOptionsImpl& options_) {
+      topicNames = {topicNames_.begin(), topicNames_.end()};
+      options = options_;
+      periodMs = std::lround(options_.periodicMs / 10.0) * 10;
+      if (periodMs < kMinPeriodMs) {
+        periodMs = kMinPeriodMs;
+      }
+    }
+
+    bool Matches(std::string_view name, bool special);
+
+    ClientData* client;
+    std::vector<std::string> topicNames;
+    int64_t subuid;
+    PubSubOptionsImpl options;
+    // in options as double, but copy here as integer; rounded to the nearest
+    // 10 ms
+    uint32_t periodMs;
+  };
+
+  wpi::Logger& m_logger;
+  LocalInterface* m_local{nullptr};
+  bool m_controlReady{false};
+
+  ClientDataLocal* m_localClient;
+  std::vector<std::unique_ptr<ClientData>> m_clients;
+  wpi::UidVector<std::unique_ptr<TopicData>, 16> m_topics;
+  wpi::StringMap<TopicData*> m_nameTopics;
+  bool m_persistentChanged{false};
+
+  // global meta topics (other meta topics are linked to from the specific
+  // client or topic)
+  TopicData* m_metaClients;
+
+  void DumpPersistent(wpi::raw_ostream& os);
+
+  // helper functions
+  TopicData* CreateTopic(ClientData* client, std::string_view name,
+                         std::string_view typeStr, const wpi::json& properties,
+                         bool special = false);
+  TopicData* CreateMetaTopic(std::string_view name);
+  void DeleteTopic(TopicData* topic);
+  void SetProperties(ClientData* client, TopicData* topic,
+                     const wpi::json& update);
+  void SetFlags(ClientData* client, TopicData* topic, unsigned int flags);
+  void SetValue(ClientData* client, TopicData* topic, const Value& value);
+
+  // update meta topic values from data structures
+  void UpdateMetaClients(const std::vector<ConnectionInfo>& conns);
+  void UpdateMetaTopicPub(TopicData* topic);
+  void UpdateMetaTopicSub(TopicData* topic);
+
+  void PropertiesChanged(ClientData* client, TopicData* topic,
+                         const wpi::json& update);
 };
 
 }  // namespace nt::net