Team2890/allwpilib
4
0
Fork 0
mirror of https://github.com/wpilibsuite/allwpilib synced 2026-06-27 02:01:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

[ntcore] Change internal interfaces and messages to use UIDs (#7189)

Browse Source 
Also make Handle functions constexpr.
This commit is contained in:
Peter Johnson
2024-10-11 10:57:36 -07:00
committed by GitHub
parent 59bc53b9b8
commit 8ca99c7cb7
26 changed files with 435 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
ntcore/src/main/native/cpp/Handle.h
View File

@@ -33,12 +33,12 @@ class Handle {
static_assert(kTypeMax <= wpi::kHandleTypeHALBase);
enum { kIndexMax = 0xfffff };
explicit Handle(NT_Handle handle) : m_handle(handle) {}
operator NT_Handle() const { return m_handle; }
constexpr explicit Handle(NT_Handle handle) : m_handle(handle) {}
constexpr operator NT_Handle() const { return m_handle; }
NT_Handle handle() const { return m_handle; }
constexpr NT_Handle handle() const { return m_handle; }
Handle(int inst, int index, Type type) {
constexpr Handle(int inst, int index, Type type) {
if (inst < 0 || index < 0) {
m_handle = 0;
return;
@@ -47,16 +47,22 @@ class Handle {
(index & 0xfffff);
}
unsigned int GetIndex() const {
constexpr unsigned int GetIndex() const {
return static_cast<unsigned int>(m_handle) & 0xfffff;
}
Type GetType() const {
constexpr Type GetType() const {
return static_cast<Type>((static_cast<int>(m_handle) >> 24) & 0x7f);
}
int GetInst() const { return (static_cast<int>(m_handle) >> 20) & 0xf; }
bool IsType(Type type) const { return type == GetType(); }
int GetTypedIndex(Type type) const { return IsType(type) ? GetIndex() : -1; }
int GetTypedInst(Type type) const { return IsType(type) ? GetInst() : -1; }
constexpr int GetInst() const {
return (static_cast<int>(m_handle) >> 20) & 0xf;
}
constexpr bool IsType(Type type) const { return type == GetType(); }
constexpr int GetTypedIndex(Type type) const {
return IsType(type) ? GetIndex() : -1;
}
constexpr int GetTypedInst(Type type) const {
return IsType(type) ? GetInst() : -1;
}
private:
NT_Handle m_handle;

46
ntcore/src/main/native/cpp/LocalStorage.cpp
View File

@@ -403,7 +403,7 @@ void LocalStorage::Impl::PropertiesUpdated(TopicData* topic,
NotifyTopic(topic, eventFlags | NT_EVENT_PROPERTIES);
// check local flag so we don't echo back received properties changes
if (m_network && sendNetwork) {
m_network->SetProperties(topic->handle, topic->name, update);
m_network->SetProperties(topic->name, update);
}
}
@@ -427,10 +427,10 @@ void LocalStorage::Impl::RefreshPubSubActive(TopicData* topic,
void LocalStorage::Impl::NetworkAnnounce(TopicData* topic,
std::string_view typeStr,
const wpi::json& properties,
NT_Publisher pubHandle) {
std::optional<int> pubuid) {
DEBUG4("LS NetworkAnnounce({}, {}, {}, {})", topic->name, typeStr,
properties.dump(), pubHandle);
if (pubHandle != 0) {
properties.dump(), pubuid.value_or(-1));
if (pubuid.has_value()) {
return; // ack of our publish; ignore
}
@@ -503,7 +503,7 @@ void LocalStorage::Impl::RemoveNetworkPublisher(TopicData* topic) {
// this may result in a duplicate publish warning on the server side,
// but send one anyway in this case just to be sure
if (nextPub->active && m_network) {
m_network->Publish(nextPub->handle, topic->handle, topic->name,
m_network->Publish(Handle{nextPub->handle}.GetIndex(), topic->name,
topic->typeStr, topic->properties, nextPub->config);
}
}
@@ -561,7 +561,7 @@ LocalStorage::PublisherData* LocalStorage::Impl::AddLocalPublisher(
}
if (publisher->active && m_network) {
m_network->Publish(publisher->handle, topic->handle, topic->name,
m_network->Publish(Handle{publisher->handle}.GetIndex(), topic->name,
topic->typeStr, topic->properties, config);
}
return publisher;
@@ -580,7 +580,7 @@ LocalStorage::Impl::RemoveLocalPublisher(NT_Publisher pubHandle) {
}
if (publisher->active && m_network) {
m_network->Unpublish(publisher->handle, topic->handle);
m_network->Unpublish(Handle{publisher->handle}.GetIndex());
}
if (publisher->active && !topic->localPublishers.empty()) {
@@ -593,7 +593,7 @@ LocalStorage::Impl::RemoveLocalPublisher(NT_Publisher pubHandle) {
topic->typeStr = nextPub->config.typeStr;
RefreshPubSubActive(topic, false);
if (nextPub->active && m_network) {
m_network->Publish(nextPub->handle, topic->handle, topic->name,
m_network->Publish(Handle{nextPub->handle}.GetIndex(), topic->name,
topic->typeStr, topic->properties,
nextPub->config);
}
@@ -619,7 +619,8 @@ LocalStorage::SubscriberData* LocalStorage::Impl::AddLocalSubscriber(
}
if (m_network && !subscriber->config.hidden) {
DEBUG4("-> NetworkSubscribe({})", topic->name);
m_network->Subscribe(subscriber->handle, {{topic->name}}, config);
m_network->Subscribe(Handle{subscriber->handle}.GetIndex(), {{topic->name}},
config);
}
// queue current value
@@ -647,7 +648,7 @@ LocalStorage::Impl::RemoveLocalSubscriber(NT_Subscriber subHandle) {
}
}
if (m_network && !subscriber->config.hidden) {
m_network->Unsubscribe(subscriber->handle);
m_network->Unsubscribe(Handle{subscriber->handle}.GetIndex());
}
}
return subscriber;
@@ -684,8 +685,8 @@ LocalStorage::MultiSubscriberData* LocalStorage::Impl::AddMultiSubscriber(
}
if (m_network && !subscriber->options.hidden) {
DEBUG4("-> NetworkSubscribe");
m_network->Subscribe(subscriber->handle, subscriber->prefixes,
subscriber->options);
m_network->Subscribe(Handle{subscriber->handle}.GetIndex(),
subscriber->prefixes, subscriber->options);
}
return subscriber;
}
@@ -703,7 +704,7 @@ LocalStorage::Impl::RemoveMultiSubscriber(NT_MultiSubscriber subHandle) {
}
}
if (m_network && !subscriber->options.hidden) {
m_network->Unsubscribe(subscriber->handle);
m_network->Unsubscribe(Handle{subscriber->handle}.GetIndex());
}
}
return subscriber;
@@ -977,7 +978,7 @@ bool LocalStorage::Impl::PublishLocalValue(PublisherData* publisher,
if (publisher->topic->IsCached()) {
publisher->topic->lastValueNetwork = value;
}
m_network->SetValue(publisher->handle, value);
m_network->SetValue(Handle{publisher->handle}.GetIndex(), value);
}
return SetValue(publisher->topic, value, NT_EVENT_VALUE_LOCAL,
suppressDuplicates, publisher);
@@ -1076,10 +1077,10 @@ LocalStorage::~LocalStorage() = default;
NT_Topic LocalStorage::NetworkAnnounce(std::string_view name,
std::string_view typeStr,
const wpi::json& properties,
NT_Publisher pubHandle) {
std::optional<int> pubuid) {
std::scoped_lock lock{m_mutex};
auto topic = m_impl.GetOrCreateTopic(name);
m_impl.NetworkAnnounce(topic, typeStr, properties, pubHandle);
m_impl.NetworkAnnounce(topic, typeStr, properties, pubuid);
return topic->handle;
}
@@ -1124,25 +1125,26 @@ void LocalStorage::Impl::StartNetwork(net::NetworkInterface* network) {
PublisherData* anyPublisher = nullptr;
for (auto&& publisher : topic->localPublishers) {
if (publisher->active) {
network->Publish(publisher->handle, topic->handle, topic->name,
network->Publish(Handle{publisher->handle}.GetIndex(), topic->name,
topic->typeStr, topic->properties, publisher->config);
anyPublisher = publisher;
}
}
if (anyPublisher && topic->lastValue) {
network->SetValue(anyPublisher->handle, topic->lastValue);
network->SetValue(Handle{anyPublisher->handle}.GetIndex(),
topic->lastValue);
}
}
for (auto&& subscriber : m_subscribers) {
if (!subscriber->config.hidden) {
network->Subscribe(subscriber->handle, {{subscriber->topic->name}},
subscriber->config);
network->Subscribe(Handle{subscriber->handle}.GetIndex(),
{{subscriber->topic->name}}, subscriber->config);
}
}
for (auto&& subscriber : m_multiSubscribers) {
if (!subscriber->options.hidden) {
network->Subscribe(subscriber->handle, subscriber->prefixes,
subscriber->options);
network->Subscribe(Handle{subscriber->handle}.GetIndex(),
subscriber->prefixes, subscriber->options);
}
}
}

5
ntcore/src/main/native/cpp/LocalStorage.h
View File

@@ -48,7 +48,7 @@ class LocalStorage final : public net::ILocalStorage {
// network interface functions
NT_Topic NetworkAnnounce(std::string_view name, std::string_view typeStr,
const wpi::json& properties,
NT_Publisher pubHandle) final;
std::optional<int> pubuid) final;
void NetworkUnannounce(std::string_view name) final;
void NetworkPropertiesUpdate(std::string_view name, const wpi::json& update,
bool ack) final;
@@ -601,7 +601,8 @@ class LocalStorage final : public net::ILocalStorage {
void RefreshPubSubActive(TopicData* topic, bool warnOnSubMismatch);
void NetworkAnnounce(TopicData* topic, std::string_view typeStr,
const wpi::json& properties, NT_Publisher pubHandle);
const wpi::json& properties,
std::optional<int> pubuid);
void RemoveNetworkPublisher(TopicData* topic);
void NetworkPropertiesUpdate(TopicData* topic, const wpi::json& update,
bool ack);

2
ntcore/src/main/native/cpp/NetworkClient.cpp
View File

@@ -407,7 +407,7 @@ void NetworkClient::WsConnected(wpi::WebSocket& ws, uv::Tcp& tcp,
m_wire = std::make_shared<net::WebSocketConnection>(
ws, connInfo.protocol_version, m_logger);
m_clientImpl = std::make_unique<net::ClientImpl>(
m_loop.Now().count(), m_inst, *m_wire, m_logger, m_timeSyncUpdated,
m_loop.Now().count(), *m_wire, m_logger, m_timeSyncUpdated,
[this](uint32_t repeatMs) {
DEBUG4("Setting periodic timer to {}", repeatMs);
if (m_sendOutgoingTimer &&

2
ntcore/src/main/native/cpp/NetworkServer.cpp
View File

@@ -254,7 +254,7 @@ void NetworkServer::ServerConnection4::ProcessWsUpgrade() {
m_websocket->binary.connect([this](std::span<const uint8_t> data, bool) {
while (!data.empty()) {
// decode message
int64_t pubuid;
int pubuid;
Value value;
std::string error;
if (!net::WireDecodeBinary(&data, &pubuid, &value, &error, 0)) {

69
ntcore/src/main/native/cpp/net/ClientImpl.cpp
View File

@@ -28,12 +28,11 @@ using namespace nt;
using namespace nt::net;
ClientImpl::ClientImpl(
uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
uint64_t curTimeMs, 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_inst{inst},
m_wire{wire},
: m_wire{wire},
m_logger{logger},
m_timeSyncUpdated{std::move(timeSyncUpdated)},
m_setPeriodic{std::move(setPeriodic)},
@@ -58,7 +57,7 @@ void ClientImpl::ProcessIncomingBinary(uint64_t curTimeMs,
}
// decode message
int64_t id;
int id;
Value value;
std::string error;
if (!WireDecodeBinary(&data, &id, &value, &error,
@@ -114,13 +113,13 @@ void ClientImpl::HandleLocal(std::vector<ClientMessage>&& msgs) {
for (auto&& elem : msgs) {
// common case is value
if (auto msg = std::get_if<ClientValueMsg>(&elem.contents)) {
SetValue(msg->pubHandle, msg->value);
SetValue(msg->pubuid, msg->value);
} else if (auto msg = std::get_if<PublishMsg>(&elem.contents)) {
Publish(msg->pubHandle, msg->topicHandle, msg->name, msg->typeStr,
msg->properties, msg->options);
m_outgoing.SendMessage(msg->pubHandle, std::move(elem));
Publish(msg->pubuid, msg->name, msg->typeStr, msg->properties,
msg->options);
m_outgoing.SendMessage(msg->pubuid, std::move(elem));
} else if (auto msg = std::get_if<UnpublishMsg>(&elem.contents)) {
Unpublish(msg->pubHandle, msg->topicHandle, std::move(elem));
Unpublish(msg->pubuid, std::move(elem));
} else {
m_outgoing.SendMessage(0, std::move(elem));
}
@@ -174,38 +173,33 @@ void ClientImpl::UpdatePeriodic() {
m_setPeriodic(m_periodMs);
}
void ClientImpl::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
std::string_view name, std::string_view typeStr,
const wpi::json& properties,
void ClientImpl::Publish(int32_t pubuid, 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);
if (static_cast<uint32_t>(pubuid) >= m_publishers.size()) {
m_publishers.resize(pubuid + 1);
}
auto& publisher = m_publishers[index];
auto& publisher = m_publishers[pubuid];
if (!publisher) {
publisher = std::make_unique<PublisherData>();
}
publisher->handle = pubHandle;
publisher->options = options;
publisher->periodMs = std::lround(options.periodicMs / 10.0) * 10;
if (publisher->periodMs < kMinPeriodMs) {
publisher->periodMs = kMinPeriodMs;
}
m_outgoing.SetPeriod(pubHandle, publisher->periodMs);
m_outgoing.SetPeriod(pubuid, publisher->periodMs);
// update period
m_periodMs = UpdatePeriodCalc(m_periodMs, publisher->periodMs);
UpdatePeriodic();
}
void ClientImpl::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle,
ClientMessage&& msg) {
unsigned int index = Handle{pubHandle}.GetIndex();
if (index >= m_publishers.size()) {
void ClientImpl::Unpublish(int32_t pubuid, ClientMessage&& msg) {
if (static_cast<uint32_t>(pubuid) >= m_publishers.size()) {
return;
}
m_publishers[index].reset();
m_publishers[pubuid].reset();
// loop over all publishers to update period
m_periodMs = kMaxPeriodMs;
@@ -216,40 +210,35 @@ void ClientImpl::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle,
}
UpdatePeriodic();
m_outgoing.SendMessage(pubHandle, std::move(msg));
m_outgoing.SendMessage(pubuid, std::move(msg));
// remove from outgoing handle map
m_outgoing.EraseHandle(pubHandle);
m_outgoing.EraseId(pubuid);
}
void ClientImpl::SetValue(NT_Publisher pubHandle, const Value& value) {
DEBUG4("SetValue({}, time={}, server_time={})", pubHandle, value.time(),
void ClientImpl::SetValue(int32_t pubuid, const Value& value) {
DEBUG4("SetValue({}, time={}, server_time={})", pubuid, value.time(),
value.server_time());
unsigned int index = Handle{pubHandle}.GetIndex();
if (index >= m_publishers.size() || !m_publishers[index]) {
if (static_cast<uint32_t>(pubuid) >= m_publishers.size() ||
!m_publishers[pubuid]) {
return;
}
auto& publisher = *m_publishers[index];
auto& publisher = *m_publishers[pubuid];
m_outgoing.SendValue(
pubHandle, value,
pubuid, value,
publisher.options.sendAll ? ValueSendMode::kAll : ValueSendMode::kNormal);
}
void ClientImpl::ServerAnnounce(std::string_view name, int64_t id,
void ClientImpl::ServerAnnounce(std::string_view name, int id,
std::string_view typeStr,
const wpi::json& properties,
std::optional<int64_t> pubuid) {
std::optional<int> pubuid) {
DEBUG4("ServerAnnounce({}, {}, {})", name, id, typeStr);
assert(m_local);
NT_Publisher pubHandle{0};
if (pubuid) {
pubHandle = Handle(m_inst, pubuid.value(), Handle::kPublisher);
}
m_topicMap[id] =
m_local->NetworkAnnounce(name, typeStr, properties, pubHandle);
m_topicMap[id] = m_local->NetworkAnnounce(name, typeStr, properties, pubuid);
}
void ClientImpl::ServerUnannounce(std::string_view name, int64_t id) {
void ClientImpl::ServerUnannounce(std::string_view name, int id) {
DEBUG4("ServerUnannounce({}, {})", name, id);
assert(m_local);
m_local->NetworkUnannounce(name);

23
ntcore/src/main/native/cpp/net/ClientImpl.h
View File

@@ -9,7 +9,6 @@
#include <functional>
#include <memory>
#include <span>
#include <string>
#include <string_view>
#include <vector>
@@ -39,7 +38,7 @@ class WireConnection;
class ClientImpl final : private ServerMessageHandler {
public:
ClientImpl(
uint64_t curTimeMs, int inst, WireConnection& wire, wpi::Logger& logger,
uint64_t curTimeMs, WireConnection& wire, wpi::Logger& logger,
std::function<void(int64_t serverTimeOffset, int64_t rtt2, bool valid)>
timeSyncUpdated,
std::function<void(uint32_t repeatMs)> setPeriodic);
@@ -55,7 +54,6 @@ class ClientImpl final : private ServerMessageHandler {
private:
struct PublisherData {
NT_Publisher handle;
PubSubOptionsImpl options;
// in options as double, but copy here as integer; rounded to the nearest
// 10 ms
@@ -65,21 +63,18 @@ class ClientImpl final : private ServerMessageHandler {
void UpdatePeriodic();
// 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 ServerAnnounce(std::string_view name, int id, std::string_view typeStr,
const wpi::json& properties,
std::optional<int> pubuid) final;
void ServerUnannounce(std::string_view name, int 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,
void Publish(int pubuid, std::string_view name, std::string_view typeStr,
const wpi::json& properties, const PubSubOptionsImpl& options);
void Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle,
ClientMessage&& msg);
void SetValue(NT_Publisher pubHandle, const Value& value);
void Unpublish(int pubuid, ClientMessage&& msg);
void SetValue(int pubuid, const Value& value);
int m_inst;
WireConnection& m_wire;
wpi::Logger& m_logger;
LocalInterface* m_local{nullptr};
@@ -91,7 +86,7 @@ class ClientImpl final : private ServerMessageHandler {
std::vector<std::unique_ptr<PublisherData>> m_publishers;
// indexed by server-provided topic id
wpi::DenseMap<int64_t, NT_Topic> m_topicMap;
wpi::DenseMap<int, NT_Topic> m_topicMap;
// ping
NetworkPing m_ping;

22
ntcore/src/main/native/cpp/net/Message.h
View File

@@ -13,7 +13,6 @@
#include "PubSubOptions.h"
#include "networktables/NetworkTableValue.h"
#include "ntcore_c.h"
namespace nt::net {
@@ -24,8 +23,7 @@ namespace nt::net {
struct PublishMsg {
static constexpr std::string_view kMethodStr = "publish";
NT_Publisher pubHandle{0};
NT_Topic topicHandle{0}; // will be 0 when coming from network
int pubuid{0};
std::string name;
std::string typeStr;
wpi::json properties;
@@ -34,31 +32,29 @@ struct PublishMsg {
struct UnpublishMsg {
static constexpr std::string_view kMethodStr = "unpublish";
NT_Publisher pubHandle{0};
NT_Topic topicHandle{0}; // will be 0 when coming from network
int pubuid{0};
};
struct SetPropertiesMsg {
static constexpr std::string_view kMethodStr = "setproperties";
NT_Topic topicHandle{0}; // will be 0 when coming from network
std::string name;
wpi::json update;
};
struct SubscribeMsg {
static constexpr std::string_view kMethodStr = "subscribe";
NT_Subscriber subHandle{0};
int subuid{0};
std::vector<std::string> topicNames;
PubSubOptionsImpl options;
};
struct UnsubscribeMsg {
static constexpr std::string_view kMethodStr = "unsubscribe";
NT_Subscriber subHandle{0};
int subuid{0};
};
struct ClientValueMsg {
NT_Publisher pubHandle{0};
int pubuid{0};
Value value;
};
@@ -77,16 +73,16 @@ struct ClientMessage {
struct AnnounceMsg {
static constexpr std::string_view kMethodStr = "announce";
std::string name;
int64_t id{0};
int id{0};
std::string typeStr;
std::optional<int64_t> pubuid;
std::optional<int> pubuid;
wpi::json properties;
};
struct UnannounceMsg {
static constexpr std::string_view kMethodStr = "unannounce";
std::string name;
int64_t id{0};
int id{0};
};
struct PropertiesUpdateMsg {
@@ -97,7 +93,7 @@ struct PropertiesUpdateMsg {
};
struct ServerValueMsg {
NT_Topic topic{0};
int topic{0};
Value value;
};

20
ntcore/src/main/native/cpp/net/NetworkInterface.h
View File

@@ -4,6 +4,8 @@
#pragma once
#include <stdint.h>
#include <span>
#include <string>
#include <string_view>
@@ -26,7 +28,7 @@ class LocalInterface {
virtual NT_Topic NetworkAnnounce(std::string_view name,
std::string_view typeStr,
const wpi::json& properties,
NT_Publisher pubHandle) = 0;
std::optional<int> pubuid) = 0;
virtual void NetworkUnannounce(std::string_view name) = 0;
virtual void NetworkPropertiesUpdate(std::string_view name,
const wpi::json& update, bool ack) = 0;
@@ -37,18 +39,16 @@ class NetworkInterface {
public:
virtual ~NetworkInterface() = default;
virtual void Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
std::string_view name, std::string_view typeStr,
const wpi::json& properties,
virtual void Publish(int pubuid, std::string_view name,
std::string_view typeStr, const wpi::json& properties,
const PubSubOptionsImpl& options) = 0;
virtual void Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) = 0;
virtual void SetProperties(NT_Topic topicHandle, std::string_view name,
virtual void Unpublish(int pubuid) = 0;
virtual void SetProperties(std::string_view name,
const wpi::json& update) = 0;
virtual void Subscribe(NT_Subscriber subHandle,
std::span<const std::string> topicNames,
virtual void Subscribe(int subuid, std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) = 0;
virtual void Unsubscribe(NT_Subscriber subHandle) = 0;
virtual void SetValue(NT_Publisher pubHandle, const Value& value) = 0;
virtual void Unsubscribe(int subuid) = 0;
virtual void SetValue(int pubuid, const Value& value) = 0;
};
class ILocalStorage : public LocalInterface {

4
ntcore/src/main/native/cpp/net/NetworkLoopQueue.cpp
View File

@@ -10,7 +10,7 @@ using namespace nt::net;
static constexpr size_t kMaxSize = 2 * 1024 * 1024;
void NetworkLoopQueue::SetValue(NT_Publisher pubHandle, const Value& value) {
void NetworkLoopQueue::SetValue(int pubuid, const Value& value) {
std::scoped_lock lock{m_mutex};
m_size += sizeof(ClientMessage) + value.size();
if (m_size > kMaxSize) {
@@ -20,5 +20,5 @@ void NetworkLoopQueue::SetValue(NT_Publisher pubHandle, const Value& value) {
}
return; // avoid potential out of memory
}
m_queue.emplace_back(ClientMessage{ClientValueMsg{pubHandle, value}});
m_queue.emplace_back(ClientMessage{ClientValueMsg{pubuid, value}});
}

15
ntcore/src/main/native/cpp/net/NetworkLoopQueue.h
View File

@@ -31,18 +31,15 @@ class NetworkLoopQueue : public NetworkInterface {
void ClearQueue();
// NetworkInterface - calls to these append to the queue
void Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
std::string_view name, std::string_view typeStr,
void Publish(int pubuid, std::string_view name, std::string_view typeStr,
const wpi::json& properties,
const PubSubOptionsImpl& options) final;
void Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) final;
void SetProperties(NT_Topic topicHandle, std::string_view name,
const wpi::json& update) final;
void Subscribe(NT_Subscriber subHandle,
std::span<const std::string> topicNames,
void Unpublish(int pubuid) final;
void SetProperties(std::string_view name, const wpi::json& update) final;
void Subscribe(int subuid, std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) final;
void Unsubscribe(NT_Subscriber subHandle) final;
void SetValue(NT_Publisher pubHandle, const Value& value) final;
void Unsubscribe(int subuid) final;
void SetValue(int pubuid, const Value& value) final;
private:
wpi::mutex m_mutex;

29
ntcore/src/main/native/cpp/net/NetworkLoopQueue.inc
View File

@@ -29,43 +29,38 @@ inline void NetworkLoopQueue::ClearQueue() {
m_sizeErrored = false;
}
inline void NetworkLoopQueue::Publish(NT_Publisher pubHandle,
NT_Topic topicHandle,
std::string_view name,
inline void NetworkLoopQueue::Publish(int pubuid, std::string_view name,
std::string_view typeStr,
const wpi::json& properties,
const PubSubOptionsImpl& options) {
std::scoped_lock lock{m_mutex};
m_queue.emplace_back(
ClientMessage{PublishMsg{pubHandle, topicHandle, std::string{name},
std::string{typeStr}, properties, options}});
m_queue.emplace_back(ClientMessage{PublishMsg{
pubuid, std::string{name}, std::string{typeStr}, properties, options}});
}
inline void NetworkLoopQueue::Unpublish(NT_Publisher pubHandle,
NT_Topic topicHandle) {
inline void NetworkLoopQueue::Unpublish(int pubuid) {
std::scoped_lock lock{m_mutex};
m_queue.emplace_back(ClientMessage{UnpublishMsg{pubHandle, topicHandle}});
m_queue.emplace_back(ClientMessage{UnpublishMsg{pubuid}});
}
inline void NetworkLoopQueue::SetProperties(NT_Topic topicHandle,
std::string_view name,
inline void NetworkLoopQueue::SetProperties(std::string_view name,
const wpi::json& update) {
std::scoped_lock lock{m_mutex};
m_queue.emplace_back(
ClientMessage{SetPropertiesMsg{topicHandle, std::string{name}, update}});
ClientMessage{SetPropertiesMsg{std::string{name}, update}});
}
inline void NetworkLoopQueue::Subscribe(NT_Subscriber subHandle,
inline void NetworkLoopQueue::Subscribe(int subuid,
std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) {
std::scoped_lock lock{m_mutex};
m_queue.emplace_back(ClientMessage{SubscribeMsg{
subHandle, {topicNames.begin(), topicNames.end()}, options}});
m_queue.emplace_back(ClientMessage{
SubscribeMsg{subuid, {topicNames.begin(), topicNames.end()}, options}});
}
inline void NetworkLoopQueue::Unsubscribe(NT_Subscriber subHandle) {
inline void NetworkLoopQueue::Unsubscribe(int subuid) {
std::scoped_lock lock{m_mutex};
m_queue.emplace_back(ClientMessage{UnsubscribeMsg{subHandle}});
m_queue.emplace_back(ClientMessage{UnsubscribeMsg{subuid}});
}
} // namespace nt::net

55
ntcore/src/main/native/cpp/net/NetworkOutgoingQueue.h
View File

@@ -9,14 +9,12 @@
#include <algorithm>
#include <concepts>
#include <numeric>
#include <optional>
#include <span>
#include <utility>
#include <vector>
#include <wpi/DenseMap.h>
#include "Handle.h"
#include "Message.h"
#include "WireConnection.h"
#include "WireEncoder.h"
@@ -71,18 +69,17 @@ class NetworkOutgoingQueue {
m_queues.emplace_back(100); // default queue is 100 ms period
}
void SetPeriod(NT_Handle handle, uint32_t periodMs);
void SetPeriod(int id, uint32_t periodMs);
void EraseHandle(NT_Handle handle) { m_handleMap.erase(handle); }
void EraseId(int id) { m_idMap.erase(id); }
template <typename T>
void SendMessage(NT_Handle handle, T&& msg) {
m_queues[m_handleMap[handle].queueIndex].Append(handle,
std::forward<T>(msg));
void SendMessage(int id, T&& msg) {
m_queues[m_idMap[id].queueIndex].Append(id, std::forward<T>(msg));
m_totalSize += sizeof(Message);
}
void SendValue(NT_Handle handle, const Value& value, ValueSendMode mode);
void SendValue(int id, const Value& value, ValueSendMode mode);
void SendOutgoing(uint64_t curTimeMs, bool flush);
@@ -95,16 +92,15 @@ class NetworkOutgoingQueue {
private:
using ValueMsg = typename MessageType::ValueMsg;
void EncodeValue(wpi::raw_ostream& os, NT_Handle handle, const Value& value);
void EncodeValue(wpi::raw_ostream& os, int id, const Value& value);
struct Message {
Message() = default;
template <typename T>
Message(T&& msg, NT_Handle handle)
: msg{std::forward<T>(msg)}, handle{handle} {}
Message(T&& msg, int id) : msg{std::forward<T>(msg)}, id{id} {}
MessageType msg;
NT_Handle handle;
int id;
};
struct Queue {
@@ -124,7 +120,7 @@ class NetworkOutgoingQueue {
unsigned int queueIndex = 0;
int valuePos = -1; // -1 if not in queue
};
wpi::DenseMap<NT_Handle, HandleInfo> m_handleMap;
wpi::DenseMap<int, HandleInfo> m_idMap;
size_t m_totalSize{0};
uint64_t m_lastSendMs{0};
int64_t m_timeOffsetUs{0};
@@ -137,8 +133,7 @@ class NetworkOutgoingQueue {
};
template <NetworkMessage MessageType>
void NetworkOutgoingQueue<MessageType>::SetPeriod(NT_Handle handle,
uint32_t periodMs) {
void NetworkOutgoingQueue<MessageType>::SetPeriod(int id, uint32_t periodMs) {
// it's quite common to set a lot of things in a row with the same period
unsigned int queueIndex;
if (m_lastSetPeriod == periodMs) {
@@ -159,13 +154,12 @@ void NetworkOutgoingQueue<MessageType>::SetPeriod(NT_Handle handle,
}
// map the handle to the queue
auto [infoIt, created] = m_handleMap.try_emplace(handle);
auto [infoIt, created] = m_idMap.try_emplace(id);
if (!created && infoIt->getSecond().queueIndex != queueIndex) {
// need to move any items from old queue to new queue
auto& oldMsgs = m_queues[infoIt->getSecond().queueIndex].msgs;
auto it = std::stable_partition(
oldMsgs.begin(), oldMsgs.end(),
[&](const auto& e) { return e.handle != handle; });
auto it = std::stable_partition(oldMsgs.begin(), oldMsgs.end(),
[&](const auto& e) { return e.id != id; });
auto& newMsgs = m_queues[queueIndex].msgs;
for (auto i = it, end = oldMsgs.end(); i != end; ++i) {
newMsgs.emplace_back(std::move(*i));
@@ -177,8 +171,7 @@ void NetworkOutgoingQueue<MessageType>::SetPeriod(NT_Handle handle,
}
template <NetworkMessage MessageType>
void NetworkOutgoingQueue<MessageType>::SendValue(NT_Handle handle,
const Value& value,
void NetworkOutgoingQueue<MessageType>::SendValue(int id, const Value& value,
ValueSendMode mode) {
if (m_local) {
mode = ValueSendMode::kImm; // always send local immediately
@@ -191,26 +184,26 @@ void NetworkOutgoingQueue<MessageType>::SendValue(NT_Handle handle,
case ValueSendMode::kDisabled: // do nothing
break;
case ValueSendMode::kImm: // send immediately
m_wire.SendBinary([&](auto& os) { EncodeValue(os, handle, value); });
m_wire.SendBinary([&](auto& os) { EncodeValue(os, id, value); });
break;
case ValueSendMode::kAll: { // append to outgoing
auto& info = m_handleMap[handle];
auto& info = m_idMap[id];
auto& queue = m_queues[info.queueIndex];
info.valuePos = queue.msgs.size();
queue.Append(handle, ValueMsg{handle, value});
queue.Append(id, ValueMsg{id, value});
m_totalSize += sizeof(Message) + value.size();
break;
}
case ValueSendMode::kNormal: {
// replace, or append if not present
auto& info = m_handleMap[handle];
auto& info = m_idMap[id];
auto& queue = m_queues[info.queueIndex];
if (info.valuePos != -1 &&
static_cast<unsigned int>(info.valuePos) < queue.msgs.size()) {
auto& elem = queue.msgs[info.valuePos];
if (auto m = std::get_if<ValueMsg>(&elem.msg.contents)) {
// double-check handle, and only replace if timestamp newer
if (elem.handle == handle &&
if (elem.id == id &&
(m->value.time() == 0 || value.time() >= m->value.time())) {
int delta = value.size() - m->value.size();
m->value = value;
@@ -220,7 +213,7 @@ void NetworkOutgoingQueue<MessageType>::SendValue(NT_Handle handle,
}
}
info.valuePos = queue.msgs.size();
queue.Append(handle, ValueMsg{handle, value});
queue.Append(id, ValueMsg{id, value});
m_totalSize += sizeof(Message) + value.size();
break;
}
@@ -271,7 +264,7 @@ void NetworkOutgoingQueue<MessageType>::SendOutgoing(uint64_t curTimeMs,
for (; it != end && unsent == 0; ++it) {
if (auto m = std::get_if<ValueMsg>(&it->msg.contents)) {
unsent = m_wire.WriteBinary(
[&](auto& os) { EncodeValue(os, it->handle, m->value); });
[&](auto& os) { EncodeValue(os, it->id, m->value); });
} else {
unsent = m_wire.WriteText([&](auto& os) {
if (!WireEncodeText(os, it->msg)) {
@@ -299,7 +292,7 @@ void NetworkOutgoingQueue<MessageType>::SendOutgoing(uint64_t curTimeMs,
}
}
msgs.erase(msgs.begin(), it - unsent);
for (auto&& kv : m_handleMap) {
for (auto&& kv : m_idMap) {
auto& info = kv.getSecond();
if (info.queueIndex == queueIndex) {
if (info.valuePos < delta) {
@@ -324,7 +317,7 @@ void NetworkOutgoingQueue<MessageType>::SendOutgoing(uint64_t curTimeMs,
template <NetworkMessage MessageType>
void NetworkOutgoingQueue<MessageType>::EncodeValue(wpi::raw_ostream& os,
NT_Handle handle,
int id,
const Value& value) {
int64_t time = value.time();
if constexpr (std::same_as<ValueMsg, ClientValueMsg>) {
@@ -336,7 +329,7 @@ void NetworkOutgoingQueue<MessageType>::EncodeValue(wpi::raw_ostream& os,
}
}
}
WireEncodeBinary(os, Handle{handle}.GetIndex(), time, value);
WireEncodeBinary(os, id, time, value);
}
} // namespace nt::net

48
ntcore/src/main/native/cpp/net/ServerImpl.cpp
View File

@@ -199,7 +199,7 @@ std::span<ServerImpl::SubscriberData*> ServerImpl::ClientData::GetSubscribers(
return {buf.data(), buf.size()};
}
void ServerImpl::ClientData4Base::ClientPublish(int64_t pubuid,
void ServerImpl::ClientData4Base::ClientPublish(int pubuid,
std::string_view name,
std::string_view typeStr,
const wpi::json& properties) {
@@ -224,7 +224,7 @@ void ServerImpl::ClientData4Base::ClientPublish(int64_t pubuid,
SendAnnounce(topic, pubuid);
}
void ServerImpl::ClientData4Base::ClientUnpublish(int64_t pubuid) {
void ServerImpl::ClientData4Base::ClientUnpublish(int pubuid) {
DEBUG3("ClientUnpublish({}, {})", m_id, pubuid);
auto publisherIt = m_publishers.find(pubuid);
if (publisherIt == m_publishers.end()) {
@@ -270,7 +270,7 @@ void ServerImpl::ClientData4Base::ClientSetProperties(std::string_view name,
}
void ServerImpl::ClientData4Base::ClientSubscribe(
int64_t subuid, std::span<const std::string> topicNames,
int subuid, std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) {
DEBUG4("ClientSubscribe({}, ({}), {})", m_id, fmt::join(topicNames, ","),
subuid);
@@ -347,7 +347,7 @@ void ServerImpl::ClientData4Base::ClientSubscribe(
}
}
void ServerImpl::ClientData4Base::ClientUnsubscribe(int64_t subuid) {
void ServerImpl::ClientData4Base::ClientUnsubscribe(int subuid) {
DEBUG3("ClientUnsubscribe({}, {})", m_id, subuid);
auto subIt = m_subscribers.find(subuid);
if (subIt == m_subscribers.end() || !subIt->getSecond()) {
@@ -377,7 +377,7 @@ void ServerImpl::ClientData4Base::ClientUnsubscribe(int64_t subuid) {
}
}
void ServerImpl::ClientData4Base::ClientSetValue(int64_t pubuid,
void ServerImpl::ClientData4Base::ClientSetValue(int pubuid,
const Value& value) {
DEBUG4("ClientSetValue({}, {})", m_id, pubuid);
auto publisherIt = m_publishers.find(pubuid);
@@ -398,7 +398,7 @@ void ServerImpl::ClientDataLocal::SendValue(TopicData* topic,
}
void ServerImpl::ClientDataLocal::SendAnnounce(TopicData* topic,
std::optional<int64_t> pubuid) {
std::optional<int> pubuid) {
if (m_server.m_local) {
auto& sent = m_announceSent[topic];
if (sent) {
@@ -407,7 +407,7 @@ void ServerImpl::ClientDataLocal::SendAnnounce(TopicData* topic,
sent = true;
topic->localHandle = m_server.m_local->NetworkAnnounce(
topic->name, topic->typeStr, topic->properties, pubuid.value_or(0));
topic->name, topic->typeStr, topic->properties, pubuid);
}
}
@@ -445,20 +445,20 @@ void ServerImpl::ClientDataLocal::HandleLocal(
for (const auto& elem : msgs) { // NOLINT
// common case is value, so check that first
if (auto msg = std::get_if<ClientValueMsg>(&elem.contents)) {
ClientSetValue(msg->pubHandle, msg->value);
ClientSetValue(msg->pubuid, msg->value);
} else if (auto msg = std::get_if<PublishMsg>(&elem.contents)) {
ClientPublish(msg->pubHandle, msg->name, msg->typeStr, msg->properties);
ClientPublish(msg->pubuid, msg->name, msg->typeStr, msg->properties);
updatepub = true;
} else if (auto msg = std::get_if<UnpublishMsg>(&elem.contents)) {
ClientUnpublish(msg->pubHandle);
ClientUnpublish(msg->pubuid);
updatepub = true;
} else if (auto msg = std::get_if<SetPropertiesMsg>(&elem.contents)) {
ClientSetProperties(msg->name, msg->update);
} else if (auto msg = std::get_if<SubscribeMsg>(&elem.contents)) {
ClientSubscribe(msg->subHandle, msg->topicNames, msg->options);
ClientSubscribe(msg->subuid, msg->topicNames, msg->options);
updatesub = true;
} else if (auto msg = std::get_if<UnsubscribeMsg>(&elem.contents)) {
ClientUnsubscribe(msg->subHandle);
ClientUnsubscribe(msg->subuid);
updatesub = true;
}
}
@@ -485,7 +485,7 @@ void ServerImpl::ClientData4::ProcessIncomingBinary(
}
// decode message
int64_t pubuid;
int pubuid;
Value value;
std::string error;
if (!WireDecodeBinary(&data, &pubuid, &value, &error, 0)) {
@@ -509,11 +509,11 @@ void ServerImpl::ClientData4::ProcessIncomingBinary(
void ServerImpl::ClientData4::SendValue(TopicData* topic, const Value& value,
ValueSendMode mode) {
m_outgoing.SendValue(topic->GetIdHandle(), value, mode);
m_outgoing.SendValue(topic->id, value, mode);
}
void ServerImpl::ClientData4::SendAnnounce(TopicData* topic,
std::optional<int64_t> pubuid) {
std::optional<int> pubuid) {
auto& sent = m_announceSent[topic];
if (sent) {
return;
@@ -532,9 +532,9 @@ void ServerImpl::ClientData4::SendAnnounce(TopicData* topic,
return;
}
}
m_outgoing.SendMessage(topic->GetIdHandle(),
AnnounceMsg{topic->name, topic->id, topic->typeStr,
pubuid, topic->properties});
m_outgoing.SendMessage(
topic->id, AnnounceMsg{topic->name, static_cast<int>(topic->id),
topic->typeStr, pubuid, topic->properties});
m_server.m_controlReady = true;
}
@@ -555,9 +555,9 @@ void ServerImpl::ClientData4::SendUnannounce(TopicData* topic) {
return;
}
}
m_outgoing.SendMessage(topic->GetIdHandle(),
UnannounceMsg{topic->name, topic->id});
m_outgoing.EraseHandle(topic->GetIdHandle());
m_outgoing.SendMessage(
topic->id, UnannounceMsg{topic->name, static_cast<int>(topic->id)});
m_outgoing.EraseId(topic->id);
m_server.m_controlReady = true;
}
@@ -579,7 +579,7 @@ void ServerImpl::ClientData4::SendPropertiesUpdate(TopicData* topic,
return;
}
}
m_outgoing.SendMessage(topic->GetIdHandle(),
m_outgoing.SendMessage(topic->id,
PropertiesUpdateMsg{topic->name, update, ack});
m_server.m_controlReady = true;
}
@@ -598,7 +598,7 @@ void ServerImpl::ClientData4::UpdatePeriod(TopicData::TopicClientData& tcd,
uint32_t period =
CalculatePeriod(tcd.subscribers, [](auto& x) { return x->periodMs; });
DEBUG4("updating {} period to {} ms", topic->name, period);
m_outgoing.SetPeriod(topic->GetIdHandle(), period);
m_outgoing.SetPeriod(topic->id, period);
}
bool ServerImpl::ClientData3::TopicData3::UpdateFlags(TopicData* topic) {
@@ -681,7 +681,7 @@ void ServerImpl::ClientData3::SendValue(TopicData* topic, const Value& value,
}
void ServerImpl::ClientData3::SendAnnounce(TopicData* topic,
std::optional<int64_t> pubuid) {
std::optional<int> pubuid) {
// ignore if we've not yet built the subscriber
if (m_subscribers.empty()) {
return;

28
ntcore/src/main/native/cpp/net/ServerImpl.h
View File

@@ -21,14 +21,12 @@
#include <wpi/UidVector.h>
#include <wpi/json.h>
#include "Handle.h"
#include "Log.h"
#include "Message.h"
#include "NetworkInterface.h"
#include "NetworkOutgoingQueue.h"
#include "NetworkPing.h"
#include "PubSubOptions.h"
#include "VectorSet.h"
#include "WireConnection.h"
#include "WireDecoder.h"
#include "WireEncoder.h"
@@ -119,8 +117,6 @@ class ServerImpl final {
void RefreshProperties();
bool SetFlags(unsigned int flags_);
NT_Handle GetIdHandle() const { return Handle(0, id, Handle::kTopic); }
wpi::Logger& m_logger; // Must be m_logger for WARN macro to work
std::string name;
unsigned int id;
@@ -190,8 +186,7 @@ class ServerImpl final {
virtual void SendValue(TopicData* topic, const Value& value,
ValueSendMode mode) = 0;
virtual void SendAnnounce(TopicData* topic,
std::optional<int64_t> pubuid) = 0;
virtual void SendAnnounce(TopicData* topic, std::optional<int> pubuid) = 0;
virtual void SendUnannounce(TopicData* topic) = 0;
virtual void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
bool ack) = 0;
@@ -223,8 +218,8 @@ class ServerImpl final {
wpi::Logger& m_logger;
wpi::DenseMap<int64_t, std::unique_ptr<PublisherData>> m_publishers;
wpi::DenseMap<int64_t, std::unique_ptr<SubscriberData>> m_subscribers;
wpi::DenseMap<int, std::unique_ptr<PublisherData>> m_publishers;
wpi::DenseMap<int, std::unique_ptr<SubscriberData>> m_subscribers;
public:
// meta topics
@@ -241,18 +236,17 @@ class ServerImpl final {
protected:
// ClientMessageHandler interface
void ClientPublish(int64_t pubuid, std::string_view name,
void ClientPublish(int pubuid, std::string_view name,
std::string_view typeStr,
const wpi::json& properties) final;
void ClientUnpublish(int64_t pubuid) final;
void ClientUnpublish(int pubuid) final;
void ClientSetProperties(std::string_view name,
const wpi::json& update) final;
void ClientSubscribe(int64_t subuid,
std::span<const std::string> topicNames,
void ClientSubscribe(int subuid, std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) final;
void ClientUnsubscribe(int64_t subuid) final;
void ClientUnsubscribe(int subuid) final;
void ClientSetValue(int64_t pubuid, const Value& value);
void ClientSetValue(int pubuid, const Value& value);
wpi::DenseMap<TopicData*, bool> m_announceSent;
};
@@ -267,7 +261,7 @@ class ServerImpl final {
void SendValue(TopicData* topic, const Value& value,
ValueSendMode mode) final;
void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
void SendAnnounce(TopicData* topic, std::optional<int> pubuid) final;
void SendUnannounce(TopicData* topic) final;
void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
bool ack) final;
@@ -293,7 +287,7 @@ class ServerImpl final {
void SendValue(TopicData* topic, const Value& value,
ValueSendMode mode) final;
void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
void SendAnnounce(TopicData* topic, std::optional<int> pubuid) final;
void SendUnannounce(TopicData* topic) final;
void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
bool ack) final;
@@ -328,7 +322,7 @@ class ServerImpl final {
void SendValue(TopicData* topic, const Value& value,
ValueSendMode mode) final;
void SendAnnounce(TopicData* topic, std::optional<int64_t> pubuid) final;
void SendAnnounce(TopicData* topic, std::optional<int> pubuid) final;
void SendUnannounce(TopicData* topic) final;
void SendPropertiesUpdate(TopicData* topic, const wpi::json& update,
bool ack) final;

4
ntcore/src/main/native/cpp/net/WireDecoder.cpp
View File

@@ -430,14 +430,14 @@ void nt::net::WireDecodeText(std::string_view in, ServerMessageHandler& out,
::WireDecodeTextImpl(in, out, logger);
}
bool nt::net::WireDecodeBinary(std::span<const uint8_t>* in, int64_t* outId,
bool nt::net::WireDecodeBinary(std::span<const uint8_t>* in, int* outId,
Value* outValue, std::string* error,
int64_t localTimeOffset) {
mpack_reader_t reader;
mpack_reader_init_data(&reader, reinterpret_cast<const char*>(in->data()),
in->size());
mpack_expect_array_match(&reader, 4);
*outId = mpack_expect_i64(&reader);
*outId = mpack_expect_int(&reader);
auto time = mpack_expect_i64(&reader);
int type = mpack_expect_int(&reader);
switch (type) {

19
ntcore/src/main/native/cpp/net/WireDecoder.h
View File

@@ -28,26 +28,26 @@ class ClientMessageHandler {
public:
virtual ~ClientMessageHandler() = default;
virtual void ClientPublish(int64_t pubuid, std::string_view name,
virtual void ClientPublish(int pubuid, std::string_view name,
std::string_view typeStr,
const wpi::json& properties) = 0;
virtual void ClientUnpublish(int64_t pubuid) = 0;
virtual void ClientUnpublish(int pubuid) = 0;
virtual void ClientSetProperties(std::string_view name,
const wpi::json& update) = 0;
virtual void ClientSubscribe(int64_t subuid,
virtual void ClientSubscribe(int subuid,
std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) = 0;
virtual void ClientUnsubscribe(int64_t subuid) = 0;
virtual void ClientUnsubscribe(int subuid) = 0;
};
class ServerMessageHandler {
public:
virtual ~ServerMessageHandler() = default;
virtual void ServerAnnounce(std::string_view name, int64_t id,
virtual void ServerAnnounce(std::string_view name, int id,
std::string_view typeStr,
const wpi::json& properties,
std::optional<int64_t> pubuid) = 0;
virtual void ServerUnannounce(std::string_view name, int64_t id) = 0;
std::optional<int> pubuid) = 0;
virtual void ServerUnannounce(std::string_view name, int id) = 0;
virtual void ServerPropertiesUpdate(std::string_view name,
const wpi::json& update, bool ack) = 0;
};
@@ -59,8 +59,7 @@ void WireDecodeText(std::string_view in, ServerMessageHandler& out,
wpi::Logger& logger);
// returns true if successfully decoded a message
bool WireDecodeBinary(std::span<const uint8_t>* in, int64_t* outId,
Value* outValue, std::string* error,
int64_t localTimeOffset);
bool WireDecodeBinary(std::span<const uint8_t>* in, int* outId, Value* outValue,
std::string* error, int64_t localTimeOffset);
} // namespace nt::net

34
ntcore/src/main/native/cpp/net/WireEncoder.cpp
View File

@@ -11,7 +11,6 @@
#include <wpi/mpack.h>
#include <wpi/raw_ostream.h>
#include "Handle.h"
#include "Message.h"
#include "PubSubOptions.h"
#include "networktables/NetworkTableValue.h"
@@ -20,7 +19,7 @@ using namespace nt;
using namespace nt::net;
using namespace mpack;
void nt::net::WireEncodePublish(wpi::raw_ostream& os, int64_t pubuid,
void nt::net::WireEncodePublish(wpi::raw_ostream& os, int pubuid,
std::string_view name, std::string_view typeStr,
const wpi::json& properties) {
wpi::json::serializer s{os, ' ', 0};
@@ -36,7 +35,7 @@ void nt::net::WireEncodePublish(wpi::raw_ostream& os, int64_t pubuid,
os << "\"}}";
}
void nt::net::WireEncodeUnpublish(wpi::raw_ostream& os, int64_t pubuid) {
void nt::net::WireEncodeUnpublish(wpi::raw_ostream& os, int pubuid) {
wpi::json::serializer s{os, ' ', 0};
os << "{\"method\":\"" << UnpublishMsg::kMethodStr << "\",\"params\":{";
os << "\"pubuid\":";
@@ -75,7 +74,7 @@ static void EncodePrefixes(wpi::raw_ostream& os, std::span<const T> topicNames,
}
template <typename T>
static void WireEncodeSubscribeImpl(wpi::raw_ostream& os, int64_t subuid,
static void WireEncodeSubscribeImpl(wpi::raw_ostream& os, int subuid,
std::span<const T> topicNames,
const PubSubOptionsImpl& options) {
wpi::json::serializer s{os, ' ', 0};
@@ -114,38 +113,37 @@ static void WireEncodeSubscribeImpl(wpi::raw_ostream& os, int64_t subuid,
os << "}}";
}
void nt::net::WireEncodeSubscribe(wpi::raw_ostream& os, int64_t subuid,
void nt::net::WireEncodeSubscribe(wpi::raw_ostream& os, int subuid,
std::span<const std::string_view> topicNames,
const PubSubOptionsImpl& options) {
WireEncodeSubscribeImpl(os, subuid, topicNames, options);
}
void nt::net::WireEncodeSubscribe(wpi::raw_ostream& os, int64_t subuid,
void nt::net::WireEncodeSubscribe(wpi::raw_ostream& os, int subuid,
std::span<const std::string> topicNames,
const PubSubOptionsImpl& options) {
WireEncodeSubscribeImpl(os, subuid, topicNames, options);
}
void nt::net::WireEncodeUnsubscribe(wpi::raw_ostream& os, int64_t subHandle) {
void nt::net::WireEncodeUnsubscribe(wpi::raw_ostream& os, int subuid) {
wpi::json::serializer s{os, ' ', 0};
os << "{\"method\":\"" << UnsubscribeMsg::kMethodStr << "\",\"params\":{";
os << "\"subuid\":";
s.dump_integer(subHandle);
s.dump_integer(subuid);
os << "}}";
}
bool nt::net::WireEncodeText(wpi::raw_ostream& os, const ClientMessage& msg) {
if (auto m = std::get_if<PublishMsg>(&msg.contents)) {
WireEncodePublish(os, Handle{m->pubHandle}.GetIndex(), m->name, m->typeStr,
m->properties);
WireEncodePublish(os, m->pubuid, m->name, m->typeStr, m->properties);
} else if (auto m = std::get_if<UnpublishMsg>(&msg.contents)) {
WireEncodeUnpublish(os, Handle{m->pubHandle}.GetIndex());
WireEncodeUnpublish(os, m->pubuid);
} else if (auto m = std::get_if<SetPropertiesMsg>(&msg.contents)) {
WireEncodeSetProperties(os, m->name, m->update);
} else if (auto m = std::get_if<SubscribeMsg>(&msg.contents)) {
WireEncodeSubscribe(os, m->subHandle, m->topicNames, m->options);
WireEncodeSubscribe(os, m->subuid, m->topicNames, m->options);
} else if (auto m = std::get_if<UnsubscribeMsg>(&msg.contents)) {
WireEncodeUnsubscribe(os, m->subHandle);
WireEncodeUnsubscribe(os, m->subuid);
} else {
return false;
}
@@ -153,9 +151,9 @@ bool nt::net::WireEncodeText(wpi::raw_ostream& os, const ClientMessage& msg) {
}
void nt::net::WireEncodeAnnounce(wpi::raw_ostream& os, std::string_view name,
int64_t id, std::string_view typeStr,
int id, std::string_view typeStr,
const wpi::json& properties,
std::optional<int64_t> pubHandle) {
std::optional<int> pubuid) {
wpi::json::serializer s{os, ' ', 0};
os << "{\"method\":\"" << AnnounceMsg::kMethodStr << "\",\"params\":{";
os << "\"id\":";
@@ -164,9 +162,9 @@ void nt::net::WireEncodeAnnounce(wpi::raw_ostream& os, std::string_view name,
s.dump_escaped(name, false);
os << "\",\"properties\":";
s.dump(properties, false, false, 0, 0);
if (pubHandle) {
if (pubuid) {
os << ",\"pubuid\":";
s.dump_integer(*pubHandle);
s.dump_integer(*pubuid);
}
os << ",\"type\":\"";
s.dump_escaped(typeStr, false);
@@ -214,7 +212,7 @@ bool nt::net::WireEncodeText(wpi::raw_ostream& os, const ServerMessage& msg) {
return true;
}
bool nt::net::WireEncodeBinary(wpi::raw_ostream& os, int64_t id, int64_t time,
bool nt::net::WireEncodeBinary(wpi::raw_ostream& os, int id, int64_t time,
const Value& value) {
char buf[128];
mpack_writer_t writer;

19
ntcore/src/main/native/cpp/net/WireEncoder.h
View File

@@ -26,24 +26,23 @@ struct ClientMessage;
struct ServerMessage;
// encoders for client text messages (avoids need to construct a Message struct)
void WireEncodePublish(wpi::raw_ostream& os, int64_t pubuid,
std::string_view name, std::string_view typeStr,
const wpi::json& properties);
void WireEncodeUnpublish(wpi::raw_ostream& os, int64_t pubuid);
void WireEncodePublish(wpi::raw_ostream& os, int pubuid, std::string_view name,
std::string_view typeStr, const wpi::json& properties);
void WireEncodeUnpublish(wpi::raw_ostream& os, int pubuid);
void WireEncodeSetProperties(wpi::raw_ostream& os, std::string_view name,
const wpi::json& update);
void WireEncodeSubscribe(wpi::raw_ostream& os, int64_t subuid,
void WireEncodeSubscribe(wpi::raw_ostream& os, int subuid,
std::span<const std::string_view> topicNames,
const PubSubOptionsImpl& options);
void WireEncodeSubscribe(wpi::raw_ostream& os, int64_t subuid,
void WireEncodeSubscribe(wpi::raw_ostream& os, int subuid,
std::span<const std::string> topicNames,
const PubSubOptionsImpl& options);
void WireEncodeUnsubscribe(wpi::raw_ostream& os, int64_t subuid);
void WireEncodeUnsubscribe(wpi::raw_ostream& os, int subuid);
// encoders for server text messages (avoids need to construct a Message struct)
void WireEncodeAnnounce(wpi::raw_ostream& os, std::string_view name, int64_t id,
void WireEncodeAnnounce(wpi::raw_ostream& os, std::string_view name, int id,
std::string_view typeStr, const wpi::json& properties,
std::optional<int64_t> pubuid);
std::optional<int> pubuid);
void WireEncodeUnannounce(wpi::raw_ostream& os, std::string_view name,
int64_t id);
void WireEncodePropertiesUpdate(wpi::raw_ostream& os, std::string_view name,
@@ -56,7 +55,7 @@ bool WireEncodeText(wpi::raw_ostream& os, const ClientMessage& msg);
bool WireEncodeText(wpi::raw_ostream& os, const ServerMessage& msg);
// encoder for binary messages
bool WireEncodeBinary(wpi::raw_ostream& os, int64_t id, int64_t time,
bool WireEncodeBinary(wpi::raw_ostream& os, int id, int64_t time,
const Value& value);
} // namespace nt::net

51
ntcore/src/main/native/cpp/net3/ClientImpl3.cpp
View File

@@ -16,7 +16,6 @@
#include <wpi/StringMap.h>
#include <wpi/timestamp.h>
#include "Handle.h"
#include "Log.h"
#include "Types_internal.h"
#include "net/Message.h"
@@ -73,14 +72,14 @@ void ClientImpl3::HandleLocal(std::span<const net::ClientMessage> msgs) {
for (const auto& elem : msgs) { // NOLINT
// common case is value
if (auto msg = std::get_if<net::ClientValueMsg>(&elem.contents)) {
SetValue(msg->pubHandle, msg->value);
SetValue(msg->pubuid, msg->value);
} else if (auto msg = std::get_if<net::PublishMsg>(&elem.contents)) {
Publish(msg->pubHandle, msg->topicHandle, msg->name, msg->typeStr,
msg->properties, msg->options);
Publish(msg->pubuid, msg->name, msg->typeStr, msg->properties,
msg->options);
} else if (auto msg = std::get_if<net::UnpublishMsg>(&elem.contents)) {
Unpublish(msg->pubHandle, msg->topicHandle);
Unpublish(msg->pubuid);
} else if (auto msg = std::get_if<net::SetPropertiesMsg>(&elem.contents)) {
SetProperties(msg->topicHandle, msg->name, msg->update);
SetProperties(msg->name, msg->update);
}
}
}
@@ -176,23 +175,20 @@ bool ClientImpl3::CheckNetworkReady(uint64_t curTimeMs) {
return true;
}
void ClientImpl3::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
std::string_view name, std::string_view typeStr,
const wpi::json& properties,
void ClientImpl3::Publish(int pubuid, std::string_view name,
std::string_view typeStr, const wpi::json& properties,
const PubSubOptionsImpl& options) {
DEBUG4("Publish('{}', '{}')", name, typeStr);
unsigned int index = Handle{pubHandle}.GetIndex();
if (index >= m_publishers.size()) {
m_publishers.resize(index + 1);
if (static_cast<unsigned int>(pubuid) >= m_publishers.size()) {
m_publishers.resize(pubuid + 1);
}
auto& publisher = m_publishers[index];
auto& publisher = m_publishers[pubuid];
if (!publisher) {
publisher = std::make_unique<PublisherData>(GetOrNewEntry(name));
publisher->entry->typeStr = typeStr;
publisher->entry->type = StringToType3(typeStr);
publisher->entry->publishers.emplace_back(publisher.get());
}
publisher->handle = pubHandle;
publisher->options = options;
publisher->periodMs = std::lround(options.periodicMs / 10.0) * 10;
if (publisher->periodMs < 10) {
@@ -204,13 +200,12 @@ void ClientImpl3::Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
m_setPeriodic(m_periodMs);
}
void ClientImpl3::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) {
DEBUG4("Unpublish({}, {})", pubHandle, topicHandle);
unsigned int index = Handle{pubHandle}.GetIndex();
if (index >= m_publishers.size()) {
void ClientImpl3::Unpublish(int pubuid) {
DEBUG4("Unpublish({})", pubuid);
if (static_cast<unsigned int>(pubuid) >= m_publishers.size()) {
return;
}
auto& publisher = m_publishers[index];
auto& publisher = m_publishers[pubuid];
publisher->entry->publishers.erase(
std::remove(publisher->entry->publishers.begin(),
publisher->entry->publishers.end(), publisher.get()),
@@ -227,9 +222,9 @@ void ClientImpl3::Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle) {
m_setPeriodic(m_periodMs);
}
void ClientImpl3::SetProperties(NT_Topic topicHandle, std::string_view name,
void ClientImpl3::SetProperties(std::string_view name,
const wpi::json& update) {
DEBUG4("SetProperties({}, {}, {})", topicHandle, name, update.dump());
DEBUG4("SetProperties({}, {})", name, update.dump());
auto entry = GetOrNewEntry(name);
bool updated = false;
for (auto&& elem : update.items()) {
@@ -250,11 +245,11 @@ void ClientImpl3::SetProperties(NT_Topic topicHandle, std::string_view name,
}
}
void ClientImpl3::SetValue(NT_Publisher pubHandle, const Value& value) {
DEBUG4("SetValue({})", pubHandle);
unsigned int index = Handle{pubHandle}.GetIndex();
assert(index < m_publishers.size() && m_publishers[index]);
auto& publisher = *m_publishers[index];
void ClientImpl3::SetValue(int pubuid, const Value& value) {
DEBUG4("SetValue({})", pubuid);
assert(static_cast<unsigned int>(pubuid) < m_publishers.size() &&
m_publishers[pubuid]);
auto& publisher = *m_publishers[pubuid];
if (value == publisher.entry->value) {
return;
}
@@ -367,8 +362,8 @@ void ClientImpl3::EntryAssign(std::string_view name, unsigned int id,
// XXX: need to handle type change specially? (e.g. with unannounce)
if (entry->topic == 0 || flagsChanged || typeChanged) {
DEBUG4("NetworkAnnounce({}, {})", name, entry->typeStr);
entry->topic =
m_local->NetworkAnnounce(name, entry->typeStr, entry->properties, 0);
entry->topic = m_local->NetworkAnnounce(name, entry->typeStr,
entry->properties, std::nullopt);
}
if (valueChanged) {
m_local->NetworkSetValue(entry->topic, entry->value);

11
ntcore/src/main/native/cpp/net3/ClientImpl3.h
View File

@@ -61,7 +61,6 @@ class ClientImpl3 final : private MessageHandler3 {
explicit PublisherData(Entry* entry) : entry{entry} {}
Entry* entry;
NT_Publisher handle;
PubSubOptionsImpl options;
// in options as double, but copy here as integer; rounded to the nearest
// 10 ms
@@ -106,13 +105,11 @@ class ClientImpl3 final : private MessageHandler3 {
bool CheckNetworkReady(uint64_t curTimeMs);
// Outgoing handlers
void Publish(NT_Publisher pubHandle, NT_Topic topicHandle,
std::string_view name, std::string_view typeStr,
void Publish(int pubuid, std::string_view name, std::string_view typeStr,
const wpi::json& properties, const PubSubOptionsImpl& options);
void Unpublish(NT_Publisher pubHandle, NT_Topic topicHandle);
void SetProperties(NT_Topic topicHandle, std::string_view name,
const wpi::json& update);
void SetValue(NT_Publisher pubHandle, const Value& value);
void Unpublish(int pubuid);
void SetProperties(std::string_view name, const wpi::json& update);
void SetValue(int pubuid, const Value& value);
// MessageHandler interface
void KeepAlive() final;