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

[ntcore] Unify listeners (#4536)

Browse Source 
This combines all 4 NT listener APIs (topic, value, connection, and
logging) into a single unified listener API.
This commit is contained in:
Peter Johnson
2022-10-31 21:52:14 -07:00
committed by GitHub
parent 32fbfb7da6
commit b114006543
71 changed files with 3613 additions and 5786 deletions
Download Patch File Download Diff File Expand all files Collapse all files

310
ntcore/src/main/native/cpp/ConnectionList.cpp
View File

@@ -4,191 +4,16 @@
#include "ConnectionList.h"
#include <atomic>
#include <optional>
#include <wpi/DataLog.h>
#include <wpi/DenseMap.h>
#include <wpi/SafeThread.h>
#include <wpi/Synchronization.h>
#include <wpi/UidVector.h>
#include <wpi/SmallVector.h>
#include <wpi/json_serializer.h>
#include <wpi/raw_ostream.h>
#include "HandleMap.h"
#include "IListenerStorage.h"
#include "ntcore_c.h"
#include "ntcore_cpp.h"
using namespace nt;
namespace {
struct PollerData {
static constexpr auto kType = Handle::kConnectionListenerPoller;
explicit PollerData(NT_ConnectionListenerPoller handle) : handle{handle} {}
wpi::SignalObject<NT_ConnectionListenerPoller> handle;
std::vector<ConnectionNotification> queue;
};
struct ListenerData {
static constexpr auto kType = Handle::kConnectionListener;
ListenerData(NT_ConnectionListener handle, PollerData* poller)
: handle{handle}, poller{poller} {}
wpi::SignalObject<NT_ConnectionListener> handle;
PollerData* poller;
};
struct DataLoggerData {
static constexpr auto kType = Handle::kConnectionDataLogger;
DataLoggerData(NT_ConnectionDataLogger handle, wpi::log::DataLog& log,
std::string_view name, int64_t time)
: handle{handle},
entry{log, name, "{\"schema\":\"NTConnectionInfo\",\"source\":\"NT\"}",
"json", time} {}
NT_ConnectionDataLogger handle;
wpi::log::StringLogEntry entry;
};
class ListenerThread final : public wpi::SafeThreadEvent {
public:
explicit ListenerThread(NT_ConnectionListenerPoller poller)
: m_poller{poller} {}
void Main() final;
NT_ConnectionListenerPoller m_poller;
wpi::DenseMap<NT_ConnectionListener,
std::function<void(const ConnectionNotification& event)>>
m_callbacks;
wpi::Event m_waitQueueWakeup;
wpi::Event m_waitQueueWaiter;
};
class CLImpl {
public:
explicit CLImpl(int inst) : m_inst{inst} {}
int m_inst;
// shared with user (must be atomic or mutex-protected)
std::atomic_bool m_connected{false};
wpi::UidVector<std::optional<ConnectionInfo>, 8> m_connections;
HandleMap<PollerData, 8> m_pollers;
HandleMap<ListenerData, 8> m_listeners;
HandleMap<DataLoggerData, 8> m_dataloggers;
wpi::SafeThreadOwner<ListenerThread> m_listenerThread;
NT_ConnectionListener AddListener(
std::function<void(const ConnectionNotification& event)> callback,
bool immediateNotify);
PollerData* CreateListenerPoller() { return m_pollers.Add(m_inst); }
void DestroyListenerPoller(NT_ConnectionListenerPoller pollerHandle);
NT_ConnectionListener AddPolledListener(
NT_ConnectionListenerPoller pollerHandle, bool immediateNotify);
void RemoveListener(NT_ConnectionListener listenerHandle);
};
} // namespace
void ListenerThread::Main() {
while (m_active) {
WPI_Handle signaledBuf[3];
auto signaled = wpi::WaitForObjects(
{m_poller, m_stopEvent.GetHandle(), m_waitQueueWakeup.GetHandle()},
signaledBuf);
if (signaled.empty() || !m_active) {
return;
}
// call all the way back out to the C++ API to ensure valid handle
auto events = nt::ReadConnectionListenerQueue(m_poller);
if (events.empty()) {
continue;
}
std::unique_lock lock{m_mutex};
for (auto&& event : events) {
auto callbackIt = m_callbacks.find(event.listener);
if (callbackIt != m_callbacks.end()) {
auto callback = callbackIt->second;
lock.unlock();
callback(event);
lock.lock();
}
}
if (std::find(signaled.begin(), signaled.end(),
m_waitQueueWakeup.GetHandle()) != signaled.end()) {
m_waitQueueWaiter.Set();
}
}
}
NT_ConnectionListener CLImpl::AddListener(
std::function<void(const ConnectionNotification& event)> callback,
bool immediateNotify) {
if (!m_listenerThread) {
m_listenerThread.Start(CreateListenerPoller()->handle);
}
if (auto thr = m_listenerThread.GetThread()) {
auto listener = AddPolledListener(thr->m_poller, immediateNotify);
if (listener) {
thr->m_callbacks.try_emplace(listener, std::move(callback));
}
return listener;
} else {
return {};
}
}
void CLImpl::DestroyListenerPoller(NT_ConnectionListenerPoller pollerHandle) {
if (auto poller = m_pollers.Remove(pollerHandle)) {
// ensure all listeners that use this poller are removed
wpi::SmallVector<NT_ConnectionListener, 16> toRemove;
for (auto&& listener : m_listeners) {
if (listener->poller == poller.get()) {
toRemove.emplace_back(listener->handle);
}
}
for (auto handle : toRemove) {
RemoveListener(handle);
}
}
}
NT_ConnectionListener CLImpl::AddPolledListener(
NT_ConnectionListenerPoller pollerHandle, bool immediateNotify) {
auto poller = m_pollers.Get(pollerHandle);
if (!poller) {
return {};
}
auto listener = m_listeners.Add(m_inst, poller);
if (immediateNotify && !m_connections.empty()) {
for (auto&& conn : m_connections) {
listener->poller->queue.emplace_back(listener->handle.GetHandle(), true,
*conn);
}
listener->poller->handle.Set();
listener->handle.Set();
}
return listener->handle;
}
void CLImpl::RemoveListener(NT_ConnectionListener listenerHandle) {
if (auto listener = m_listeners.Remove(listenerHandle)) {
if (auto thr = m_listenerThread.GetThread()) {
if (thr->m_poller == listener->poller->handle) {
thr->m_callbacks.erase(listenerHandle);
}
}
}
}
static std::string ConnInfoToJson(bool connected, const ConnectionInfo& info) {
std::string str;
wpi::raw_string_ostream os{str};
@@ -207,50 +32,35 @@ static std::string ConnInfoToJson(bool connected, const ConnectionInfo& info) {
return str;
}
class ConnectionList::Impl : public CLImpl {
public:
explicit Impl(int inst) : CLImpl{inst} {}
};
ConnectionList::ConnectionList(int inst)
: m_impl{std::make_unique<Impl>(inst)} {}
ConnectionList::ConnectionList(int inst, IListenerStorage& listenerStorage)
: m_inst{inst}, m_listenerStorage{listenerStorage} {}
ConnectionList::~ConnectionList() = default;
int ConnectionList::AddConnection(const ConnectionInfo& info) {
std::scoped_lock lock{m_mutex};
m_impl->m_connected = true;
for (auto&& listener : m_impl->m_listeners) {
listener->poller->queue.emplace_back(listener->handle.GetHandle(), true,
info);
listener->poller->handle.Set();
listener->handle.Set();
}
if (!m_impl->m_dataloggers.empty()) {
m_connected = true;
m_listenerStorage.Notify({}, NT_EVENT_CONNECTED, &info);
if (!m_dataloggers.empty()) {
auto now = Now();
for (auto&& datalogger : m_impl->m_dataloggers) {
for (auto&& datalogger : m_dataloggers) {
datalogger->entry.Append(ConnInfoToJson(true, info), now);
}
}
return m_impl->m_connections.emplace_back(info);
return m_connections.emplace_back(info);
}
void ConnectionList::RemoveConnection(int handle) {
std::scoped_lock lock{m_mutex};
auto val = m_impl->m_connections.erase(handle);
if (m_impl->m_connections.empty()) {
m_impl->m_connected = false;
auto val = m_connections.erase(handle);
if (m_connections.empty()) {
m_connected = false;
}
if (val) {
for (auto&& listener : m_impl->m_listeners) {
listener->poller->queue.emplace_back(listener->handle.GetHandle(), false,
*val);
listener->poller->handle.Set();
listener->handle.Set();
}
if (!m_impl->m_dataloggers.empty()) {
m_listenerStorage.Notify({}, NT_EVENT_DISCONNECTED, &(*val));
if (!m_dataloggers.empty()) {
auto now = Now();
for (auto&& datalogger : m_impl->m_dataloggers) {
for (auto&& datalogger : m_dataloggers) {
datalogger->entry.Append(ConnInfoToJson(false, *val), now);
}
}
@@ -259,92 +69,50 @@ void ConnectionList::RemoveConnection(int handle) {
void ConnectionList::ClearConnections() {
std::scoped_lock lock{m_mutex};
m_impl->m_connected = false;
for (auto&& conn : m_impl->m_connections) {
for (auto&& listener : m_impl->m_listeners) {
listener->poller->queue.emplace_back(listener->handle.GetHandle(), false,
*conn);
listener->poller->handle.Set();
listener->handle.Set();
}
m_connected = false;
for (auto&& conn : m_connections) {
m_listenerStorage.Notify({}, NT_EVENT_DISCONNECTED, &(*conn));
}
m_impl->m_connections.clear();
m_connections.clear();
}
std::vector<ConnectionInfo> ConnectionList::GetConnections() const {
std::scoped_lock lock{m_mutex};
std::vector<ConnectionInfo> info;
info.reserve(m_impl->m_connections.size());
for (auto&& conn : m_impl->m_connections) {
info.reserve(m_connections.size());
for (auto&& conn : m_connections) {
info.emplace_back(*conn);
}
return info;
}
bool ConnectionList::IsConnected() const {
return m_impl->m_connected;
return m_connected;
}
NT_ConnectionListener ConnectionList::AddListener(
bool immediateNotify,
std::function<void(const ConnectionNotification& event)> callback) {
void ConnectionList::AddListener(NT_Listener listener, unsigned int eventMask) {
std::scoped_lock lock{m_mutex};
return m_impl->AddListener(std::move(callback), immediateNotify);
}
bool ConnectionList::WaitForListenerQueue(double timeout) {
std::scoped_lock lock{m_mutex};
WPI_EventHandle h;
if (auto thr = m_impl->m_listenerThread.GetThread()) {
h = thr->m_waitQueueWaiter.GetHandle();
thr->m_waitQueueWakeup.Set();
} else {
return false;
eventMask &= (NT_EVENT_CONNECTION | NT_EVENT_IMMEDIATE);
m_listenerStorage.Activate(listener, eventMask);
if ((eventMask & (NT_EVENT_CONNECTED | NT_EVENT_IMMEDIATE)) ==
(NT_EVENT_CONNECTED | NT_EVENT_IMMEDIATE) &&
!m_connections.empty()) {
wpi::SmallVector<const ConnectionInfo*, 16> infos;
infos.reserve(m_connections.size());
for (auto&& conn : m_connections) {
infos.emplace_back(&(*conn));
}
m_listenerStorage.Notify({&listener, 1},
NT_EVENT_CONNECTED | NT_EVENT_IMMEDIATE, infos);
}
bool timedOut;
return wpi::WaitForObject(h, timeout, &timedOut);
}
NT_ConnectionListenerPoller ConnectionList::CreateListenerPoller() {
std::scoped_lock lock{m_mutex};
return m_impl->CreateListenerPoller()->handle;
}
void ConnectionList::DestroyListenerPoller(
NT_ConnectionListenerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
m_impl->DestroyListenerPoller(pollerHandle);
}
NT_ConnectionListener ConnectionList::AddPolledListener(
NT_ConnectionListenerPoller pollerHandle, bool immediateNotify) {
std::scoped_lock lock{m_mutex};
return m_impl->AddPolledListener(pollerHandle, immediateNotify);
}
std::vector<ConnectionNotification> ConnectionList::ReadListenerQueue(
NT_ConnectionListenerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
if (auto poller = m_impl->m_pollers.Get(pollerHandle)) {
std::vector<ConnectionNotification> rv;
rv.swap(poller->queue);
return rv;
} else {
return {};
}
}
void ConnectionList::RemoveListener(NT_ConnectionListener listenerHandle) {
std::scoped_lock lock{m_mutex};
m_impl->RemoveListener(listenerHandle);
}
NT_ConnectionDataLogger ConnectionList::StartDataLog(wpi::log::DataLog& log,
std::string_view name) {
std::scoped_lock lock{m_mutex};
auto now = Now();
auto datalogger = m_impl->m_dataloggers.Add(m_impl->m_inst, log, name, now);
for (auto&& conn : m_impl->m_connections) {
auto datalogger = m_dataloggers.Add(m_inst, log, name, now);
for (auto&& conn : m_connections) {
datalogger->entry.Append(ConnInfoToJson(true, *conn), now);
}
return datalogger->handle;
@@ -352,7 +120,7 @@ NT_ConnectionDataLogger ConnectionList::StartDataLog(wpi::log::DataLog& log,
void ConnectionList::StopDataLog(NT_ConnectionDataLogger logger) {
std::scoped_lock lock{m_mutex};
if (auto datalogger = m_impl->m_dataloggers.Remove(logger)) {
if (auto datalogger = m_dataloggers.Remove(logger)) {
datalogger->entry.Finish(Now());
}
}

49
ntcore/src/main/native/cpp/ConnectionList.h
View File

@@ -4,21 +4,31 @@
#pragma once
#include <stdint.h>
#include <atomic>
#include <functional>
#include <memory>
#include <optional>
#include <string_view>
#include <vector>
#include <wpi/DataLog.h>
#include <wpi/UidVector.h>
#include <wpi/mutex.h>
#include "Handle.h"
#include "HandleMap.h"
#include "IConnectionList.h"
#include "ntcore_cpp.h"
namespace nt {
class IListenerStorage;
class ConnectionList final : public IConnectionList {
public:
explicit ConnectionList(int inst);
ConnectionList(int inst, IListenerStorage& listenerStorage);
~ConnectionList() final;
// IConnectionList interface
@@ -30,28 +40,35 @@ class ConnectionList final : public IConnectionList {
std::vector<ConnectionInfo> GetConnections() const final;
bool IsConnected() const final;
NT_ConnectionListener AddListener(
bool immediateNotify,
std::function<void(const ConnectionNotification& event)> callback);
bool WaitForListenerQueue(double timeout);
NT_ConnectionListenerPoller CreateListenerPoller();
void DestroyListenerPoller(NT_ConnectionListenerPoller pollerHandle);
NT_ConnectionListener AddPolledListener(
NT_ConnectionListenerPoller pollerHandle, bool immediateNotify);
std::vector<ConnectionNotification> ReadListenerQueue(
NT_ConnectionListenerPoller pollerHandle);
void RemoveListener(NT_ConnectionListener listenerHandle);
void AddListener(NT_Listener listener, unsigned int eventMask);
NT_ConnectionDataLogger StartDataLog(wpi::log::DataLog& log,
std::string_view name);
void StopDataLog(NT_ConnectionDataLogger logger);
private:
class Impl;
std::unique_ptr<Impl> m_impl;
int m_inst;
IListenerStorage& m_listenerStorage;
mutable wpi::mutex m_mutex;
// shared with user (must be atomic or mutex-protected)
std::atomic_bool m_connected{false};
wpi::UidVector<std::optional<ConnectionInfo>, 8> m_connections;
struct DataLoggerData {
static constexpr auto kType = Handle::kConnectionDataLogger;
DataLoggerData(NT_ConnectionDataLogger handle, wpi::log::DataLog& log,
std::string_view name, int64_t time)
: handle{handle},
entry{log, name,
"{\"schema\":\"NTConnectionInfo\",\"source\":\"NT\"}", "json",
time} {}
NT_ConnectionDataLogger handle;
wpi::log::StringLogEntry entry;
};
HandleMap<DataLoggerData, 8> m_dataloggers;
};
} // namespace nt

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

@@ -18,24 +18,16 @@ namespace nt {
class Handle {
public:
enum Type {
kConnectionListener = wpi::kHandleTypeNTBase,
kConnectionListenerPoller,
kListener = wpi::kHandleTypeNTBase,
kListenerPoller,
kEntry,
kEntryListener,
kEntryListenerPoller,
kInstance,
kLogger,
kLoggerPoller,
kDataLogger,
kConnectionDataLogger,
kMultiSubscriber,
kTopic,
kTopicListener,
kTopicListenerPoller,
kSubscriber,
kPublisher,
kValueListener,
kValueListenerPoller,
kTypeMax
};
static_assert(kTypeMax <= wpi::kHandleTypeHALBase);

49
ntcore/src/main/native/cpp/IListenerStorage.h Normal file
View File

@@ -0,0 +1,49 @@
// Copyright (c) FIRST and other WPILib contributors.
// Open Source Software; you can modify and/or share it under the terms of
// the WPILib BSD license file in the root directory of this project.
#pragma once
#include <functional>
#include <memory>
#include <span>
#include <string_view>
#include "ntcore_cpp.h"
namespace nt {
class IListenerStorage {
public:
// Return false if event should not be issued (final check).
// This is called only during Notify() processing.
using FinishEventFunc = std::function<bool(unsigned int mask, Event* event)>;
virtual ~IListenerStorage() = default;
virtual void Activate(NT_Listener listener, unsigned int mask,
FinishEventFunc finishEvent = {}) = 0;
// If handles is not empty, notifies ONLY those listeners
virtual void Notify(std::span<const NT_Listener> handles, unsigned int flags,
std::span<ConnectionInfo const* const> infos) = 0;
virtual void Notify(std::span<const NT_Listener> handles, unsigned int flags,
std::span<const TopicInfo> infos) = 0;
virtual void Notify(std::span<const NT_Listener> handles, unsigned int flags,
NT_Topic topic, NT_Handle subentry,
const Value& value) = 0;
virtual void Notify(unsigned int flags, unsigned int level,
std::string_view filename, unsigned int line,
std::string_view message) = 0;
void Notify(std::span<const NT_Listener> handles, unsigned int flags,
const ConnectionInfo* info) {
Notify(handles, flags, {&info, 1});
}
void Notify(std::span<const NT_Listener> handles, unsigned int flags,
const TopicInfo& info) {
Notify(handles, flags, {&info, 1});
}
};
} // namespace nt

11
ntcore/src/main/native/cpp/InstanceImpl.cpp
View File

@@ -13,11 +13,12 @@ wpi::mutex InstanceImpl::s_mutex;
using namespace std::placeholders;
InstanceImpl::InstanceImpl(int inst)
: logger_impl(inst),
logger(
std::bind(&LoggerImpl::Log, &logger_impl, _1, _2, _3, _4)), // NOLINT
connectionList(inst),
localStorage(inst, logger),
: listenerStorage{inst},
logger_impl{listenerStorage},
logger{
std::bind(&LoggerImpl::Log, &logger_impl, _1, _2, _3, _4)}, // NOLINT
connectionList{inst, listenerStorage},
localStorage{inst, listenerStorage, logger},
m_inst{inst} {
logger.set_min_level(logger_impl.GetMinLevel());
}

2
ntcore/src/main/native/cpp/InstanceImpl.h
View File

@@ -15,6 +15,7 @@
#include "ConnectionList.h"
#include "Handle.h"
#include "ListenerStorage.h"
#include "LocalStorage.h"
#include "Log.h"
#include "LoggerImpl.h"
@@ -55,6 +56,7 @@ class InstanceImpl {
std::shared_ptr<NetworkServer> GetServer();
std::shared_ptr<INetworkClient> GetClient();
ListenerStorage listenerStorage;
LoggerImpl logger_impl;
wpi::Logger logger;
ConnectionList connectionList;

351
ntcore/src/main/native/cpp/ListenerStorage.cpp Normal file
View File

@@ -0,0 +1,351 @@
// Copyright (c) FIRST and other WPILib contributors.
// Open Source Software; you can modify and/or share it under the terms of
// the WPILib BSD license file in the root directory of this project.
#include "ListenerStorage.h"
#include <algorithm>
#include <wpi/DenseMap.h>
#include <wpi/SmallVector.h>
#include "ntcore_c.h"
using namespace nt;
class ListenerStorage::Thread final : public wpi::SafeThreadEvent {
public:
explicit Thread(NT_ListenerPoller poller) : m_poller{poller} {}
void Main() final;
NT_ListenerPoller m_poller;
wpi::DenseMap<NT_Listener, ListenerCallback> m_callbacks;
wpi::Event m_waitQueueWakeup;
wpi::Event m_waitQueueWaiter;
};
void ListenerStorage::Thread::Main() {
while (m_active) {
WPI_Handle signaledBuf[3];
auto signaled = wpi::WaitForObjects(
{m_poller, m_stopEvent.GetHandle(), m_waitQueueWakeup.GetHandle()},
signaledBuf);
if (signaled.empty() || !m_active) {
return;
}
// call all the way back out to the C++ API to ensure valid handle
auto events = nt::ReadListenerQueue(m_poller);
if (events.empty()) {
continue;
}
std::unique_lock lock{m_mutex};
for (auto&& event : events) {
auto callbackIt = m_callbacks.find(event.listener);
if (callbackIt != m_callbacks.end()) {
auto callback = callbackIt->second;
lock.unlock();
callback(event);
lock.lock();
}
}
if (std::find(signaled.begin(), signaled.end(),
m_waitQueueWakeup.GetHandle()) != signaled.end()) {
m_waitQueueWaiter.Set();
}
}
}
ListenerStorage::ListenerStorage(int inst) : m_inst{inst} {}
ListenerStorage::~ListenerStorage() = default;
void ListenerStorage::Activate(NT_Listener listenerHandle, unsigned int mask,
FinishEventFunc finishEvent) {
std::scoped_lock lock{m_mutex};
if (auto listener = m_listeners.Get(listenerHandle)) {
listener->sources.emplace_back(std::move(finishEvent), mask);
unsigned int deltaMask = mask & (~listener->eventMask);
listener->eventMask |= mask;
if ((deltaMask & NT_EVENT_CONNECTION) != 0) {
m_connListeners.Add(listener);
}
if ((deltaMask & NT_EVENT_TOPIC) != 0) {
m_topicListeners.Add(listener);
}
if ((deltaMask & NT_EVENT_VALUE_ALL) != 0) {
m_valueListeners.Add(listener);
}
// detect the higher log bits too; see LoggerImpl
if ((deltaMask & NT_EVENT_LOGMESSAGE) != 0 ||
(deltaMask & 0x1ff0000) != 0) {
m_logListeners.Add(listener);
}
}
}
void ListenerStorage::Notify(std::span<const NT_Listener> handles,
unsigned int flags,
std::span<ConnectionInfo const* const> infos) {
if (flags == 0) {
return;
}
std::scoped_lock lock{m_mutex};
auto doSignal = [&](ListenerData& listener) {
if ((flags & listener.eventMask) != 0) {
for (auto&& [finishEvent, mask] : listener.sources) {
if ((flags & mask) != 0) {
for (auto&& info : infos) {
listener.poller->queue.emplace_back(listener.handle, flags, *info);
// finishEvent is never set (see ConnectionList)
}
}
}
listener.handle.Set();
listener.poller->handle.Set();
}
};
if (!handles.empty()) {
for (auto handle : handles) {
if (auto listener = m_listeners.Get(handle)) {
doSignal(*listener);
}
}
} else {
for (auto&& listener : m_connListeners) {
doSignal(*listener);
}
}
}
void ListenerStorage::Notify(std::span<const NT_Listener> handles,
unsigned int flags,
std::span<const TopicInfo> infos) {
if (flags == 0) {
return;
}
std::scoped_lock lock{m_mutex};
auto doSignal = [&](ListenerData& listener) {
if ((flags & listener.eventMask) != 0) {
int count = 0;
for (auto&& [finishEvent, mask] : listener.sources) {
if ((flags & mask) != 0) {
for (auto&& info : infos) {
listener.poller->queue.emplace_back(listener.handle, flags, info);
if (finishEvent &&
!finishEvent(mask, &listener.poller->queue.back())) {
listener.poller->queue.pop_back();
} else {
++count;
}
}
}
}
if (count > 0) {
listener.handle.Set();
listener.poller->handle.Set();
}
}
};
if (!handles.empty()) {
for (auto handle : handles) {
if (auto listener = m_listeners.Get(handle)) {
doSignal(*listener);
}
}
} else {
for (auto&& listener : m_topicListeners) {
doSignal(*listener);
}
}
}
void ListenerStorage::Notify(std::span<const NT_Listener> handles,
unsigned int flags, NT_Topic topic,
NT_Handle subentry, const Value& value) {
if (flags == 0) {
return;
}
std::scoped_lock lock{m_mutex};
auto doSignal = [&](ListenerData& listener) {
if ((flags & listener.eventMask) != 0) {
int count = 0;
for (auto&& [finishEvent, mask] : listener.sources) {
if ((flags & mask) != 0) {
listener.poller->queue.emplace_back(listener.handle, flags, topic,
subentry, value);
if (finishEvent &&
!finishEvent(mask, &listener.poller->queue.back())) {
listener.poller->queue.pop_back();
} else {
++count;
}
}
}
if (count > 0) {
listener.handle.Set();
listener.poller->handle.Set();
}
}
};
if (!handles.empty()) {
for (auto handle : handles) {
if (auto listener = m_listeners.Get(handle)) {
doSignal(*listener);
}
}
} else {
for (auto&& listener : m_valueListeners) {
doSignal(*listener);
}
}
}
void ListenerStorage::Notify(unsigned int flags, unsigned int level,
std::string_view filename, unsigned int line,
std::string_view message) {
if (flags == 0) {
return;
}
std::scoped_lock lock{m_mutex};
for (auto&& listener : m_logListeners) {
if ((flags & listener->eventMask) != 0) {
int count = 0;
for (auto&& [finishEvent, mask] : listener->sources) {
if ((flags & mask) != 0) {
listener->poller->queue.emplace_back(listener->handle, flags, level,
filename, line, message);
if (finishEvent &&
!finishEvent(mask, &listener->poller->queue.back())) {
listener->poller->queue.pop_back();
} else {
++count;
}
}
}
if (count > 0) {
listener->handle.Set();
listener->poller->handle.Set();
}
}
}
}
NT_Listener ListenerStorage::AddListener(ListenerCallback callback) {
std::scoped_lock lock{m_mutex};
if (!m_thread) {
m_thread.Start(m_pollers.Add(m_inst)->handle);
}
if (auto thr = m_thread.GetThread()) {
auto listener = DoAddListener(thr->m_poller);
if (listener) {
thr->m_callbacks.try_emplace(listener, std::move(callback));
}
return listener;
} else {
return {};
}
}
NT_Listener ListenerStorage::AddListener(NT_ListenerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
return DoAddListener(pollerHandle);
}
NT_Listener ListenerStorage::DoAddListener(NT_ListenerPoller pollerHandle) {
if (auto poller = m_pollers.Get(pollerHandle)) {
return m_listeners.Add(m_inst, poller)->handle;
} else {
return {};
}
}
NT_ListenerPoller ListenerStorage::CreateListenerPoller() {
std::scoped_lock lock{m_mutex};
return m_pollers.Add(m_inst)->handle;
}
std::vector<std::pair<NT_Listener, unsigned int>>
ListenerStorage::DestroyListenerPoller(NT_ListenerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
if (auto poller = m_pollers.Remove(pollerHandle)) {
// ensure all listeners that use this poller are removed
wpi::SmallVector<NT_Listener, 16> toRemove;
for (auto&& listener : m_listeners) {
if (listener->poller == poller.get()) {
toRemove.emplace_back(listener->handle);
}
}
return DoRemoveListeners(toRemove);
} else {
return {};
}
}
std::vector<Event> ListenerStorage::ReadListenerQueue(
NT_ListenerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
if (auto poller = m_pollers.Get(pollerHandle)) {
std::vector<Event> rv;
rv.swap(poller->queue);
return rv;
} else {
return {};
}
}
std::vector<std::pair<NT_Listener, unsigned int>>
ListenerStorage::RemoveListener(NT_Listener listenerHandle) {
std::scoped_lock lock{m_mutex};
return DoRemoveListeners({&listenerHandle, 1});
}
bool ListenerStorage::WaitForListenerQueue(double timeout) {
std::scoped_lock lock{m_mutex};
WPI_EventHandle h;
if (auto thr = m_thread.GetThread()) {
h = thr->m_waitQueueWaiter.GetHandle();
thr->m_waitQueueWakeup.Set();
} else {
return false;
}
bool timedOut;
return wpi::WaitForObject(h, timeout, &timedOut);
}
std::vector<std::pair<NT_Listener, unsigned int>>
ListenerStorage::DoRemoveListeners(std::span<const NT_Listener> handles) {
std::vector<std::pair<NT_Listener, unsigned int>> rv;
auto thr = m_thread.GetThread();
for (auto handle : handles) {
if (auto listener = m_listeners.Remove(handle)) {
rv.emplace_back(handle, listener->eventMask);
if (thr) {
if (thr->m_poller == listener->poller->handle) {
thr->m_callbacks.erase(handle);
}
}
if ((listener->eventMask & NT_EVENT_CONNECTION) != 0) {
m_connListeners.Remove(listener.get());
}
if ((listener->eventMask & NT_EVENT_TOPIC) != 0) {
m_topicListeners.Remove(listener.get());
}
if ((listener->eventMask & NT_EVENT_VALUE_ALL) != 0) {
m_valueListeners.Remove(listener.get());
}
if ((listener->eventMask & NT_EVENT_LOGMESSAGE) != 0 ||
(listener->eventMask & 0x1ff0000) != 0) {
m_logListeners.Remove(listener.get());
}
}
}
return rv;
}

111
ntcore/src/main/native/cpp/ListenerStorage.h Normal file
View File

@@ -0,0 +1,111 @@
// Copyright (c) FIRST and other WPILib contributors.
// Open Source Software; you can modify and/or share it under the terms of
// the WPILib BSD license file in the root directory of this project.
#pragma once
#include <functional>
#include <memory>
#include <span>
#include <string_view>
#include <utility>
#include <vector>
#include <wpi/SafeThread.h>
#include <wpi/SmallVector.h>
#include <wpi/Synchronization.h>
#include <wpi/mutex.h>
#include "Handle.h"
#include "HandleMap.h"
#include "IListenerStorage.h"
#include "ntcore_cpp.h"
namespace nt {
class ListenerStorage final : public IListenerStorage {
public:
explicit ListenerStorage(int inst);
ListenerStorage(const ListenerStorage&) = delete;
ListenerStorage& operator=(const ListenerStorage&) = delete;
~ListenerStorage() final;
// IListenerStorage interface
void Activate(NT_Listener listenerHandle, unsigned int mask,
FinishEventFunc finishEvent = {}) final;
void Notify(std::span<const NT_Listener> handles, unsigned int flags,
std::span<ConnectionInfo const* const> infos) final;
void Notify(std::span<const NT_Listener> handles, unsigned int flags,
std::span<const TopicInfo> infos) final;
void Notify(std::span<const NT_Listener> handles, unsigned int flags,
NT_Topic topic, NT_Handle subentry, const Value& value) final;
void Notify(unsigned int flags, unsigned int level, std::string_view filename,
unsigned int line, std::string_view message) final;
// user-facing functions
NT_Listener AddListener(ListenerCallback callback);
NT_Listener AddListener(NT_ListenerPoller pollerHandle);
NT_ListenerPoller CreateListenerPoller();
// returns listener handle and mask for each listener that was destroyed
[[nodiscard]] std::vector<std::pair<NT_Listener, unsigned int>>
DestroyListenerPoller(NT_ListenerPoller pollerHandle);
std::vector<Event> ReadListenerQueue(NT_ListenerPoller pollerHandle);
// returns listener handle and mask for each listener that was destroyed
[[nodiscard]] std::vector<std::pair<NT_Listener, unsigned int>>
RemoveListener(NT_Listener listenerHandle);
bool WaitForListenerQueue(double timeout);
private:
// these assume the mutex is already held
NT_Listener DoAddListener(NT_ListenerPoller pollerHandle);
std::vector<std::pair<NT_Listener, unsigned int>> DoRemoveListeners(
std::span<const NT_Listener> handles);
int m_inst;
mutable wpi::mutex m_mutex;
struct PollerData {
static constexpr auto kType = Handle::kListenerPoller;
explicit PollerData(NT_ListenerPoller handle) : handle{handle} {}
wpi::SignalObject<NT_ListenerPoller> handle;
std::vector<Event> queue;
};
HandleMap<PollerData, 8> m_pollers;
struct ListenerData {
static constexpr auto kType = Handle::kListener;
ListenerData(NT_Listener handle, PollerData* poller)
: handle{handle}, poller{poller} {}
wpi::SignalObject<NT_Listener> handle;
PollerData* poller;
wpi::SmallVector<std::pair<FinishEventFunc, unsigned int>, 2> sources;
unsigned int eventMask{0};
};
HandleMap<ListenerData, 8> m_listeners;
// Utility wrapper for making a set-like vector
template <typename T>
class VectorSet : public std::vector<T> {
public:
void Add(T value) { this->push_back(value); }
void Remove(T value) { std::erase(*this, value); }
};
VectorSet<ListenerData*> m_connListeners;
VectorSet<ListenerData*> m_topicListeners;
VectorSet<ListenerData*> m_valueListeners;
VectorSet<ListenerData*> m_logListeners;
class Thread;
wpi::SafeThreadOwner<Thread> m_thread;
};
} // namespace nt

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

File diff suppressed because it is too large Load Diff

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

@@ -25,9 +25,12 @@ class Logger;
namespace nt {
class IListenerStorage;
class LocalStorage final : public net::ILocalStorage {
public:
LocalStorage(int inst, wpi::Logger& logger);
LocalStorage(int inst, IListenerStorage& listenerStorage,
wpi::Logger& logger);
LocalStorage(const LocalStorage&) = delete;
LocalStorage& operator=(const LocalStorage&) = delete;
~LocalStorage() final;
@@ -189,51 +192,15 @@ class LocalStorage final : public net::ILocalStorage {
int64_t GetEntryLastChange(NT_Entry entry);
//
// Topic listener functions
// Listener functions
//
NT_TopicListener AddTopicListener(
std::span<const std::string_view> prefixes, unsigned int mask,
std::function<void(const TopicNotification&)> callback);
NT_TopicListener AddTopicListener(
NT_Handle handle, unsigned int mask,
std::function<void(const TopicNotification&)> callback);
bool WaitForTopicListenerQueue(double timeout);
void AddListener(NT_Listener listener,
std::span<const std::string_view> prefixes,
unsigned int mask);
void AddListener(NT_Listener listener, NT_Handle handle, unsigned int mask);
NT_TopicListenerPoller CreateTopicListenerPoller();
void DestroyTopicListenerPoller(NT_TopicListenerPoller poller);
NT_TopicListener AddPolledTopicListener(
NT_TopicListenerPoller poller, std::span<const std::string_view> prefixes,
unsigned int mask);
NT_TopicListener AddPolledTopicListener(NT_TopicListenerPoller poller,
NT_Handle handle, unsigned int mask);
std::vector<TopicNotification> ReadTopicListenerQueue(
NT_TopicListenerPoller poller);
void RemoveTopicListener(NT_TopicListener listener);
//
// Value listener functions
//
NT_ValueListener AddValueListener(
NT_Handle subentry, unsigned int mask,
std::function<void(const ValueNotification&)> callback);
bool WaitForValueListenerQueue(double timeout);
NT_ValueListenerPoller CreateValueListenerPoller();
void DestroyValueListenerPoller(NT_ValueListenerPoller poller);
NT_ValueListener AddPolledValueListener(NT_ValueListenerPoller poller,
NT_Handle subentry,
unsigned int mask);
std::vector<ValueNotification> ReadValueListenerQueue(
NT_ValueListenerPoller poller);
void RemoveValueListener(NT_ValueListener listener);
void RemoveListener(NT_Listener listener, unsigned int mask);
//
// Data log functions

205
ntcore/src/main/native/cpp/LoggerImpl.cpp
View File

@@ -5,24 +5,27 @@
#include "LoggerImpl.h"
#include <fmt/format.h>
#include <wpi/DenseMap.h>
#include <wpi/Logger.h>
#include <wpi/SmallVector.h>
#include <wpi/fs.h>
#include "IListenerStorage.h"
using namespace nt;
static void DefaultLogger(unsigned int level, const char* file,
unsigned int line, const char* msg) {
if (level == 20) {
if (level == wpi::WPI_LOG_INFO) {
fmt::print(stderr, "NT: {}\n", msg);
return;
}
std::string_view levelmsg;
if (level >= 50) {
if (level >= wpi::WPI_LOG_CRITICAL) {
levelmsg = "CRITICAL";
} else if (level >= 40) {
} else if (level >= wpi::WPI_LOG_ERROR) {
levelmsg = "ERROR";
} else if (level >= 30) {
} else if (level >= wpi::WPI_LOG_WARNING) {
levelmsg = "WARNING";
} else {
return;
@@ -30,108 +33,107 @@ static void DefaultLogger(unsigned int level, const char* file,
fmt::print(stderr, "NT: {}: {} ({}:{})\n", levelmsg, msg, file, line);
}
class LoggerImpl::Thread final : public wpi::SafeThreadEvent {
public:
explicit Thread(NT_LoggerPoller poller) : m_poller{poller} {}
static constexpr unsigned int kFlagCritical = 1u << 16;
static constexpr unsigned int kFlagError = 1u << 17;
static constexpr unsigned int kFlagWarning = 1u << 18;
static constexpr unsigned int kFlagInfo = 1u << 19;
static constexpr unsigned int kFlagDebug = 1u << 20;
static constexpr unsigned int kFlagDebug1 = 1u << 21;
static constexpr unsigned int kFlagDebug2 = 1u << 22;
static constexpr unsigned int kFlagDebug3 = 1u << 23;
static constexpr unsigned int kFlagDebug4 = 1u << 24;
void Main() final;
NT_LoggerPoller m_poller;
wpi::DenseMap<NT_Logger, std::function<void(const LogMessage& msg)>>
m_callbacks;
};
void LoggerImpl::Thread::Main() {
while (m_active) {
WPI_Handle signaledBuf[2];
auto signaled =
wpi::WaitForObjects({m_poller, m_stopEvent.GetHandle()}, signaledBuf);
if (signaled.empty() || !m_active) {
return;
}
// call all the way back out to the C++ API to ensure valid handle
auto events = nt::ReadLoggerQueue(m_poller);
if (events.empty()) {
continue;
}
std::unique_lock lock{m_mutex};
for (auto&& event : events) {
auto callbackIt = m_callbacks.find(event.logger);
if (callbackIt != m_callbacks.end()) {
auto callback = callbackIt->second;
lock.unlock();
callback(event);
lock.lock();
}
}
static unsigned int LevelToFlag(unsigned int level) {
if (level >= wpi::WPI_LOG_CRITICAL) {
return EventFlags::kLogMessage | kFlagCritical;
} else if (level >= wpi::WPI_LOG_ERROR) {
return EventFlags::kLogMessage | kFlagError;
} else if (level >= wpi::WPI_LOG_WARNING) {
return EventFlags::kLogMessage | kFlagWarning;
} else if (level >= wpi::WPI_LOG_INFO) {
return EventFlags::kLogMessage | kFlagInfo;
} else if (level >= wpi::WPI_LOG_DEBUG) {
return EventFlags::kLogMessage | kFlagDebug;
} else if (level >= wpi::WPI_LOG_DEBUG1) {
return EventFlags::kLogMessage | kFlagDebug1;
} else if (level >= wpi::WPI_LOG_DEBUG2) {
return EventFlags::kLogMessage | kFlagDebug2;
} else if (level >= wpi::WPI_LOG_DEBUG3) {
return EventFlags::kLogMessage | kFlagDebug3;
} else if (level >= wpi::WPI_LOG_DEBUG4) {
return EventFlags::kLogMessage | kFlagDebug4;
} else {
return EventFlags::kLogMessage;
}
}
LoggerImpl::LoggerImpl(int inst) : m_inst{inst} {}
static unsigned int LevelsToEventMask(unsigned int minLevel,
unsigned int maxLevel) {
unsigned int mask = 0;
if (minLevel <= wpi::WPI_LOG_CRITICAL && maxLevel >= wpi::WPI_LOG_CRITICAL) {
mask |= kFlagCritical;
}
if (minLevel <= wpi::WPI_LOG_ERROR && maxLevel >= wpi::WPI_LOG_ERROR) {
mask |= kFlagError;
}
if (minLevel <= wpi::WPI_LOG_WARNING && maxLevel >= wpi::WPI_LOG_WARNING) {
mask |= kFlagWarning;
}
if (minLevel <= wpi::WPI_LOG_INFO && maxLevel >= wpi::WPI_LOG_INFO) {
mask |= kFlagInfo;
}
if (minLevel <= wpi::WPI_LOG_DEBUG && maxLevel >= wpi::WPI_LOG_DEBUG) {
mask |= kFlagDebug;
}
if (minLevel <= wpi::WPI_LOG_DEBUG1 && maxLevel >= wpi::WPI_LOG_DEBUG1) {
mask |= kFlagDebug1;
}
if (minLevel <= wpi::WPI_LOG_DEBUG2 && maxLevel >= wpi::WPI_LOG_DEBUG2) {
mask |= kFlagDebug2;
}
if (minLevel <= wpi::WPI_LOG_DEBUG3 && maxLevel >= wpi::WPI_LOG_DEBUG3) {
mask |= kFlagDebug3;
}
if (minLevel <= wpi::WPI_LOG_DEBUG4 && maxLevel >= wpi::WPI_LOG_DEBUG4) {
mask |= kFlagDebug4;
}
if (mask == 0) {
mask = EventFlags::kLogMessage;
}
return mask;
}
LoggerImpl::LoggerImpl(IListenerStorage& listenerStorage)
: m_listenerStorage{listenerStorage} {}
LoggerImpl::~LoggerImpl() = default;
NT_Logger LoggerImpl::Add(std::function<void(const LogMessage& msg)> callback,
unsigned int minLevel, unsigned int maxLevel) {
if (!m_thread) {
m_thread.Start(CreatePoller());
}
if (auto thr = m_thread.GetThread()) {
auto listener = AddPolled(thr->m_poller, minLevel, maxLevel);
if (listener) {
thr->m_callbacks.try_emplace(listener, std::move(callback));
}
return listener;
} else {
return {};
}
void LoggerImpl::AddListener(NT_Listener listener, unsigned int minLevel,
unsigned int maxLevel) {
++m_listenerCount;
std::scoped_lock lock{m_mutex};
m_listenerLevels.emplace_back(listener, minLevel, maxLevel);
m_listenerStorage.Activate(listener, LevelsToEventMask(minLevel, maxLevel),
[](unsigned int mask, Event* event) {
event->flags = NT_EVENT_LOGMESSAGE;
return true;
});
}
NT_LoggerPoller LoggerImpl::CreatePoller() {
void LoggerImpl::RemoveListener(NT_Listener listener) {
--m_listenerCount;
std::scoped_lock lock{m_mutex};
return m_pollers.Add(m_inst)->handle;
}
void LoggerImpl::DestroyPoller(NT_LoggerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
m_pollers.Remove(pollerHandle);
}
NT_Logger LoggerImpl::AddPolled(NT_LoggerPoller pollerHandle,
unsigned int minLevel, unsigned int maxLevel) {
std::scoped_lock lock{m_mutex};
if (auto poller = m_pollers.Get(pollerHandle)) {
return m_listeners.Add(m_inst, poller, minLevel, maxLevel)->handle;
} else {
return {};
}
}
std::vector<LogMessage> LoggerImpl::ReadQueue(NT_LoggerPoller pollerHandle) {
std::scoped_lock lock{m_mutex};
if (auto poller = m_pollers.Get(pollerHandle)) {
std::vector<LogMessage> rv;
rv.swap(poller->queue);
return rv;
} else {
return {};
}
}
void LoggerImpl::Remove(NT_Logger listenerHandle) {
std::scoped_lock lock{m_mutex};
m_listeners.Remove(listenerHandle);
if (auto thr = m_thread.GetThread()) {
thr->m_callbacks.erase(listenerHandle);
}
std::erase_if(m_listenerLevels,
[&](auto& v) { return v.listener == listener; });
}
unsigned int LoggerImpl::GetMinLevel() {
// return 0;
std::scoped_lock lock{m_mutex};
unsigned int level = NT_LOG_INFO;
for (auto&& listener : m_listeners) {
if (listener && listener->minLevel < level) {
level = listener->minLevel;
for (auto&& listenerLevel : m_listenerLevels) {
if (listenerLevel.minLevel < level) {
level = listenerLevel.minLevel;
}
}
return level;
@@ -140,19 +142,10 @@ unsigned int LoggerImpl::GetMinLevel() {
void LoggerImpl::Log(unsigned int level, const char* file, unsigned int line,
const char* msg) {
auto filename = fs::path{file}.filename();
{
std::scoped_lock lock{m_mutex};
if (m_listeners.empty()) {
DefaultLogger(level, filename.string().c_str(), line, msg);
} else {
for (auto&& listener : m_listeners) {
if (level >= listener->minLevel && level <= listener->maxLevel) {
listener->poller->queue.emplace_back(listener->handle.GetHandle(),
level, file, line, msg);
listener->poller->handle.Set();
listener->handle.Set();
}
}
}
if (m_listenerCount == 0) {
DefaultLogger(level, filename.string().c_str(), line, msg);
} else {
m_listenerStorage.Notify(LevelToFlag(level), level, filename.string(), line,
msg);
}
}

61
ntcore/src/main/native/cpp/LoggerImpl.h
View File

@@ -4,34 +4,29 @@
#pragma once
#include <utility>
#include <atomic>
#include <vector>
#include <wpi/SafeThread.h>
#include <wpi/Synchronization.h>
#include <wpi/mutex.h>
#include "Handle.h"
#include "HandleMap.h"
#include "IListenerStorage.h"
#include "ntcore_c.h"
#include "ntcore_cpp.h"
namespace nt {
class IListenerStorage;
class LoggerImpl {
public:
explicit LoggerImpl(int inst);
explicit LoggerImpl(IListenerStorage& listenerStorage);
LoggerImpl(const LoggerImpl&) = delete;
LoggerImpl& operator=(const LoggerImpl&) = delete;
~LoggerImpl();
NT_Logger Add(std::function<void(const LogMessage& msg)> callback,
unsigned int minLevel, unsigned int maxLevel);
NT_LoggerPoller CreatePoller();
void DestroyPoller(NT_LoggerPoller pollerHandle);
NT_Logger AddPolled(NT_LoggerPoller pollerHandle, unsigned int minLevel,
unsigned int maxLevel);
std::vector<LogMessage> ReadQueue(NT_LoggerPoller pollerHandle);
void Remove(NT_Logger listenerHandle);
void AddListener(NT_Listener listener, unsigned int minLevel,
unsigned int maxLevel);
void RemoveListener(NT_Listener listener);
unsigned int GetMinLevel();
@@ -39,38 +34,20 @@ class LoggerImpl {
const char* msg);
private:
int m_inst;
mutable wpi::mutex m_mutex;
IListenerStorage& m_listenerStorage;
std::atomic_int m_listenerCount{0};
wpi::mutex m_mutex;
struct PollerData {
static constexpr auto kType = Handle::kLoggerPoller;
struct ListenerLevels {
ListenerLevels(NT_Listener listener, unsigned int minLevel,
unsigned int maxLevel)
: listener{listener}, minLevel{minLevel}, maxLevel{maxLevel} {}
explicit PollerData(NT_LoggerPoller handle) : handle{handle} {}
wpi::SignalObject<NT_LoggerPoller> handle;
std::vector<LogMessage> queue;
};
HandleMap<PollerData, 8> m_pollers;
struct ListenerData {
static constexpr auto kType = Handle::kLogger;
ListenerData(NT_Logger handle, PollerData* poller, unsigned int minLevel,
unsigned int maxLevel)
: handle{handle},
poller{poller},
minLevel{minLevel},
maxLevel{maxLevel} {}
wpi::SignalObject<NT_Logger> handle;
PollerData* poller;
NT_Listener listener;
unsigned int minLevel;
unsigned int maxLevel;
};
HandleMap<ListenerData, 8> m_listeners;
class Thread;
wpi::SafeThreadOwner<Thread> m_thread;
std::vector<ListenerLevels> m_listenerLevels;
};
} // namespace nt

339
ntcore/src/main/native/cpp/jni/NetworkTablesJNI.cpp
View File

@@ -34,15 +34,14 @@ void JNI_UnloadTypes(JNIEnv* env);
static JavaVM* jvm = nullptr;
static JClass booleanCls;
static JClass connectionInfoCls;
static JClass connectionNotificationCls;
static JClass doubleCls;
static JClass eventCls;
static JClass floatCls;
static JClass logMessageCls;
static JClass longCls;
static JClass topicInfoCls;
static JClass topicNotificationCls;
static JClass valueCls;
static JClass valueNotificationCls;
static JClass valueEventDataCls;
static JException illegalArgEx;
static JException interruptedEx;
static JException nullPointerEx;
@@ -50,16 +49,14 @@ static JException nullPointerEx;
static const JClassInit classes[] = {
{"java/lang/Boolean", &booleanCls},
{"edu/wpi/first/networktables/ConnectionInfo", &connectionInfoCls},
{"edu/wpi/first/networktables/ConnectionNotification",
&connectionNotificationCls},
{"java/lang/Double", &doubleCls},
{"edu/wpi/first/networktables/NetworkTableEvent", &eventCls},
{"java/lang/Float", &floatCls},
{"edu/wpi/first/networktables/LogMessage", &logMessageCls},
{"java/lang/Long", &longCls},
{"edu/wpi/first/networktables/TopicInfo", &topicInfoCls},
{"edu/wpi/first/networktables/TopicNotification", &topicNotificationCls},
{"edu/wpi/first/networktables/NetworkTableValue", &valueCls},
{"edu/wpi/first/networktables/ValueNotification", &valueNotificationCls}};
{"edu/wpi/first/networktables/ValueEventData", &valueEventDataCls}};
static const JExceptionInit exceptions[] = {
{"java/lang/IllegalArgumentException", &illegalArgEx},
@@ -216,29 +213,12 @@ static jobject MakeJObject(JNIEnv* env, const nt::ConnectionInfo& info) {
static_cast<jint>(info.protocol_version));
}
static jobject MakeJObject(JNIEnv* env, jobject inst,
const nt::ConnectionNotification& notification) {
static jobject MakeJObject(JNIEnv* env, const nt::LogMessage& msg) {
static jmethodID constructor = env->GetMethodID(
connectionNotificationCls, "<init>",
"(Ledu/wpi/first/networktables/NetworkTableInstance;IZLedu/wpi/first/"
"networktables/ConnectionInfo;)V");
JLocal<jobject> conn{env, MakeJObject(env, notification.conn)};
return env->NewObject(connectionNotificationCls, constructor, inst,
static_cast<jint>(notification.listener),
static_cast<jboolean>(notification.connected),
conn.obj());
}
static jobject MakeJObject(JNIEnv* env, jobject inst,
const nt::LogMessage& msg) {
static jmethodID constructor = env->GetMethodID(
logMessageCls, "<init>",
"(Ledu/wpi/first/networktables/NetworkTableInstance;IILjava/lang/"
"String;ILjava/lang/String;)V");
logMessageCls, "<init>", "(ILjava/lang/String;ILjava/lang/String;)V");
JLocal<jstring> filename{env, MakeJString(env, msg.filename)};
JLocal<jstring> message{env, MakeJString(env, msg.message)};
return env->NewObject(logMessageCls, constructor, inst,
static_cast<jint>(msg.logger),
return env->NewObject(logMessageCls, constructor,
static_cast<jint>(msg.level), filename.obj(),
static_cast<jint>(msg.line), message.obj());
}
@@ -257,28 +237,42 @@ static jobject MakeJObject(JNIEnv* env, jobject inst,
}
static jobject MakeJObject(JNIEnv* env, jobject inst,
const nt::TopicNotification& notification) {
const nt::ValueEventData& data) {
static jmethodID constructor =
env->GetMethodID(topicNotificationCls, "<init>",
"(ILedu/wpi/first/networktables/TopicInfo;I)V");
JLocal<jobject> info{env, MakeJObject(env, inst, notification.info)};
return env->NewObject(topicNotificationCls, constructor,
static_cast<jint>(notification.listener), info.obj(),
static_cast<jint>(notification.flags));
env->GetMethodID(valueEventDataCls, "<init>",
"(Ledu/wpi/first/networktables/NetworkTableInstance;II"
"Ledu/wpi/first/networktables/NetworkTableValue;)V");
JLocal<jobject> value{env, MakeJValue(env, data.value)};
return env->NewObject(valueEventDataCls, constructor, inst,
static_cast<jint>(data.topic),
static_cast<jint>(data.subentry), value.obj());
}
static jobject MakeJObject(JNIEnv* env, jobject inst,
const nt::ValueNotification& notification) {
static jobject MakeJObject(JNIEnv* env, jobject inst, const nt::Event& event) {
static jmethodID constructor =
env->GetMethodID(valueNotificationCls, "<init>",
"(Ledu/wpi/first/networktables/NetworkTableInstance;III"
"Ledu/wpi/first/networktables/NetworkTableValue;I)V");
JLocal<jobject> value{env, MakeJValue(env, notification.value)};
return env->NewObject(valueNotificationCls, constructor, inst,
static_cast<jint>(notification.listener),
static_cast<jint>(notification.topic),
static_cast<jint>(notification.subentry), value.obj(),
static_cast<jint>(notification.flags));
env->GetMethodID(eventCls, "<init>",
"(Ledu/wpi/first/networktables/NetworkTableInstance;II"
"Ledu/wpi/first/networktables/ConnectionInfo;"
"Ledu/wpi/first/networktables/TopicInfo;"
"Ledu/wpi/first/networktables/ValueEventData;"
"Ledu/wpi/first/networktables/LogMessage;)V");
JLocal<jobject> connInfo{env, nullptr};
JLocal<jobject> topicInfo{env, nullptr};
JLocal<jobject> valueData{env, nullptr};
JLocal<jobject> logMessage{env, nullptr};
if (auto v = event.GetConnectionInfo()) {
connInfo = JLocal<jobject>{env, MakeJObject(env, *v)};
} else if (auto v = event.GetTopicInfo()) {
topicInfo = JLocal<jobject>{env, MakeJObject(env, inst, *v)};
} else if (auto v = event.GetValueEventData()) {
valueData = JLocal<jobject>{env, MakeJObject(env, inst, *v)};
} else if (auto v = event.GetLogMessage()) {
logMessage = JLocal<jobject>{env, MakeJObject(env, *v)};
}
return env->NewObject(eventCls, constructor, inst,
static_cast<jint>(event.listener),
static_cast<jint>(event.flags), connInfo.obj(),
topicInfo.obj(), valueData.obj(), logMessage.obj());
}
static jobjectArray MakeJObject(JNIEnv* env, std::span<const nt::Value> arr) {
@@ -293,52 +287,9 @@ static jobjectArray MakeJObject(JNIEnv* env, std::span<const nt::Value> arr) {
return jarr;
}
static jobjectArray MakeJObject(
JNIEnv* env, jobject inst,
std::span<const nt::ConnectionNotification> arr) {
jobjectArray jarr =
env->NewObjectArray(arr.size(), connectionNotificationCls, nullptr);
if (!jarr) {
return nullptr;
}
for (size_t i = 0; i < arr.size(); ++i) {
JLocal<jobject> elem{env, MakeJObject(env, inst, arr[i])};
env->SetObjectArrayElement(jarr, i, elem.obj());
}
return jarr;
}
static jobjectArray MakeJObject(JNIEnv* env, jobject inst,
std::span<const nt::LogMessage> arr) {
jobjectArray jarr = env->NewObjectArray(arr.size(), logMessageCls, nullptr);
if (!jarr) {
return nullptr;
}
for (size_t i = 0; i < arr.size(); ++i) {
JLocal<jobject> elem{env, MakeJObject(env, inst, arr[i])};
env->SetObjectArrayElement(jarr, i, elem.obj());
}
return jarr;
}
static jobjectArray MakeJObject(JNIEnv* env, jobject inst,
std::span<const nt::TopicNotification> arr) {
jobjectArray jarr =
env->NewObjectArray(arr.size(), topicNotificationCls, nullptr);
if (!jarr) {
return nullptr;
}
for (size_t i = 0; i < arr.size(); ++i) {
JLocal<jobject> elem{env, MakeJObject(env, inst, arr[i])};
env->SetObjectArrayElement(jarr, i, elem.obj());
}
return jarr;
}
static jobjectArray MakeJObject(JNIEnv* env, jobject inst,
std::span<const nt::ValueNotification> arr) {
jobjectArray jarr =
env->NewObjectArray(arr.size(), valueNotificationCls, nullptr);
std::span<const nt::Event> arr) {
jobjectArray jarr = env->NewObjectArray(arr.size(), eventCls, nullptr);
if (!jarr) {
return nullptr;
}
@@ -1013,35 +964,35 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_getTopicInfo
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: createTopicListenerPoller
* Method: createListenerPoller
* Signature: (I)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_createTopicListenerPoller
Java_edu_wpi_first_networktables_NetworkTablesJNI_createListenerPoller
(JNIEnv*, jclass, jint inst)
{
return nt::CreateTopicListenerPoller(inst);
return nt::CreateListenerPoller(inst);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: destroyTopicListenerPoller
* Method: destroyListenerPoller
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_destroyTopicListenerPoller
Java_edu_wpi_first_networktables_NetworkTablesJNI_destroyListenerPoller
(JNIEnv*, jclass, jint poller)
{
nt::DestroyTopicListenerPoller(poller);
nt::DestroyListenerPoller(poller);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: addPolledTopicListener
* Method: addListener
* Signature: (I[Ljava/lang/Object;I)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledTopicListener__I_3Ljava_lang_String_2I
Java_edu_wpi_first_networktables_NetworkTablesJNI_addListener__I_3Ljava_lang_String_2I
(JNIEnv* env, jclass, jint poller, jobjectArray prefixes, jint flags)
{
if (!prefixes) {
@@ -1066,163 +1017,43 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledTopicListener__I_3Lja
arrview.emplace_back(arr.back());
}
return nt::AddPolledTopicListener(poller, arrview, flags);
return nt::AddPolledListener(poller, arrview, flags);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: addPolledTopicListener
* Method: addListener
* Signature: (III)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledTopicListener__III
Java_edu_wpi_first_networktables_NetworkTablesJNI_addListener__III
(JNIEnv* env, jclass, jint poller, jint handle, jint flags)
{
return nt::AddPolledTopicListener(poller, handle, flags);
return nt::AddPolledListener(poller, handle, flags);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: readTopicListenerQueue
* Method: readListenerQueue
* Signature: (Ljava/lang/Object;I)[Ljava/lang/Object;
*/
JNIEXPORT jobjectArray JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_readTopicListenerQueue
Java_edu_wpi_first_networktables_NetworkTablesJNI_readListenerQueue
(JNIEnv* env, jclass, jobject inst, jint poller)
{
return MakeJObject(env, inst, nt::ReadTopicListenerQueue(poller));
return MakeJObject(env, inst, nt::ReadListenerQueue(poller));
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: removeTopicListener
* Method: removeListener
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_removeTopicListener
Java_edu_wpi_first_networktables_NetworkTablesJNI_removeListener
(JNIEnv*, jclass, jint topicListener)
{
nt::RemoveTopicListener(topicListener);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: createValueListenerPoller
* Signature: (I)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_createValueListenerPoller
(JNIEnv*, jclass, jint inst)
{
return nt::CreateValueListenerPoller(inst);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: destroyValueListenerPoller
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_destroyValueListenerPoller
(JNIEnv*, jclass, jint poller)
{
nt::DestroyValueListenerPoller(poller);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: addPolledValueListener
* Signature: (III)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledValueListener
(JNIEnv* env, jclass, jint poller, jint topic, jint flags)
{
return nt::AddPolledValueListener(poller, topic, flags);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: readValueListenerQueue
* Signature: (Ljava/lang/Object;I)[Ljava/lang/Object;
*/
JNIEXPORT jobjectArray JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_readValueListenerQueue
(JNIEnv* env, jclass, jobject inst, jint poller)
{
return MakeJObject(env, inst, nt::ReadValueListenerQueue(poller));
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: removeValueListener
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_removeValueListener
(JNIEnv*, jclass, jint topicListener)
{
nt::RemoveValueListener(topicListener);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: createConnectionListenerPoller
* Signature: (I)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_createConnectionListenerPoller
(JNIEnv*, jclass, jint inst)
{
return nt::CreateConnectionListenerPoller(inst);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: destroyConnectionListenerPoller
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_destroyConnectionListenerPoller
(JNIEnv*, jclass, jint poller)
{
nt::DestroyConnectionListenerPoller(poller);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: addPolledConnectionListener
* Signature: (IZ)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledConnectionListener
(JNIEnv* env, jclass, jint poller, jboolean immediateNotify)
{
return nt::AddPolledConnectionListener(poller, immediateNotify);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: readConnectionListenerQueue
* Signature: (Ljava/lang/Object;I)[Ljava/lang/Object;
*/
JNIEXPORT jobjectArray JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_readConnectionListenerQueue
(JNIEnv* env, jclass, jobject inst, jint poller)
{
return MakeJObject(env, inst, nt::ReadConnectionListenerQueue(poller));
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: removeConnectionListener
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_removeConnectionListener
(JNIEnv*, jclass, jint connListenerUid)
{
nt::RemoveConnectionListener(connListenerUid);
nt::RemoveListener(topicListener);
}
/*
@@ -1561,62 +1392,14 @@ Java_edu_wpi_first_networktables_NetworkTablesJNI_stopConnectionDataLog
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: createLoggerPoller
* Signature: (I)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_createLoggerPoller
(JNIEnv*, jclass, jint inst)
{
return nt::CreateLoggerPoller(inst);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: destroyLoggerPoller
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_destroyLoggerPoller
(JNIEnv*, jclass, jint poller)
{
nt::DestroyLoggerPoller(poller);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: addPolledLogger
* Method: addLogger
* Signature: (III)I
*/
JNIEXPORT jint JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_addPolledLogger
Java_edu_wpi_first_networktables_NetworkTablesJNI_addLogger
(JNIEnv*, jclass, jint poller, jint minLevel, jint maxLevel)
{
return nt::AddPolledLogger(poller, minLevel, maxLevel);
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: readLoggerQueue
* Signature: (Ljava/lang/Object;I)[Ljava/lang/Object;
*/
JNIEXPORT jobjectArray JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_readLoggerQueue
(JNIEnv* env, jclass, jobject inst, jint poller)
{
return MakeJObject(env, inst, nt::ReadLoggerQueue(poller));
}
/*
* Class: edu_wpi_first_networktables_NetworkTablesJNI
* Method: removeLogger
* Signature: (I)V
*/
JNIEXPORT void JNICALL
Java_edu_wpi_first_networktables_NetworkTablesJNI_removeLogger
(JNIEnv*, jclass, jint logger)
{
nt::RemoveLogger(logger);
}
} // extern "C"

17
ntcore/src/main/native/cpp/networktables/NetworkTableInstance.cpp
View File

@@ -101,16 +101,9 @@ void NetworkTableInstance::SetServer(std::span<const std::string_view> servers,
SetServer(serversArr);
}
NT_TopicListener NetworkTableInstance::AddTopicListener(
MultiSubscriber& subscriber, int eventMask,
std::function<void(const TopicNotification&)> listener) {
return ::nt::AddTopicListener(subscriber.GetHandle(), eventMask,
std::move(listener));
}
NT_ValueListener NetworkTableInstance::AddValueListener(
MultiSubscriber& subscriber, unsigned int eventMask,
std::function<void(const ValueNotification&)> listener) {
return ::nt::AddValueListener(subscriber.GetHandle(), eventMask,
std::move(listener));
NT_Listener NetworkTableInstance::AddListener(MultiSubscriber& subscriber,
int eventMask,
ListenerCallback listener) {
return ::nt::AddListener(subscriber.GetHandle(), eventMask,
std::move(listener));
}

328
ntcore/src/main/native/cpp/ntcore_c.cpp
View File

@@ -37,35 +37,42 @@ static void ConvertToC(const ConnectionInfo& in, NT_ConnectionInfo* out) {
out->protocol_version = in.protocol_version;
}
static void ConvertToC(const TopicNotification& in, NT_TopicNotification* out) {
out->listener = in.listener;
ConvertToC(in.info, &out->info);
out->flags = in.flags;
}
static void ConvertToC(const ValueNotification& in, NT_ValueNotification* out) {
out->listener = in.listener;
static void ConvertToC(const ValueEventData& in, NT_ValueEventData* out) {
out->topic = in.topic;
out->subentry = in.subentry;
ConvertToC(in.value, &out->value);
out->flags = in.flags;
}
static void ConvertToC(const ConnectionNotification& in,
NT_ConnectionNotification* out) {
out->listener = in.listener;
out->connected = in.connected;
ConvertToC(in.conn, &out->conn);
}
static void ConvertToC(const LogMessage& in, NT_LogMessage* out) {
out->logger = in.logger;
out->level = in.level;
ConvertToC(in.filename, &out->filename);
out->line = in.line;
ConvertToC(in.message, &out->message);
}
static void ConvertToC(const Event& in, NT_Event* out) {
out->listener = in.listener;
out->flags = in.flags;
if ((in.flags & NT_EVENT_VALUE_ALL) != 0) {
if (auto v = in.GetValueEventData()) {
return ConvertToC(*v, &out->data.valueData);
}
} else if ((in.flags & NT_EVENT_TOPIC) != 0) {
if (auto v = in.GetTopicInfo()) {
return ConvertToC(*v, &out->data.topicInfo);
}
} else if ((in.flags & NT_EVENT_CONNECTION) != 0) {
if (auto v = in.GetConnectionInfo()) {
return ConvertToC(*v, &out->data.connInfo);
}
} else if ((in.flags & NT_EVENT_LOGMESSAGE) != 0) {
if (auto v = in.GetLogMessage()) {
return ConvertToC(*v, &out->data.logMessage);
}
}
out->flags = NT_EVENT_NONE; // sanity to make sure we don't dispose
}
static void DisposeConnectionInfo(NT_ConnectionInfo* info) {
std::free(info->remote_id.str);
std::free(info->remote_ip.str);
@@ -77,16 +84,21 @@ static void DisposeTopicInfo(NT_TopicInfo* info) {
std::free(info->properties.str);
}
static void DisposeTopicNotification(NT_TopicNotification* info) {
DisposeTopicInfo(&info->info);
static void DisposeLogMessage(NT_LogMessage* msg) {
std::free(msg->filename);
std::free(msg->message);
}
static void DisposeValueNotification(NT_ValueNotification* info) {
NT_DisposeValue(&info->value);
}
static void DisposeConnectionNotification(NT_ConnectionNotification* info) {
DisposeConnectionInfo(&info->conn);
static void DisposeEvent(NT_Event* event) {
if ((event->flags & NT_EVENT_VALUE_ALL) != 0) {
NT_DisposeValue(&event->data.valueData.value);
} else if ((event->flags & NT_EVENT_TOPIC) != 0) {
DisposeTopicInfo(&event->data.topicInfo);
} else if ((event->flags & NT_EVENT_CONNECTION) != 0) {
DisposeConnectionInfo(&event->data.connInfo);
} else if ((event->flags & NT_EVENT_LOGMESSAGE) != 0) {
DisposeLogMessage(&event->data.logMessage);
}
}
extern "C" {
@@ -383,169 +395,87 @@ NT_Topic NT_GetTopicFromHandle(NT_Handle pubsubentry) {
* Callback Creation Functions
*/
NT_TopicListener NT_AddTopicListener(NT_Inst inst, const char* prefix,
size_t prefix_len, unsigned int mask,
void* data,
NT_TopicListenerCallback callback) {
NT_ListenerPoller NT_CreateListenerPoller(NT_Inst inst) {
return nt::CreateListenerPoller(inst);
}
void NT_DestroyListenerPoller(NT_ListenerPoller poller) {
nt::DestroyListenerPoller(poller);
}
struct NT_Event* NT_ReadListenerQueue(NT_ListenerPoller poller, size_t* len) {
auto arr_cpp = nt::ReadListenerQueue(poller);
return ConvertToC<NT_Event>(arr_cpp, len);
}
void NT_RemoveListener(NT_Listener listener) {
nt::RemoveListener(listener);
}
NT_Bool NT_WaitForListenerQueue(NT_Handle handle, double timeout) {
return nt::WaitForListenerQueue(handle, timeout);
}
NT_Listener NT_AddListenerSingle(NT_Inst inst, const char* prefix,
size_t prefix_len, unsigned int mask,
void* data, NT_ListenerCallback callback) {
std::string_view p{prefix, prefix_len};
return nt::AddTopicListener(inst, {{p}}, mask, [=](auto& event) {
NT_TopicNotification event_c;
return nt::AddListener(inst, {{p}}, mask, [=](auto& event) {
NT_Event event_c;
ConvertToC(event, &event_c);
callback(data, &event_c);
DisposeTopicNotification(&event_c);
DisposeEvent(&event_c);
});
}
NT_TopicListener NT_AddTopicListenerMultiple(
NT_Inst inst, const NT_String* prefixes, size_t prefixes_len,
unsigned int mask, void* data, NT_TopicListenerCallback callback) {
NT_Listener NT_AddListenerMultiple(NT_Inst inst, const NT_String* prefixes,
size_t prefixes_len, unsigned int mask,
void* data, NT_ListenerCallback callback) {
wpi::SmallVector<std::string_view, 8> p;
p.reserve(prefixes_len);
for (size_t i = 0; i < prefixes_len; ++i) {
p.emplace_back(prefixes[i].str, prefixes[i].len);
}
return nt::AddTopicListener(inst, p, mask, [=](auto& event) {
NT_TopicNotification event_c;
return nt::AddListener(inst, p, mask, [=](auto& event) {
NT_Event event_c;
ConvertToC(event, &event_c);
callback(data, &event_c);
DisposeTopicNotification(&event_c);
DisposeEvent(&event_c);
});
}
NT_TopicListener NT_AddTopicListenerSingle(NT_Topic topic, unsigned int mask,
void* data,
NT_TopicListenerCallback callback) {
return nt::AddTopicListener(topic, mask, [=](auto& event) {
NT_TopicNotification event_c;
NT_Listener NT_AddListener(NT_Topic topic, unsigned int mask, void* data,
NT_ListenerCallback callback) {
return nt::AddListener(topic, mask, [=](auto& event) {
NT_Event event_c;
ConvertToC(event, &event_c);
callback(data, &event_c);
DisposeTopicNotification(&event_c);
DisposeEvent(&event_c);
});
}
NT_Bool NT_WaitForTopicListenerQueue(NT_Handle handle, double timeout) {
return nt::WaitForTopicListenerQueue(handle, timeout);
}
NT_TopicListenerPoller NT_CreateTopicListenerPoller(NT_Inst inst) {
return nt::CreateTopicListenerPoller(inst);
}
void NT_DestroyTopicListenerPoller(NT_TopicListenerPoller poller) {
nt::DestroyTopicListenerPoller(poller);
}
NT_TopicListener NT_AddPolledTopicListener(NT_TopicListenerPoller poller,
const char* prefix,
size_t prefix_len,
unsigned int mask) {
NT_Listener NT_AddPolledListenerSingle(NT_ListenerPoller poller,
const char* prefix, size_t prefix_len,
unsigned int mask) {
std::string_view p{prefix, prefix_len};
return nt::AddPolledTopicListener(poller, {{p}}, mask);
return nt::AddPolledListener(poller, {{p}}, mask);
}
NT_TopicListener NT_AddPolledTopicListenerMultiple(
NT_TopicListenerPoller poller, const NT_String* prefixes,
size_t prefixes_len, unsigned int mask) {
NT_Listener NT_AddPolledListenerMultiple(NT_ListenerPoller poller,
const NT_String* prefixes,
size_t prefixes_len,
unsigned int mask) {
wpi::SmallVector<std::string_view, 8> p;
p.reserve(prefixes_len);
for (size_t i = 0; i < prefixes_len; ++i) {
p.emplace_back(prefixes[i].str, prefixes[i].len);
}
return nt::AddPolledTopicListener(poller, p, mask);
return nt::AddPolledListener(poller, p, mask);
}
NT_TopicListener NT_AddPolledTopicListenerSingle(NT_TopicListenerPoller poller,
NT_Topic topic,
unsigned int mask) {
return nt::AddPolledTopicListener(poller, topic, mask);
}
struct NT_TopicNotification* NT_ReadTopicListenerQueue(
NT_TopicListenerPoller poller, size_t* len) {
auto arr_cpp = nt::ReadTopicListenerQueue(poller);
return ConvertToC<NT_TopicNotification>(arr_cpp, len);
}
void NT_RemoveTopicListener(NT_TopicListener topic_listener) {
nt::RemoveTopicListener(topic_listener);
}
NT_ValueListener NT_AddValueListener(NT_Handle subentry, unsigned int mask,
void* data,
NT_ValueListenerCallback callback) {
return nt::AddValueListener(subentry, mask, [=](auto& event) {
NT_ValueNotification event_c;
ConvertToC(event, &event_c);
callback(data, &event_c);
DisposeValueNotification(&event_c);
});
}
NT_Bool NT_WaitForValueListenerQueue(NT_Handle handle, double timeout) {
return nt::WaitForValueListenerQueue(handle, timeout);
}
NT_ValueListenerPoller NT_CreateValueListenerPoller(NT_Inst inst) {
return nt::CreateValueListenerPoller(inst);
}
void NT_DestroyValueListenerPoller(NT_ValueListenerPoller poller) {
nt::DestroyValueListenerPoller(poller);
}
NT_ValueListener NT_AddPolledValueListener(NT_ValueListenerPoller poller,
NT_Handle subentry,
unsigned int mask) {
return nt::AddPolledValueListener(poller, subentry, mask);
}
struct NT_ValueNotification* NT_ReadValueListenerQueue(
NT_ValueListenerPoller poller, size_t* len) {
auto arr_cpp = nt::ReadValueListenerQueue(poller);
return ConvertToC<NT_ValueNotification>(arr_cpp, len);
}
void NT_RemoveValueListener(NT_ValueListener value_listener) {
nt::RemoveValueListener(value_listener);
}
NT_ConnectionListener NT_AddConnectionListener(
NT_Inst inst, NT_Bool immediate_notify, void* data,
NT_ConnectionListenerCallback callback) {
return nt::AddConnectionListener(inst, immediate_notify != 0,
[=](const ConnectionNotification& event) {
NT_ConnectionNotification event_c;
ConvertToC(event, &event_c);
callback(data, &event_c);
DisposeConnectionNotification(&event_c);
});
}
NT_Bool NT_WaitForConnectionListenerQueue(NT_Handle handle, double timeout) {
return nt::WaitForConnectionListenerQueue(handle, timeout);
}
NT_ConnectionListenerPoller NT_CreateConnectionListenerPoller(NT_Inst inst) {
return nt::CreateConnectionListenerPoller(inst);
}
void NT_DestroyConnectionListenerPoller(NT_ConnectionListenerPoller poller) {
nt::DestroyConnectionListenerPoller(poller);
}
NT_ConnectionListener NT_AddPolledConnectionListener(
NT_ConnectionListenerPoller poller, NT_Bool immediate_notify) {
return nt::AddPolledConnectionListener(poller, immediate_notify);
}
struct NT_ConnectionNotification* NT_ReadConnectionListenerQueue(
NT_ConnectionListenerPoller poller, size_t* len) {
auto arr_cpp = nt::ReadConnectionListenerQueue(poller);
return ConvertToC<NT_ConnectionNotification>(arr_cpp, len);
}
void NT_RemoveConnectionListener(NT_ConnectionListener conn_listener) {
nt::RemoveConnectionListener(conn_listener);
NT_Listener NT_AddPolledListener(NT_ListenerPoller poller, NT_Topic topic,
unsigned int mask) {
return nt::AddPolledListener(poller, topic, mask);
}
/*
@@ -641,41 +571,22 @@ void NT_SetNow(uint64_t timestamp) {
nt::SetNow(timestamp);
}
NT_Logger NT_AddLogger(NT_Inst inst, void* data, NT_LogFunc func,
unsigned int min_level, unsigned int max_level) {
return nt::AddLogger(
inst,
[=](const LogMessage& msg) {
NT_LogMessage msg_c;
ConvertToC(msg, &msg_c);
func(data, &msg_c);
NT_DisposeLogMessage(&msg_c);
},
min_level, max_level);
NT_Listener NT_AddLogger(NT_Inst inst, unsigned int min_level,
unsigned int max_level, void* data,
NT_ListenerCallback func) {
return nt::AddLogger(inst, min_level, max_level, [=](auto& event) {
NT_Event event_c;
ConvertToC(event, &event_c);
func(data, &event_c);
NT_DisposeEvent(&event_c);
});
}
NT_LoggerPoller NT_CreateLoggerPoller(NT_Inst inst) {
return nt::CreateLoggerPoller(inst);
}
void NT_DestroyLoggerPoller(NT_LoggerPoller poller) {
nt::DestroyLoggerPoller(poller);
}
NT_Logger NT_AddPolledLogger(NT_LoggerPoller poller, unsigned int min_level,
unsigned int max_level) {
NT_Listener NT_AddPolledLogger(NT_ListenerPoller poller, unsigned int min_level,
unsigned int max_level) {
return nt::AddPolledLogger(poller, min_level, max_level);
}
struct NT_LogMessage* NT_ReadLoggerQueue(NT_LoggerPoller poller, size_t* len) {
auto arr_cpp = nt::ReadLoggerQueue(poller);
return ConvertToC<NT_LogMessage>(arr_cpp, len);
}
void NT_RemoveLogger(NT_Logger logger) {
nt::RemoveLogger(logger);
}
void NT_DisposeValue(NT_Value* value) {
switch (value->type) {
case NT_UNASSIGNED:
@@ -759,50 +670,15 @@ void NT_DisposeTopicInfo(NT_TopicInfo* info) {
DisposeTopicInfo(info);
}
void NT_DisposeTopicNotificationArray(NT_TopicNotification* arr, size_t count) {
void NT_DisposeEventArray(NT_Event* arr, size_t count) {
for (size_t i = 0; i < count; i++) {
DisposeTopicNotification(&arr[i]);
DisposeEvent(&arr[i]);
}
std::free(arr);
}
void NT_DisposeTopicNotification(NT_TopicNotification* info) {
DisposeTopicNotification(info);
}
void NT_DisposeValueNotificationArray(NT_ValueNotification* arr, size_t count) {
for (size_t i = 0; i < count; i++) {
DisposeValueNotification(&arr[i]);
}
std::free(arr);
}
void NT_DisposeValueNotification(NT_ValueNotification* info) {
DisposeValueNotification(info);
}
void NT_DisposeConnectionNotificationArray(NT_ConnectionNotification* arr,
size_t count) {
for (size_t i = 0; i < count; i++) {
DisposeConnectionNotification(&arr[i]);
}
std::free(arr);
}
void NT_DisposeConnectionNotification(NT_ConnectionNotification* info) {
DisposeConnectionNotification(info);
}
void NT_DisposeLogMessageArray(NT_LogMessage* arr, size_t count) {
for (size_t i = 0; i < count; i++) {
NT_DisposeLogMessage(&arr[i]);
}
std::free(arr);
}
void NT_DisposeLogMessage(NT_LogMessage* info) {
std::free(info->filename);
std::free(info->message);
void NT_DisposeEvent(NT_Event* event) {
DisposeEvent(event);
}
/* Interop Utility Functions */

303
ntcore/src/main/native/cpp/ntcore_cpp.cpp
View File

@@ -9,6 +9,7 @@
#include <cstdio>
#include <cstdlib>
#include <fmt/format.h>
#include <wpi/json.h>
#include <wpi/timestamp.h>
@@ -17,6 +18,7 @@
#include "Log.h"
#include "Types_internal.h"
#include "ntcore.h"
#include "ntcore_c.h"
static std::atomic_bool gNowSet{false};
static std::atomic<int64_t> gNowTime;
@@ -54,7 +56,7 @@ void DestroyInstance(NT_Inst inst) {
NT_Inst GetInstanceFromHandle(NT_Handle handle) {
Handle h{handle};
auto type = h.GetType();
if (type >= Handle::kConnectionListener && type < Handle::kTypeMax) {
if (type >= Handle::kListener && type < Handle::kTypeMax) {
return Handle(h.GetInst(), 0, Handle::kInstance);
}
@@ -386,200 +388,123 @@ void UnsubscribeMultiple(NT_MultiSubscriber sub) {
* Callback Creation Functions
*/
NT_TopicListener AddTopicListener(
NT_Inst inst, std::span<const std::string_view> prefixes, unsigned int mask,
std::function<void(const TopicNotification&)> callback) {
static void CleanupListeners(
InstanceImpl& ii,
std::span<const std::pair<NT_Listener, unsigned int>> listeners) {
bool updateMinLevel = false;
for (auto&& [listener, mask] : listeners) {
// connection doesn't need removal notification
if ((mask & (NT_EVENT_TOPIC | NT_EVENT_VALUE_ALL)) != 0) {
ii.localStorage.RemoveListener(listener, mask);
}
if ((mask & NT_EVENT_LOGMESSAGE) != 0) {
ii.logger_impl.RemoveListener(listener);
updateMinLevel = true;
}
}
if (updateMinLevel) {
ii.logger.set_min_level(ii.logger_impl.GetMinLevel());
}
}
static void DoAddListener(InstanceImpl& ii, NT_Listener listener,
NT_Handle handle, unsigned int mask) {
if (Handle{handle}.IsType(Handle::kInstance)) {
if ((mask & NT_EVENT_CONNECTION) != 0) {
ii.connectionList.AddListener(listener, mask);
}
if ((mask & NT_EVENT_LOGMESSAGE) != 0) {
ii.logger_impl.AddListener(listener, NT_LOG_INFO, UINT_MAX);
}
} else if ((mask & (NT_EVENT_TOPIC | NT_EVENT_VALUE_ALL)) != 0) {
ii.localStorage.AddListener(listener, handle, mask);
}
}
NT_ListenerPoller CreateListenerPoller(NT_Inst inst) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->localStorage.AddTopicListener(prefixes, mask,
std::move(callback));
return ii->listenerStorage.CreateListenerPoller();
} else {
return {};
}
}
NT_TopicListener AddTopicListener(
NT_Handle handle, unsigned int mask,
std::function<void(const TopicNotification&)> callback) {
if (auto ii = InstanceImpl::GetTyped(handle, Handle::kTopic)) {
return ii->localStorage.AddTopicListener(handle, mask, std::move(callback));
void DestroyListenerPoller(NT_ListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kListenerPoller)) {
CleanupListeners(*ii, ii->listenerStorage.DestroyListenerPoller(poller));
}
}
std::vector<Event> ReadListenerQueue(NT_ListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kListenerPoller)) {
return ii->listenerStorage.ReadListenerQueue(poller);
} else {
return {};
}
}
bool WaitForTopicListenerQueue(NT_Handle handle, double timeout) {
void RemoveListener(NT_Listener listener) {
if (auto ii = InstanceImpl::GetTyped(listener, Handle::kListener)) {
CleanupListeners(*ii, ii->listenerStorage.RemoveListener(listener));
}
}
bool WaitForListenerQueue(NT_Handle handle, double timeout) {
if (auto ii = InstanceImpl::GetHandle(handle)) {
return ii->localStorage.WaitForTopicListenerQueue(timeout);
return ii->listenerStorage.WaitForListenerQueue(timeout);
} else {
return {};
}
}
NT_TopicListenerPoller CreateTopicListenerPoller(NT_Inst inst) {
NT_Listener AddListener(NT_Inst inst,
std::span<const std::string_view> prefixes,
unsigned int mask, ListenerCallback callback) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->localStorage.CreateTopicListenerPoller();
} else {
return {};
if ((mask & (NT_EVENT_TOPIC | NT_EVENT_VALUE_ALL)) != 0) {
auto listener = ii->listenerStorage.AddListener(std::move(callback));
ii->localStorage.AddListener(listener, prefixes, mask);
return listener;
}
}
return {};
}
void DestroyTopicListenerPoller(NT_TopicListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kTopicListenerPoller)) {
ii->localStorage.DestroyTopicListenerPoller(poller);
}
}
NT_TopicListener AddPolledTopicListener(
NT_TopicListenerPoller poller, std::span<const std::string_view> prefixes,
unsigned int mask) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kTopicListenerPoller)) {
return ii->localStorage.AddPolledTopicListener(poller, prefixes, mask);
} else {
return {};
}
}
NT_TopicListener AddPolledTopicListener(NT_TopicListenerPoller poller,
NT_Handle handle, unsigned int mask) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kTopicListenerPoller)) {
return ii->localStorage.AddPolledTopicListener(poller, handle, mask);
} else {
return {};
}
}
std::vector<TopicNotification> ReadTopicListenerQueue(
NT_TopicListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kTopicListenerPoller)) {
return ii->localStorage.ReadTopicListenerQueue(poller);
} else {
return {};
}
}
void RemoveTopicListener(NT_TopicListener listener) {
if (auto ii = InstanceImpl::GetTyped(listener, Handle::kTopicListener)) {
return ii->localStorage.RemoveTopicListener(listener);
}
}
NT_ValueListener AddValueListener(
NT_Handle subentry, unsigned int mask,
std::function<void(const ValueNotification&)> callback) {
if (auto ii = InstanceImpl::GetHandle(subentry)) {
return ii->localStorage.AddValueListener(subentry, mask,
std::move(callback));
} else {
return {};
}
}
bool WaitForValueListenerQueue(NT_Handle handle, double timeout) {
NT_Listener AddListener(NT_Handle handle, unsigned int mask,
ListenerCallback callback) {
if (auto ii = InstanceImpl::GetHandle(handle)) {
return ii->localStorage.WaitForValueListenerQueue(timeout);
auto listener = ii->listenerStorage.AddListener(std::move(callback));
DoAddListener(*ii, listener, handle, mask);
return listener;
} else {
return {};
}
}
NT_ValueListenerPoller CreateValueListenerPoller(NT_Inst inst) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->localStorage.CreateValueListenerPoller();
NT_Listener AddPolledListener(NT_ListenerPoller poller,
std::span<const std::string_view> prefixes,
unsigned int mask) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kListenerPoller)) {
if ((mask & (NT_EVENT_TOPIC | NT_EVENT_VALUE_ALL)) != 0) {
auto listener = ii->listenerStorage.AddListener(poller);
ii->localStorage.AddListener(listener, prefixes, mask);
return listener;
}
}
return {};
}
NT_Listener AddPolledListener(NT_ListenerPoller poller, NT_Handle handle,
unsigned int mask) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kListenerPoller)) {
auto listener = ii->listenerStorage.AddListener(poller);
DoAddListener(*ii, listener, handle, mask);
return listener;
} else {
return {};
}
}
void DestroyValueListenerPoller(NT_ValueListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kValueListenerPoller)) {
ii->localStorage.DestroyValueListenerPoller(poller);
}
}
NT_ValueListener AddPolledValueListener(NT_ValueListenerPoller poller,
NT_Handle subentry, unsigned int mask) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kValueListenerPoller)) {
return ii->localStorage.AddPolledValueListener(poller, subentry, mask);
} else {
return {};
}
}
std::vector<ValueNotification> ReadValueListenerQueue(
NT_ValueListenerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kValueListenerPoller)) {
return ii->localStorage.ReadValueListenerQueue(poller);
} else {
return {};
}
}
void RemoveValueListener(NT_ValueListener listener) {
if (auto ii = InstanceImpl::GetTyped(listener, Handle::kValueListener)) {
return ii->localStorage.RemoveValueListener(listener);
}
}
NT_ConnectionListener AddConnectionListener(
NT_Inst inst, bool immediate_notify,
std::function<void(const ConnectionNotification& event)> callback) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->connectionList.AddListener(immediate_notify,
std::move(callback));
} else {
return {};
}
}
bool WaitForConnectionListenerQueue(NT_Handle handle, double timeout) {
if (auto ii = InstanceImpl::GetHandle(handle)) {
return ii->connectionList.WaitForListenerQueue(timeout);
} else {
return {};
}
}
NT_ConnectionListenerPoller CreateConnectionListenerPoller(NT_Inst inst) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->connectionList.CreateListenerPoller();
} else {
return {};
}
}
void DestroyConnectionListenerPoller(NT_ConnectionListenerPoller poller) {
if (auto ii =
InstanceImpl::GetTyped(poller, Handle::kConnectionListenerPoller)) {
return ii->connectionList.DestroyListenerPoller(poller);
}
}
NT_ConnectionListener AddPolledConnectionListener(
NT_ConnectionListenerPoller poller, bool immediate_notify) {
if (auto ii =
InstanceImpl::GetTyped(poller, Handle::kConnectionListenerPoller)) {
return ii->connectionList.AddPolledListener(poller, immediate_notify);
} else {
return {};
}
}
std::vector<ConnectionNotification> ReadConnectionListenerQueue(
NT_ConnectionListenerPoller poller) {
if (auto ii =
InstanceImpl::GetTyped(poller, Handle::kConnectionListenerPoller)) {
return ii->connectionList.ReadListenerQueue(poller);
} else {
return {};
}
}
void RemoveConnectionListener(NT_ConnectionListener listener) {
if (auto ii = InstanceImpl::GetTyped(listener, Handle::kConnectionListener)) {
return ii->connectionList.RemoveListener(listener);
}
}
int64_t Now() {
if (gNowSet) {
return gNowTime;
@@ -796,58 +721,32 @@ bool IsConnected(NT_Inst inst) {
}
}
NT_Logger AddLogger(NT_Inst inst,
std::function<void(const LogMessage& msg)> func,
unsigned int minLevel, unsigned int maxLevel) {
NT_Listener AddLogger(NT_Inst inst, unsigned int minLevel,
unsigned int maxLevel, ListenerCallback func) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
if (minLevel < ii->logger.min_level()) {
ii->logger.set_min_level(minLevel);
}
return ii->logger_impl.Add(std::move(func), minLevel, maxLevel);
auto listener = ii->listenerStorage.AddListener(std::move(func));
ii->logger_impl.AddListener(listener, minLevel, maxLevel);
return listener;
} else {
return {};
}
}
NT_LoggerPoller CreateLoggerPoller(NT_Inst inst) {
if (auto ii = InstanceImpl::GetTyped(inst, Handle::kInstance)) {
return ii->logger_impl.CreatePoller();
} else {
return {};
}
}
void DestroyLoggerPoller(NT_LoggerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kLoggerPoller)) {
ii->logger_impl.DestroyPoller(poller);
}
}
NT_Logger AddPolledLogger(NT_LoggerPoller poller, unsigned int min_level,
unsigned int max_level) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kLoggerPoller)) {
if (min_level < ii->logger.min_level()) {
ii->logger.set_min_level(min_level);
NT_Listener AddPolledLogger(NT_ListenerPoller poller, unsigned int minLevel,
unsigned int maxLevel) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kListenerPoller)) {
if (minLevel < ii->logger.min_level()) {
ii->logger.set_min_level(minLevel);
}
return ii->logger_impl.AddPolled(poller, min_level, max_level);
auto listener = ii->listenerStorage.AddListener(poller);
ii->logger_impl.AddListener(listener, minLevel, maxLevel);
return listener;
} else {
return {};
}
}
std::vector<LogMessage> ReadLoggerQueue(NT_LoggerPoller poller) {
if (auto ii = InstanceImpl::GetTyped(poller, Handle::kLoggerPoller)) {
return ii->logger_impl.ReadQueue(poller);
} else {
return {};
}
}
void RemoveLogger(NT_Logger logger) {
if (auto ii = InstanceImpl::GetTyped(logger, Handle::kLogger)) {
ii->logger_impl.Remove(logger);
ii->logger.set_min_level(ii->logger_impl.GetMinLevel());
}
}
} // namespace nt