[ntcore] Various NT4 fixes (#4474)

* TopicListener: Fix Add() return values
* Update PubSubOption poll storage documentation
* Update NetworkTableEntry::GetValue() doc
* Add documentation regarding asynchronous callbacks
* Unpublish entry: set publisher to nullptr
* Implement ValueListenerPoller default constructor
* Remove SetNetworkIdentity, make parameter to StartClient
* URI-escape client ID, improve error message
* Add connected message with client id; also improve disconnected message a bit
* Handle SetServers either before or after StartClient
* Fix client use-after-free; also delay reconnect after disconnect to rate limit
* Don't re-announce to already subscribed client; we especially don't want to send the last value again
* Always accept in-order sets, only use timestamp for tiebreak
* Fix LocalStorage::StartNetwork race
* Remove unused/unimplemented function

Also:
* [glass] Remove debug print
* [glass] Fix mpack string decoding
* [cameraserver] Fix up startclient

ntcore/src/main/native/cpp/InstanceImpl.cpp

@@ -120,23 +120,29 @@ void InstanceImpl::StopServer() {
   networkMode = NT_NET_MODE_NONE;
 }
 
-void InstanceImpl::StartClient3() {
+void InstanceImpl::StartClient3(std::string_view identity) {
   std::scoped_lock lock{m_mutex};
   if (networkMode != NT_NET_MODE_NONE) {
     return;
   }
   m_networkClient = std::make_shared<NetworkClient3>(
-      m_inst, m_identity, localStorage, connectionList, logger);
+      m_inst, identity, localStorage, connectionList, logger);
+  if (!m_servers.empty()) {
+    m_networkClient->SetServers(m_servers);
+  }
   networkMode = NT_NET_MODE_CLIENT3;
 }
 
-void InstanceImpl::StartClient4() {
+void InstanceImpl::StartClient4(std::string_view identity) {
   std::scoped_lock lock{m_mutex};
   if (networkMode != NT_NET_MODE_NONE) {
     return;
   }
   m_networkClient = std::make_shared<NetworkClient>(
-      m_inst, m_identity, localStorage, connectionList, logger);
+      m_inst, identity, localStorage, connectionList, logger);
+  if (!m_servers.empty()) {
+    m_networkClient->SetServers(m_servers);
+  }
   networkMode = NT_NET_MODE_CLIENT4;
 }
 
@@ -149,9 +155,13 @@ void InstanceImpl::StopClient() {
   networkMode = NT_NET_MODE_NONE;
 }
 
-void InstanceImpl::SetIdentity(std::string_view identity) {
+void InstanceImpl::SetServers(
+    std::span<const std::pair<std::string, unsigned int>> servers) {
   std::scoped_lock lock{m_mutex};
-  m_identity = identity;
+  m_servers = {servers.begin(), servers.end()};
+  if (m_networkClient) {
+    m_networkClient->SetServers(servers);
+  }
 }
 
 std::shared_ptr<NetworkServer> InstanceImpl::GetServer() {

ntcore/src/main/native/cpp/InstanceImpl.h

@@ -8,6 +8,7 @@
 #include <memory>
 #include <string>
 #include <string_view>
+#include <utility>
 #include <vector>
 
 #include <wpi/mutex.h>
@@ -45,10 +46,11 @@ class InstanceImpl {
                    std::string_view listenAddress, unsigned int port3,
                    unsigned int port4);
   void StopServer();
-  void StartClient3();
-  void StartClient4();
+  void StartClient3(std::string_view identity);
+  void StartClient4(std::string_view identity);
   void StopClient();
-  void SetIdentity(std::string_view identity);
+  void SetServers(
+      std::span<const std::pair<std::string, unsigned int>> servers);
 
   std::shared_ptr<NetworkServer> GetServer();
   std::shared_ptr<INetworkClient> GetClient();
@@ -68,9 +70,9 @@ class InstanceImpl {
   static wpi::mutex s_mutex;
 
   wpi::mutex m_mutex;
-  std::string m_identity;
   std::shared_ptr<NetworkServer> m_networkServer;
   std::shared_ptr<INetworkClient> m_networkClient;
+  std::vector<std::pair<std::string, unsigned int>> m_servers;
   int m_inst;
 };
 

ntcore/src/main/native/cpp/LocalStorage.cpp

@@ -77,6 +77,7 @@ struct TopicData {
   std::string name;
 
   Value lastValue;  // also stores timestamp
+  bool lastValueNetwork{false};
   NT_Type type{NT_UNASSIGNED};
   std::string typeStr;
   unsigned int flags{0};            // for NT3 APIs
@@ -646,6 +647,7 @@ void LSImpl::CheckReset(TopicData* topic) {
     return;
   }
   topic->lastValue = {};
+  topic->lastValueNetwork = false;
   topic->type = NT_UNASSIGNED;
   topic->typeStr.clear();
   topic->flags = 0;
@@ -658,10 +660,13 @@ bool LSImpl::SetValue(TopicData* topic, const Value& value,
   if (topic->type != NT_UNASSIGNED && topic->type != value.type()) {
     return false;
   }
-  if (!topic->lastValue || value.time() >= topic->lastValue.time()) {
+  bool isNetwork = (eventFlags & NT_VALUE_NOTIFY_LOCAL) == 0;
+  if (!topic->lastValue || topic->lastValueNetwork == isNetwork ||
+      value.time() >= topic->lastValue.time()) {
     // TODO: notify option even if older value
     topic->type = value.type();
     topic->lastValue = value;
+    topic->lastValueNetwork = isNetwork;
     NotifyValue(topic, eventFlags);
   }
   if (topic->datalogType == value.type()) {
@@ -999,6 +1004,7 @@ std::unique_ptr<PublisherData> LSImpl::RemoveLocalPublisher(
 
 SubscriberData* LSImpl::AddLocalSubscriber(TopicData* topic,
                                            const PubSubConfig& config) {
+  DEBUG4("AddLocalSubscriber({})", topic->name);
   auto subscriber = m_subscribers.Add(m_inst, topic, config);
   topic->localSubscribers.Add(subscriber);
   // set subscriber to active if the type matches
@@ -1011,6 +1017,7 @@ SubscriberData* LSImpl::AddLocalSubscriber(TopicData* topic,
         topic->name, config.typeStr, topic->typeStr);
   }
   if (m_network) {
+    DEBUG4("-> NetworkSubscribe({})", topic->name);
     m_network->Subscribe(subscriber->handle, {{topic->name}}, config);
   }
   return subscriber;
@@ -1578,7 +1585,8 @@ void LocalStorage::NetworkSetValue(NT_Topic topicHandle, const Value& value) {
   }
 }
 
-void LocalStorage::StartNetwork(net::NetworkStartupInterface& startup) {
+void LocalStorage::StartNetwork(net::NetworkStartupInterface& startup,
+                                net::NetworkInterface* network) {
   std::scoped_lock lock{m_mutex};
   // publish all active publishers to the network and send last values
   // only send value once per topic
@@ -1596,19 +1604,13 @@ void LocalStorage::StartNetwork(net::NetworkStartupInterface& startup) {
     }
   }
   for (auto&& subscriber : m_impl->m_subscribers) {
-    if (subscriber->active) {
-      startup.Subscribe(subscriber->handle, {{subscriber->topic->name}},
-                        subscriber->config);
-    }
+    startup.Subscribe(subscriber->handle, {{subscriber->topic->name}},
+                      subscriber->config);
   }
   for (auto&& subscriber : m_impl->m_multiSubscribers) {
     startup.Subscribe(subscriber->handle, subscriber->prefixes,
                       subscriber->options);
   }
-}
-
-void LocalStorage::SetNetwork(net::NetworkInterface* network) {
-  std::scoped_lock lock{m_mutex};
   m_impl->m_network = network;
 }
 
@@ -1914,6 +1916,7 @@ void LocalStorage::Unpublish(NT_Handle pubentryHandle) {
   } else if (auto entry = m_impl->m_entries.Get(pubentryHandle)) {
     if (entry->publisher) {
       m_impl->RemoveLocalPublisher(entry->publisher->handle);
+      entry->publisher = nullptr;
     }
   } else {
     // TODO: report warning

ntcore/src/main/native/cpp/LocalStorage.h

@@ -41,8 +41,8 @@ class LocalStorage final : public net::ILocalStorage {
                                bool ack) final;
   void NetworkSetValue(NT_Topic topicHandle, const Value& value) final;
 
-  void StartNetwork(net::NetworkStartupInterface& startup) final;
-  void SetNetwork(net::NetworkInterface* network) final;
+  void StartNetwork(net::NetworkStartupInterface& startup,
+                    net::NetworkInterface* network) final;
   void ClearNetwork() final;
 
   // User functions.  These are the actual implementations of the corresponding

ntcore/src/main/native/cpp/NetworkClient.cpp

@@ -11,9 +11,11 @@
 #include <vector>
 
 #include <fmt/format.h>
+#include <wpi/SmallString.h>
 #include <wpi/StringExtras.h>
 #include <wpinet/DsClient.h>
 #include <wpinet/EventLoopRunner.h>
+#include <wpinet/HttpUtil.h>
 #include <wpinet/ParallelTcpConnector.h>
 #include <wpinet/WebSocket.h>
 #include <wpinet/uv/Async.h>
@@ -114,7 +116,7 @@ class NCImpl4 : public NCImpl {
   void WsConnected(wpi::WebSocket& ws, uv::Tcp& tcp);
   void Disconnect(std::string_view reason) override;
 
-  std::unique_ptr<net::WebSocketConnection> m_wire;
+  std::shared_ptr<net::WebSocketConnection> m_wire;
   std::unique_ptr<net::ClientImpl> m_clientImpl;
 };
 
@@ -191,7 +193,8 @@ void NCImpl::Disconnect(std::string_view reason) {
   m_connHandle = 0;
 
   // start trying to connect again
-  m_parallelConnect->Disconnected();
+  uv::Timer::SingleShot(m_loop, kReconnectRate,
+                        [this] { m_parallelConnect->Disconnected(); });
 }
 
 NCImpl3::NCImpl3(int inst, std::string_view id,
@@ -214,8 +217,10 @@ NCImpl3::NCImpl3(int inst, std::string_view id,
     // set up flush async
     m_flush = uv::Async<>::Create(m_loop);
     m_flush->wakeup.connect([this] {
-      HandleLocal();
-      m_clientImpl->SendPeriodic(m_loop.Now().count());
+      if (m_clientImpl) {
+        HandleLocal();
+        m_clientImpl->SendPeriodic(m_loop.Now().count());
+      }
     });
     m_flushAtomic = m_flush.get();
 
@@ -299,9 +304,8 @@ void NCImpl3::TcpConnected(uv::Tcp& tcp) {
 
         {
           net3::ClientStartup3 startup{*m_clientImpl};
-          m_localStorage.StartNetwork(startup);
+          m_localStorage.StartNetwork(startup, &m_localQueue);
         }
-        m_localStorage.SetNetwork(&m_localQueue);
         m_clientImpl->SetLocal(&m_localStorage);
       });
 
@@ -373,7 +377,9 @@ NCImpl4::~NCImpl4() {
 
 void NCImpl4::HandleLocal() {
   m_localQueue.ReadQueue(&m_localMsgs);
-  m_clientImpl->HandleLocal(std::move(m_localMsgs));
+  if (m_clientImpl) {
+    m_clientImpl->HandleLocal(std::move(m_localMsgs));
+  }
 }
 
 void NCImpl4::TcpConnected(uv::Tcp& tcp) {
@@ -387,9 +393,10 @@ void NCImpl4::TcpConnected(uv::Tcp& tcp) {
   }
   wpi::WebSocket::ClientOptions options;
   options.handshakeTimeout = kWebsocketHandshakeTimeout;
-  auto ws =
-      wpi::WebSocket::CreateClient(tcp, fmt::format("/nt/{}", m_id), "",
-                                   {{"networktables.first.wpi.edu"}}, options);
+  wpi::SmallString<128> idBuf;
+  auto ws = wpi::WebSocket::CreateClient(
+      tcp, fmt::format("/nt/{}", wpi::EscapeURI(m_id, idBuf)), "",
+      {{"networktables.first.wpi.edu"}}, options);
   ws->SetMaxMessageSize(kMaxMessageSize);
   ws->open.connect([this, &tcp, ws = ws.get()](std::string_view) {
     if (m_connList.IsConnected()) {
@@ -410,7 +417,7 @@ void NCImpl4::WsConnected(wpi::WebSocket& ws, uv::Tcp& tcp) {
   INFO("CONNECTED NT4 to {} port {}", connInfo.remote_ip, connInfo.remote_port);
   m_connHandle = m_connList.AddConnection(connInfo);
 
-  m_wire = std::make_unique<net::WebSocketConnection>(ws);
+  m_wire = std::make_shared<net::WebSocketConnection>(ws);
   m_clientImpl = std::make_unique<net::ClientImpl>(
       m_loop.Now().count(), m_inst, *m_wire, m_logger,
       [this](uint32_t repeatMs) {
@@ -420,9 +427,8 @@ void NCImpl4::WsConnected(wpi::WebSocket& ws, uv::Tcp& tcp) {
       });
   {
     net::ClientStartup startup{*m_clientImpl};
-    m_localStorage.StartNetwork(startup);
+    m_localStorage.StartNetwork(startup, &m_localQueue);
   }
-  m_localStorage.SetNetwork(&m_localQueue);
   m_clientImpl->SetLocal(&m_localStorage);
   ws.closed.connect([this, &ws](uint16_t, std::string_view reason) {
     if (!ws.GetStream().IsLoopClosing()) {
@@ -430,10 +436,14 @@ void NCImpl4::WsConnected(wpi::WebSocket& ws, uv::Tcp& tcp) {
     }
   });
   ws.text.connect([this](std::string_view data, bool) {
-    m_clientImpl->ProcessIncomingText(data);
+    if (m_clientImpl) {
+      m_clientImpl->ProcessIncomingText(data);
+    }
   });
   ws.binary.connect([this](std::span<const uint8_t> data, bool) {
-    m_clientImpl->ProcessIncomingBinary(data);
+    if (m_clientImpl) {
+      m_clientImpl->ProcessIncomingBinary(data);
+    }
   });
 }
 
@@ -482,7 +492,9 @@ void NetworkClient::FlushLocal() {
 void NetworkClient::Flush() {
   m_impl->m_loopRunner.ExecAsync([this](uv::Loop&) {
     m_impl->HandleLocal();
-    m_impl->m_clientImpl->SendValues(m_impl->m_loop.Now().count());
+    if (m_impl->m_clientImpl) {
+      m_impl->m_clientImpl->SendValues(m_impl->m_loop.Now().count());
+    }
   });
 }
 

ntcore/src/main/native/cpp/NetworkServer.cpp

@@ -11,12 +11,14 @@
 #include <system_error>
 #include <vector>
 
+#include <wpi/SmallString.h>
 #include <wpi/StringExtras.h>
 #include <wpi/fs.h>
 #include <wpi/mutex.h>
 #include <wpi/raw_istream.h>
 #include <wpi/raw_ostream.h>
 #include <wpinet/EventLoopRunner.h>
+#include <wpinet/HttpUtil.h>
 #include <wpinet/HttpWebSocketServerConnection.h>
 #include <wpinet/UrlParser.h>
 #include <wpinet/uv/Async.h>
@@ -87,7 +89,7 @@ class ServerConnection4 final
   void ProcessRequest() final;
   void ProcessWsUpgrade() final;
 
-  std::unique_ptr<net::WebSocketConnection> m_wire;
+  std::shared_ptr<net::WebSocketConnection> m_wire;
 };
 
 class ServerConnection3 : public ServerConnection {
@@ -97,7 +99,7 @@ class ServerConnection3 : public ServerConnection {
                     wpi::Logger& logger);
 
  private:
-  std::unique_ptr<net3::UvStreamConnection3> m_wire;
+  std::shared_ptr<net3::UvStreamConnection3> m_wire;
 };
 
 class NSImpl {
@@ -223,20 +225,24 @@ void ServerConnection4::ProcessWsUpgrade() {
   }
   DEBUG4("path: '{}'", path);
 
+  wpi::SmallString<128> nameBuf;
   std::string_view name;
+  bool err = false;
   if (wpi::starts_with(path, "/nt/")) {
-    name = wpi::drop_front(path, 4);
+    name = wpi::UnescapeURI(wpi::drop_front(path, 4), nameBuf, &err);
   }
-  if (name.empty()) {
-    INFO("invalid path '{}' (from {}), closing", path, m_connInfo);
-    m_websocket->Fail(404, fmt::format("invalid path '{}'", path));
+  if (err || name.empty()) {
+    INFO("invalid path '{}' (from {}), must match /nt/[clientId], closing",
+         path, m_connInfo);
+    m_websocket->Fail(
+        404, fmt::format("invalid path '{}', must match /nt/[clientId]", path));
     return;
   }
 
   m_websocket->SetMaxMessageSize(kMaxMessageSize);
 
   m_websocket->open.connect([this, name = std::string{name}](std::string_view) {
-    m_wire = std::make_unique<net::WebSocketConnection>(*m_websocket);
+    m_wire = std::make_shared<net::WebSocketConnection>(*m_websocket);
     // TODO: set local flag appropriately
     m_clientId = m_server.m_serverImpl.AddClient(
         name, m_connInfo, false, *m_wire,
@@ -247,11 +253,12 @@ void ServerConnection4::ProcessWsUpgrade() {
       m_websocket->Fail(409, fmt::format("duplicate name '{}'", name));
       return;
     }
+    INFO("CONNECTED NT4 client '{}' (from {})", name, m_connInfo);
     m_info.remote_id = name;
     m_server.AddConnection(this, m_info);
     m_websocket->closed.connect([this](uint16_t, std::string_view reason) {
-      INFO("NT4 connection '{}' closed (from {})", m_info.remote_id,
-           m_connInfo);
+      INFO("DISCONNECTED NT4 client '{}' (from {}): {}", m_info.remote_id,
+           m_connInfo, reason);
       ConnectionClosed();
     });
     m_websocket->text.connect([this](std::string_view data, bool) {
@@ -269,7 +276,7 @@ ServerConnection3::ServerConnection3(std::shared_ptr<uv::Stream> stream,
                                      NSImpl& server, std::string_view addr,
                                      unsigned int port, wpi::Logger& logger)
     : ServerConnection{server, addr, port, logger},
-      m_wire{std::make_unique<net3::UvStreamConnection3>(*stream)} {
+      m_wire{std::make_shared<net3::UvStreamConnection3>(*stream)} {
   m_info.remote_ip = addr;
   m_info.remote_port = port;
 
@@ -280,6 +287,7 @@ ServerConnection3::ServerConnection3(std::shared_ptr<uv::Stream> stream,
         m_info.remote_id = name;
         m_info.protocol_version = proto;
         m_server.AddConnection(this, m_info);
+        INFO("CONNECTED NT4 client '{}' (from {})", name, m_connInfo);
       },
       [this](uint32_t repeatMs) { UpdatePeriodicTimer(repeatMs); });
 
@@ -298,7 +306,7 @@ ServerConnection3::ServerConnection3(std::shared_ptr<uv::Stream> stream,
     m_wire->GetStream().Shutdown([this] { m_wire->GetStream().Close(); });
   });
   stream->closed.connect([this] {
-    INFO("NT3 connection '{}' closed (from {}): {}", m_info.remote_id,
+    INFO("DISCONNECTED NT3 client '{}' (from {}): {}", m_info.remote_id,
          m_connInfo, m_wire->GetDisconnectReason());
     ConnectionClosed();
   });
@@ -332,9 +340,8 @@ NSImpl::NSImpl(std::string_view persistentFilename,
     // connect local storage to server
     {
       net::ServerStartup startup{m_serverImpl};
-      m_localStorage.StartNetwork(startup);
+      m_localStorage.StartNetwork(startup, &m_localQueue);
     }
-    m_localStorage.SetNetwork(&m_localQueue);
     m_serverImpl.SetLocal(&m_localStorage);
 
     // load persistent file first, then initialize

ntcore/src/main/native/cpp/jni/NetworkTablesJNI.cpp

@@ -1225,22 +1225,6 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_removeConnectionListener
   nt::RemoveConnectionListener(connListenerUid);
 }
 
-/*
- * Class:     edu_wpi_first_networktables_NetworkTablesJNI
- * Method:    setNetworkIdentity
- * Signature: (ILjava/lang/String;)V
- */
-JNIEXPORT void JNICALL
-Java_edu_wpi_first_networktables_NetworkTablesJNI_setNetworkIdentity
-  (JNIEnv* env, jclass, jint inst, jstring name)
-{
-  if (!name) {
-    nullPointerEx.Throw(env, "name cannot be null");
-    return;
-  }
-  nt::SetNetworkIdentity(inst, JStringRef{env, name}.str());
-}
-
 /*
  * Class:     edu_wpi_first_networktables_NetworkTablesJNI
  * Method:    getNetworkMode
@@ -1314,25 +1298,33 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_stopServer
 /*
  * Class:     edu_wpi_first_networktables_NetworkTablesJNI
  * Method:    startClient3
- * Signature: (I)V
+ * Signature: (ILjava/lang/String;)V
  */
 JNIEXPORT void JNICALL
 Java_edu_wpi_first_networktables_NetworkTablesJNI_startClient3
-  (JNIEnv*, jclass, jint inst)
+  (JNIEnv* env, jclass, jint inst, jstring identity)
 {
-  nt::StartClient3(inst);
+  if (!identity) {
+    nullPointerEx.Throw(env, "identity cannot be null");
+    return;
+  }
+  nt::StartClient3(inst, JStringRef{env, identity}.str());
 }
 
 /*
  * Class:     edu_wpi_first_networktables_NetworkTablesJNI
  * Method:    startClient4
- * Signature: (I)V
+ * Signature: (ILjava/lang/String;)V
  */
 JNIEXPORT void JNICALL
 Java_edu_wpi_first_networktables_NetworkTablesJNI_startClient4
-  (JNIEnv*, jclass, jint inst)
+  (JNIEnv* env, jclass, jint inst, jstring identity)
 {
-  nt::StartClient4(inst);
+  if (!identity) {
+    nullPointerEx.Throw(env, "identity cannot be null");
+    return;
+  }
+  nt::StartClient4(inst, JStringRef{env, identity}.str());
 }
 
 /*

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

@@ -54,7 +54,6 @@ class CImpl : public ServerMessageHandler {
 
   void ProcessIncomingBinary(std::span<const uint8_t> data);
   void HandleLocal(std::vector<ClientMessage>&& msgs);
-  void SendOutgoing(std::span<const ClientMessage> msgs);
   bool SendControl(uint64_t curTimeMs);
   void SendValues(uint64_t curTimeMs);
   bool CheckNetworkReady();

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

@@ -59,8 +59,8 @@ class NetworkInterface : public NetworkStartupInterface {
 
 class ILocalStorage : public LocalInterface {
  public:
-  virtual void StartNetwork(NetworkStartupInterface& startup) = 0;
-  virtual void SetNetwork(NetworkInterface* network) = 0;
+  virtual void StartNetwork(NetworkStartupInterface& startup,
+                            NetworkInterface* network) = 0;
   virtual void ClearNetwork() = 0;
 };
 

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

@@ -325,6 +325,7 @@ struct TopicData {
   std::string name;
   unsigned int id;
   Value lastValue;
+  ClientData* lastValueClient = nullptr;
   std::string typeStr;
   wpi::json properties = wpi::json::object();
   bool persistent{false};
@@ -627,6 +628,15 @@ void ClientData4Base::ClientSubscribe(int64_t subuid,
       removed = topic->subscribers.Remove(sub.get());
     }
 
+    // is client already subscribed?
+    bool wasSubscribed = false;
+    for (auto subscriber : topic->subscribers) {
+      if (subscriber->client == this) {
+        wasSubscribed = true;
+        break;
+      }
+    }
+
     bool added = false;
     if (sub->Matches(topic->name, topic->special)) {
       topic->subscribers.Add(sub.get());
@@ -645,7 +655,7 @@ void ClientData4Base::ClientSubscribe(int64_t subuid,
       }
     }
 
-    if (added && !removed) {
+    if (!wasSubscribed && added && !removed) {
       // announce topic to client
       DEBUG4("client {}: announce {}", m_id, topic->name);
       SendAnnounce(topic.get(), std::nullopt);
@@ -2086,11 +2096,13 @@ void SImpl::SetFlags(ClientData* client, TopicData* topic, unsigned int flags) {
 }
 
 void SImpl::SetValue(ClientData* client, TopicData* topic, const Value& value) {
-  // update retained value if timestamp newer
-  if (!topic->lastValue || value.time() > topic->lastValue.time()) {
+  // update retained value if from same client or timestamp newer
+  if (!topic->lastValue || topic->lastValueClient == client ||
+      value.time() >= topic->lastValue.time()) {
     DEBUG4("updating '{}' last value (time was {} is {})", topic->name,
            topic->lastValue.time(), value.time());
     topic->lastValue = value;
+    topic->lastValueClient = client;
 
     // if persistent, update flag
     if (topic->persistent) {

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

@@ -44,10 +44,12 @@ void WebSocketConnection::Flush() {
   }
 
   ++m_sendsActive;
-  m_ws.SendFrames(m_ws_frames, [this](auto bufs, auto) {
-    m_buf_pool.insert(m_buf_pool.end(), bufs.begin(), bufs.end());
-    if (m_sendsActive > 0) {
-      --m_sendsActive;
+  m_ws.SendFrames(m_ws_frames, [selfweak = weak_from_this()](auto bufs, auto) {
+    if (auto self = selfweak.lock()) {
+      self->m_buf_pool.insert(self->m_buf_pool.end(), bufs.begin(), bufs.end());
+      if (self->m_sendsActive > 0) {
+        --self->m_sendsActive;
+      }
     }
   });
   m_frames.clear();

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

@@ -4,6 +4,7 @@
 
 #pragma once
 
+#include <memory>
 #include <vector>
 
 #include <wpi/SmallVector.h>
@@ -15,7 +16,9 @@
 
 namespace nt::net {
 
-class WebSocketConnection final : public WireConnection {
+class WebSocketConnection final
+    : public WireConnection,
+      public std::enable_shared_from_this<WebSocketConnection> {
  public:
   explicit WebSocketConnection(wpi::WebSocket& ws);
   ~WebSocketConnection() override;

ntcore/src/main/native/cpp/net3/UvStreamConnection3.cpp

@@ -23,10 +23,12 @@ void UvStreamConnection3::Flush() {
     return;
   }
   ++m_sendsActive;
-  m_stream.Write(m_buffers, [this](auto bufs, auto) {
-    m_buf_pool.insert(m_buf_pool.end(), bufs.begin(), bufs.end());
-    if (m_sendsActive > 0) {
-      --m_sendsActive;
+  m_stream.Write(m_buffers, [selfweak = weak_from_this()](auto bufs, auto) {
+    if (auto self = selfweak.lock()) {
+      self->m_buf_pool.insert(self->m_buf_pool.end(), bufs.begin(), bufs.end());
+      if (self->m_sendsActive > 0) {
+        --self->m_sendsActive;
+      }
     }
   });
   m_buffers.clear();

ntcore/src/main/native/cpp/net3/UvStreamConnection3.h

@@ -4,6 +4,7 @@
 
 #pragma once
 
+#include <memory>
 #include <string>
 #include <string_view>
 #include <vector>
@@ -20,7 +21,9 @@ class Stream;
 
 namespace nt::net3 {
 
-class UvStreamConnection3 final : public WireConnection3 {
+class UvStreamConnection3 final
+    : public WireConnection3,
+      public std::enable_shared_from_this<UvStreamConnection3> {
   static constexpr size_t kAllocSize = 4096;
 
  public:

ntcore/src/main/native/cpp/ntcore_c.cpp

@@ -542,10 +542,6 @@ void NT_RemoveConnectionListener(NT_ConnectionListener conn_listener) {
  * Client/Server Functions
  */
 
-void NT_SetNetworkIdentity(NT_Inst inst, const char* name, size_t name_len) {
-  nt::SetNetworkIdentity(inst, {name, name_len});
-}
-
 unsigned int NT_GetNetworkMode(NT_Inst inst) {
   return nt::GetNetworkMode(inst);
 }
@@ -568,12 +564,12 @@ void NT_StopServer(NT_Inst inst) {
   nt::StopServer(inst);
 }
 
-void NT_StartClient3(NT_Inst inst) {
-  nt::StartClient3(inst);
+void NT_StartClient3(NT_Inst inst, const char* identity) {
+  nt::StartClient3(inst, identity);
 }
 
-void NT_StartClient4(NT_Inst inst) {
-  nt::StartClient4(inst);
+void NT_StartClient4(NT_Inst inst, const char* identity) {
+  nt::StartClient4(inst, identity);
 }
 
 void NT_StopClient(NT_Inst inst) {

ntcore/src/main/native/cpp/ntcore_cpp.cpp

@@ -624,12 +624,6 @@ void StopConnectionDataLog(NT_ConnectionDataLogger logger) {
  * Client/Server Functions
  */
 
-void SetNetworkIdentity(NT_Inst inst, std::string_view name) {
-  if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
-    ii->SetIdentity(name);
-  }
-}
-
 unsigned int GetNetworkMode(NT_Inst inst) {
   if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
     return ii->networkMode;
@@ -664,15 +658,15 @@ void StopServer(NT_Inst inst) {
   }
 }
 
-void StartClient3(NT_Inst inst) {
+void StartClient3(NT_Inst inst, std::string_view identity) {
   if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
-    ii->StartClient3();
+    ii->StartClient3(identity);
   }
 }
 
-void StartClient4(NT_Inst inst) {
+void StartClient4(NT_Inst inst, std::string_view identity) {
   if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
-    ii->StartClient4();
+    ii->StartClient4(identity);
   }
 }
 
@@ -690,44 +684,39 @@ void SetServer(
     NT_Inst inst,
     std::span<const std::pair<std::string_view, unsigned int>> servers) {
   if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
-    if (auto client = ii->GetClient()) {
-      std::vector<std::pair<std::string, unsigned int>> serversCopy;
-      serversCopy.reserve(servers.size());
-      for (auto&& server : servers) {
-        serversCopy.emplace_back(std::string{server.first}, server.second);
-      }
-      client->SetServers(serversCopy);
+    std::vector<std::pair<std::string, unsigned int>> serversCopy;
+    serversCopy.reserve(servers.size());
+    for (auto&& server : servers) {
+      serversCopy.emplace_back(std::string{server.first}, server.second);
     }
+    ii->SetServers(serversCopy);
   }
 }
 
 void SetServerTeam(NT_Inst inst, unsigned int team, unsigned int port) {
   if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
-    if (auto client = ii->GetClient()) {
-      std::vector<std::pair<std::string, unsigned int>> servers;
-      servers.reserve(5);
+    std::vector<std::pair<std::string, unsigned int>> servers;
+    servers.reserve(5);
 
-      // 10.te.am.2
-      servers.emplace_back(
-          fmt::format("10.{}.{}.2", static_cast<int>(team / 100),
-                      static_cast<int>(team % 100)),
-          port);
+    // 10.te.am.2
+    servers.emplace_back(fmt::format("10.{}.{}.2", static_cast<int>(team / 100),
+                                     static_cast<int>(team % 100)),
+                         port);
 
-      // 172.22.11.2
-      servers.emplace_back("172.22.11.2", port);
+    // 172.22.11.2
+    servers.emplace_back("172.22.11.2", port);
 
-      // roboRIO-<team>-FRC.local
-      servers.emplace_back(fmt::format("roboRIO-{}-FRC.local", team), port);
+    // roboRIO-<team>-FRC.local
+    servers.emplace_back(fmt::format("roboRIO-{}-FRC.local", team), port);
 
-      // roboRIO-<team>-FRC.lan
-      servers.emplace_back(fmt::format("roboRIO-{}-FRC.lan", team), port);
+    // roboRIO-<team>-FRC.lan
+    servers.emplace_back(fmt::format("roboRIO-{}-FRC.lan", team), port);
 
-      // roboRIO-<team>-FRC.frc-field.local
-      servers.emplace_back(fmt::format("roboRIO-{}-FRC.frc-field.local", team),
-                           port);
+    // roboRIO-<team>-FRC.frc-field.local
+    servers.emplace_back(fmt::format("roboRIO-{}-FRC.frc-field.local", team),
+                         port);
 
-      client->SetServers(servers);
-    }
+    ii->SetServers(servers);
   }
 }