1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
// Copyright Epic Games, Inc. All Rights Reserved.
#pragma once
#include "hordetransport.h"
#include <zencore/logbase.h>
ZEN_THIRD_PARTY_INCLUDES_START
#include <asio.hpp>
ZEN_THIRD_PARTY_INCLUDES_END
#if ZEN_PLATFORM_WINDOWS
# undef SendMessage
#endif
#include <deque>
#include <functional>
#include <memory>
#include <system_error>
#include <unordered_map>
#include <vector>
namespace zen::horde {
class AsyncComputeTransport;
/** Handler called when a data frame arrives for a channel. */
using FrameHandler = std::function<void(std::vector<uint8_t> Data)>;
/** Handler called when a channel is detached by the remote peer. */
using DetachHandler = std::function<void()>;
/** Handler for async send completion. */
using SendHandler = std::function<void(const std::error_code&)>;
/** Async multiplexed socket that routes data between channels over a single transport.
*
* Uses an async recv pump, a serialized send queue, and a periodic ping timer —
* all running on a shared io_context.
*
* Wire format per frame: [channelId(4B)][size(4B)][data].
* Control messages use negative sizes: -2 = detach, -3 = ping.
*/
class AsyncComputeSocket : public std::enable_shared_from_this<AsyncComputeSocket>
{
public:
AsyncComputeSocket(std::unique_ptr<AsyncComputeTransport> Transport, asio::io_context& IoContext);
~AsyncComputeSocket();
AsyncComputeSocket(const AsyncComputeSocket&) = delete;
AsyncComputeSocket& operator=(const AsyncComputeSocket&) = delete;
/** Register callbacks for a channel. Must be called before StartRecvPump(). */
void RegisterChannel(int ChannelId, FrameHandler OnFrame, DetachHandler OnDetach);
/** Begin the async recv pump and ping timer. */
void StartRecvPump();
/** Enqueue a data frame for async transmission. */
void AsyncSendFrame(int ChannelId, std::vector<uint8_t> Data, SendHandler Handler = {});
/** Send a control frame (detach) for a channel. */
void AsyncSendDetach(int ChannelId, SendHandler Handler = {});
/** Close the transport and cancel all pending operations. */
void Close();
private:
struct FrameHeader
{
int32_t Channel = 0;
int32_t Size = 0;
};
struct PendingWrite
{
FrameHeader Header;
std::vector<uint8_t> Data;
SendHandler Handler;
};
static constexpr int32_t ControlDetach = -2;
static constexpr int32_t ControlPing = -3;
LoggerRef Log() { return m_Log; }
void DoRecvHeader();
void DoRecvPayload(FrameHeader Header);
void FlushNextSend();
void StartPingTimer();
void HandleError();
LoggerRef m_Log;
std::unique_ptr<AsyncComputeTransport> m_Transport;
asio::strand<asio::any_io_executor> m_Strand;
asio::steady_timer m_PingTimer;
std::unordered_map<int, FrameHandler> m_FrameHandlers;
std::unordered_map<int, DetachHandler> m_DetachHandlers;
FrameHeader m_RecvHeader;
std::deque<PendingWrite> m_SendQueue;
bool m_Closed = false;
};
} // namespace zen::horde
|