Add plugin receive loop & remote player state
Add a networking receive loop and in-plugin remote-player storage. Introduces F4TRemotePlayerState (header + implementation) to track assigned playerId and remote player snapshots with thread-safe access. Expands F4TNetworking to start/stop a background receive thread, parse newline-separated JSON packets (welcome, transform, disconnect), validate fields, update remote state, throttle logs, and handle socket/thread synchronization. Add nlohmann_json to xmake and plugin build. Update docs and dev log to reflect the new receive behavior and current prototype scope.
This commit is contained in:
@@ -1,15 +1,26 @@
|
||||
#include "pch.h"
|
||||
|
||||
#include "F4TNetworking.h"
|
||||
#include "F4TRemotePlayerState.h"
|
||||
|
||||
#include <WinSock2.h>
|
||||
#include <WS2tcpip.h>
|
||||
|
||||
#include <array>
|
||||
#include <atomic>
|
||||
#include <charconv>
|
||||
#include <chrono>
|
||||
#include <cstddef>
|
||||
#include <cstdio>
|
||||
#include <limits>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <unordered_map>
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
namespace
|
||||
{
|
||||
@@ -17,19 +28,37 @@ namespace
|
||||
constexpr auto kLocalServerPort = 7777;
|
||||
constexpr auto kReconnectCooldown = 10s;
|
||||
constexpr auto kConnectTimeoutMilliseconds = 100;
|
||||
constexpr auto kReceiveIdleSleep = 25ms;
|
||||
constexpr auto kRemoteUpdateLogInterval = 1s;
|
||||
constexpr auto kWarningLogInterval = 2s;
|
||||
constexpr auto kMaximumReceiveBufferSize = 64uz * 1024uz;
|
||||
|
||||
SOCKET g_socket = INVALID_SOCKET;
|
||||
std::mutex g_socketMutex;
|
||||
bool g_winsockInitialized = false;
|
||||
auto g_lastConnectAttempt = std::chrono::steady_clock::time_point{};
|
||||
std::atomic_bool g_receiveThreadRunning{ false };
|
||||
std::thread g_receiveThread;
|
||||
std::unordered_map<std::uint32_t, std::chrono::steady_clock::time_point> g_lastRemoteUpdateLogTimes;
|
||||
std::unordered_map<std::string, std::chrono::steady_clock::time_point> g_lastWarningLogTimes;
|
||||
|
||||
using Json = nlohmann::json;
|
||||
|
||||
void CloseSocket()
|
||||
{
|
||||
const std::scoped_lock lock(g_socketMutex);
|
||||
if (g_socket != INVALID_SOCKET) {
|
||||
closesocket(g_socket);
|
||||
g_socket = INVALID_SOCKET;
|
||||
}
|
||||
}
|
||||
|
||||
SOCKET GetSocket()
|
||||
{
|
||||
const std::scoped_lock lock(g_socketMutex);
|
||||
return g_socket;
|
||||
}
|
||||
|
||||
bool EnsureWinsockInitialized()
|
||||
{
|
||||
if (g_winsockInitialized) {
|
||||
@@ -92,6 +121,321 @@ namespace
|
||||
g_lastConnectAttempt = now;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool ShouldLogWarning(std::string a_key)
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
const auto lastLog = g_lastWarningLogTimes.find(a_key);
|
||||
if (lastLog != g_lastWarningLogTimes.end() && now - lastLog->second < kWarningLogInterval) {
|
||||
return false;
|
||||
}
|
||||
|
||||
g_lastWarningLogTimes[std::move(a_key)] = now;
|
||||
return true;
|
||||
}
|
||||
|
||||
void LogThrottledWarning(const std::string& a_key, const char* a_message)
|
||||
{
|
||||
if (ShouldLogWarning(a_key)) {
|
||||
REX::WARN("{}", a_message);
|
||||
}
|
||||
}
|
||||
|
||||
void TrimPacketLine(std::string& a_line)
|
||||
{
|
||||
while (!a_line.empty() && (a_line.back() == '\r' || a_line.back() == ' ' || a_line.back() == '\t')) {
|
||||
a_line.pop_back();
|
||||
}
|
||||
}
|
||||
|
||||
std::optional<std::uint32_t> ReadPlayerId(const Json& a_packet)
|
||||
{
|
||||
const auto playerId = a_packet.find("playerId");
|
||||
if (playerId == a_packet.end()) {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
try {
|
||||
if (playerId->is_number_integer()) {
|
||||
const auto value = playerId->get<std::int64_t>();
|
||||
if (value >= 0 && value <= (std::numeric_limits<std::uint32_t>::max)()) {
|
||||
return static_cast<std::uint32_t>(value);
|
||||
}
|
||||
}
|
||||
|
||||
if (playerId->is_string()) {
|
||||
const auto text = playerId->get<std::string>();
|
||||
std::uint32_t value = 0;
|
||||
const auto* begin = text.data();
|
||||
const auto* end = begin + text.size();
|
||||
const auto result = std::from_chars(begin, end, value);
|
||||
if (result.ec == std::errc{} && result.ptr == end) {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
} catch (const std::exception&) {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
std::optional<double> ReadDouble(const Json& a_packet, const char* a_fieldName)
|
||||
{
|
||||
const auto field = a_packet.find(a_fieldName);
|
||||
if (field == a_packet.end()) {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
try {
|
||||
if (field->is_number()) {
|
||||
return field->get<double>();
|
||||
}
|
||||
|
||||
if (field->is_string()) {
|
||||
const auto text = field->get<std::string>();
|
||||
std::size_t parsedCharacters = 0;
|
||||
const auto value = std::stod(text, std::addressof(parsedCharacters));
|
||||
if (parsedCharacters == text.size()) {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
} catch (const std::exception&) {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
std::optional<std::string> ReadString(const Json& a_packet, const char* a_fieldName)
|
||||
{
|
||||
const auto field = a_packet.find(a_fieldName);
|
||||
if (field == a_packet.end() || !field->is_string()) {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
return field->get<std::string>();
|
||||
}
|
||||
|
||||
bool ShouldLogRemoteUpdate(std::uint32_t a_playerId)
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
const auto lastLog = g_lastRemoteUpdateLogTimes.find(a_playerId);
|
||||
if (lastLog != g_lastRemoteUpdateLogTimes.end() && now - lastLog->second < kRemoteUpdateLogInterval) {
|
||||
return false;
|
||||
}
|
||||
|
||||
g_lastRemoteUpdateLogTimes[a_playerId] = now;
|
||||
return true;
|
||||
}
|
||||
|
||||
void HandleWelcomePacket(const Json& a_packet)
|
||||
{
|
||||
const auto playerId = ReadPlayerId(a_packet);
|
||||
if (!playerId) {
|
||||
LogThrottledWarning("welcome_playerId", "Ignoring welcome packet without a valid playerId.");
|
||||
return;
|
||||
}
|
||||
|
||||
F4T::RemotePlayerState::SetAssignedPlayerId(*playerId);
|
||||
REX::INFO("Assigned server playerId: {}", *playerId);
|
||||
}
|
||||
|
||||
void HandleTransformPacket(const Json& a_packet)
|
||||
{
|
||||
const auto playerId = ReadPlayerId(a_packet);
|
||||
const auto x = ReadDouble(a_packet, "x");
|
||||
const auto y = ReadDouble(a_packet, "y");
|
||||
const auto z = ReadDouble(a_packet, "z");
|
||||
const auto angleZ = ReadDouble(a_packet, "angleZ");
|
||||
const auto movementType = ReadString(a_packet, "movementType");
|
||||
const auto cellId = ReadString(a_packet, "cellId");
|
||||
const auto worldspaceId = ReadString(a_packet, "worldspaceId");
|
||||
|
||||
if (!playerId || !x || !y || !z || !angleZ || !movementType || !cellId || !worldspaceId) {
|
||||
LogThrottledWarning("transform_invalid", "Ignoring transform packet with missing or invalid required fields.");
|
||||
return;
|
||||
}
|
||||
|
||||
if (F4T::RemotePlayerState::IsAssignedPlayerId(*playerId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
F4T::RemotePlayerState::RemotePlayerState remoteState{
|
||||
*playerId,
|
||||
static_cast<float>(*x),
|
||||
static_cast<float>(*y),
|
||||
static_cast<float>(*z),
|
||||
static_cast<float>(*angleZ),
|
||||
*movementType,
|
||||
*cellId,
|
||||
*worldspaceId,
|
||||
ReadDouble(a_packet, "clientTime"),
|
||||
ReadDouble(a_packet, "serverTime"),
|
||||
std::chrono::steady_clock::now()
|
||||
};
|
||||
|
||||
F4T::RemotePlayerState::UpdateRemotePlayer(remoteState);
|
||||
|
||||
// TODO: A later milestone can read this plain state from the game thread and
|
||||
// spawn or move remote actors. The networking thread must not touch actors.
|
||||
if (ShouldLogRemoteUpdate(*playerId)) {
|
||||
REX::INFO(
|
||||
"Remote player {} updated: X={:.2f}, Y={:.2f}, Z={:.2f}, AngleZ={:.2f}, Cell={}, Worldspace={}",
|
||||
remoteState.playerId,
|
||||
remoteState.x,
|
||||
remoteState.y,
|
||||
remoteState.z,
|
||||
remoteState.angleZ,
|
||||
remoteState.cellId,
|
||||
remoteState.worldspaceId);
|
||||
}
|
||||
}
|
||||
|
||||
void HandleDisconnectPacket(const Json& a_packet)
|
||||
{
|
||||
const auto playerId = ReadPlayerId(a_packet);
|
||||
if (!playerId) {
|
||||
LogThrottledWarning("disconnect_playerId", "Ignoring disconnect packet without a valid playerId.");
|
||||
return;
|
||||
}
|
||||
|
||||
g_lastRemoteUpdateLogTimes.erase(*playerId);
|
||||
|
||||
if (F4T::RemotePlayerState::RemoveRemotePlayer(*playerId)) {
|
||||
REX::INFO("Remote player {} disconnected and was removed.", *playerId);
|
||||
} else {
|
||||
REX::INFO("Received disconnect for unknown remote player {}.", *playerId);
|
||||
}
|
||||
}
|
||||
|
||||
void HandleUnknownPacket(const Json& a_packet)
|
||||
{
|
||||
const auto type = ReadString(a_packet, "type").value_or("<missing>");
|
||||
if (ShouldLogWarning("unknown_" + type)) {
|
||||
REX::WARN("Ignoring unknown server packet type: {}", type);
|
||||
}
|
||||
}
|
||||
|
||||
void HandlePacketLine(const std::string& a_line)
|
||||
{
|
||||
if (a_line.empty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
Json packet;
|
||||
try {
|
||||
packet = Json::parse(a_line);
|
||||
} catch (const std::exception& a_error) {
|
||||
if (ShouldLogWarning("json_parse")) {
|
||||
REX::WARN("Ignoring invalid JSON packet from server: {}", a_error.what());
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (!packet.is_object()) {
|
||||
LogThrottledWarning("json_object", "Ignoring server packet that was not a JSON object.");
|
||||
return;
|
||||
}
|
||||
|
||||
const auto packetType = ReadString(packet, "type");
|
||||
if (!packetType) {
|
||||
LogThrottledWarning("packet_type", "Ignoring server packet without a string type field.");
|
||||
return;
|
||||
}
|
||||
|
||||
if (*packetType == "welcome") {
|
||||
HandleWelcomePacket(packet);
|
||||
} else if (*packetType == "transform") {
|
||||
HandleTransformPacket(packet);
|
||||
} else if (*packetType == "disconnect") {
|
||||
HandleDisconnectPacket(packet);
|
||||
} else {
|
||||
HandleUnknownPacket(packet);
|
||||
}
|
||||
}
|
||||
|
||||
void ProcessReceiveBuffer(std::string& a_receiveBuffer)
|
||||
{
|
||||
std::size_t newlinePosition = 0;
|
||||
while ((newlinePosition = a_receiveBuffer.find('\n')) != std::string::npos) {
|
||||
auto line = a_receiveBuffer.substr(0, newlinePosition);
|
||||
a_receiveBuffer.erase(0, newlinePosition + 1);
|
||||
TrimPacketLine(line);
|
||||
HandlePacketLine(line);
|
||||
}
|
||||
|
||||
if (a_receiveBuffer.size() > kMaximumReceiveBufferSize) {
|
||||
a_receiveBuffer.clear();
|
||||
LogThrottledWarning("receive_buffer", "Cleared oversized server receive buffer.");
|
||||
}
|
||||
}
|
||||
|
||||
void ReceiveLoop()
|
||||
{
|
||||
std::array<char, 4096> receiveChunk{};
|
||||
std::string receiveBuffer;
|
||||
|
||||
while (g_receiveThreadRunning.load()) {
|
||||
int bytesReceived = 0;
|
||||
{
|
||||
const std::scoped_lock lock(g_socketMutex);
|
||||
if (g_socket == INVALID_SOCKET) {
|
||||
break;
|
||||
}
|
||||
|
||||
bytesReceived = recv(g_socket, receiveChunk.data(), static_cast<int>(receiveChunk.size()), 0);
|
||||
}
|
||||
|
||||
if (bytesReceived > 0) {
|
||||
receiveBuffer.append(receiveChunk.data(), static_cast<std::size_t>(bytesReceived));
|
||||
ProcessReceiveBuffer(receiveBuffer);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (bytesReceived == 0) {
|
||||
REX::WARN("Fallout 4 Together local server closed the connection.");
|
||||
CloseSocket();
|
||||
break;
|
||||
}
|
||||
|
||||
const auto error = WSAGetLastError();
|
||||
if (error == WSAEWOULDBLOCK) {
|
||||
std::this_thread::sleep_for(kReceiveIdleSleep);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (g_receiveThreadRunning.load()) {
|
||||
REX::WARN("Lost connection to Fallout 4 Together local server while receiving.");
|
||||
}
|
||||
CloseSocket();
|
||||
break;
|
||||
}
|
||||
|
||||
g_receiveThreadRunning.store(false);
|
||||
}
|
||||
|
||||
void StopReceiveThread()
|
||||
{
|
||||
g_receiveThreadRunning.store(false);
|
||||
CloseSocket();
|
||||
|
||||
if (g_receiveThread.joinable() && g_receiveThread.get_id() != std::this_thread::get_id()) {
|
||||
g_receiveThread.join();
|
||||
}
|
||||
}
|
||||
|
||||
void StartReceiveThread()
|
||||
{
|
||||
if (g_receiveThread.joinable()) {
|
||||
g_receiveThread.join();
|
||||
}
|
||||
|
||||
// Receiving server packets is deliberately isolated from Fallout 4 objects.
|
||||
// This thread only parses network data and updates plain remote-player state.
|
||||
g_receiveThreadRunning.store(true);
|
||||
g_receiveThread = std::thread(ReceiveLoop);
|
||||
}
|
||||
}
|
||||
|
||||
namespace F4T::Networking
|
||||
@@ -146,16 +490,26 @@ namespace F4T::Networking
|
||||
return false;
|
||||
}
|
||||
|
||||
g_socket = localSocket;
|
||||
{
|
||||
const std::scoped_lock lock(g_socketMutex);
|
||||
g_socket = localSocket;
|
||||
}
|
||||
|
||||
F4T::RemotePlayerState::ClearAssignedPlayerId();
|
||||
F4T::RemotePlayerState::ClearRemotePlayers();
|
||||
g_lastRemoteUpdateLogTimes.clear();
|
||||
StartReceiveThread();
|
||||
|
||||
REX::INFO("Connected to Fallout 4 Together local server.");
|
||||
// TODO: Read and log the server welcome packet once the plugin has a small
|
||||
// non-blocking receive path. For now the server owns player IDs entirely.
|
||||
return true;
|
||||
}
|
||||
|
||||
void DisconnectFromLocalServer()
|
||||
{
|
||||
CloseSocket();
|
||||
StopReceiveThread();
|
||||
F4T::RemotePlayerState::ClearAssignedPlayerId();
|
||||
F4T::RemotePlayerState::ClearRemotePlayers();
|
||||
g_lastRemoteUpdateLogTimes.clear();
|
||||
|
||||
if (g_winsockInitialized) {
|
||||
WSACleanup();
|
||||
@@ -165,7 +519,7 @@ namespace F4T::Networking
|
||||
|
||||
bool IsConnectedToServer()
|
||||
{
|
||||
return g_socket != INVALID_SOCKET;
|
||||
return GetSocket() != INVALID_SOCKET;
|
||||
}
|
||||
|
||||
bool SendTransformPacket(
|
||||
@@ -244,7 +598,16 @@ namespace F4T::Networking
|
||||
|
||||
packetSize += appended;
|
||||
|
||||
const auto bytesSent = send(g_socket, packet.data(), packetSize, 0);
|
||||
int bytesSent = 0;
|
||||
{
|
||||
const std::scoped_lock lock(g_socketMutex);
|
||||
if (g_socket == INVALID_SOCKET) {
|
||||
return false;
|
||||
}
|
||||
|
||||
bytesSent = send(g_socket, packet.data(), packetSize, 0);
|
||||
}
|
||||
|
||||
if (bytesSent == SOCKET_ERROR) {
|
||||
const auto error = WSAGetLastError();
|
||||
if (error != WSAEWOULDBLOCK) {
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
#include "pch.h"
|
||||
|
||||
#include "F4TRemotePlayerState.h"
|
||||
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
namespace
|
||||
{
|
||||
std::mutex g_stateMutex;
|
||||
std::optional<std::uint32_t> g_assignedPlayerId;
|
||||
std::unordered_map<std::uint32_t, F4T::RemotePlayerState::RemotePlayerState> g_remotePlayers;
|
||||
}
|
||||
|
||||
namespace F4T::RemotePlayerState
|
||||
{
|
||||
void SetAssignedPlayerId(std::uint32_t a_playerId)
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
g_assignedPlayerId = a_playerId;
|
||||
}
|
||||
|
||||
std::optional<std::uint32_t> GetAssignedPlayerId()
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
return g_assignedPlayerId;
|
||||
}
|
||||
|
||||
bool IsAssignedPlayerId(std::uint32_t a_playerId)
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
return g_assignedPlayerId.has_value() && *g_assignedPlayerId == a_playerId;
|
||||
}
|
||||
|
||||
void ClearAssignedPlayerId()
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
g_assignedPlayerId.reset();
|
||||
}
|
||||
|
||||
void UpdateRemotePlayer(RemotePlayerState a_state)
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
g_remotePlayers[a_state.playerId] = std::move(a_state);
|
||||
}
|
||||
|
||||
bool RemoveRemotePlayer(std::uint32_t a_playerId)
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
return g_remotePlayers.erase(a_playerId) > 0;
|
||||
}
|
||||
|
||||
void ClearRemotePlayers()
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
g_remotePlayers.clear();
|
||||
}
|
||||
|
||||
std::vector<RemotePlayerState> GetRemotePlayerSnapshot()
|
||||
{
|
||||
const std::scoped_lock lock(g_stateMutex);
|
||||
|
||||
std::vector<RemotePlayerState> snapshot;
|
||||
snapshot.reserve(g_remotePlayers.size());
|
||||
for (const auto& [_, player] : g_remotePlayers) {
|
||||
snapshot.push_back(player);
|
||||
}
|
||||
|
||||
return snapshot;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user