1+ // Copyright (c) 2026, Grid Protection Alliance. Licensed under the MIT License.
2+ #include " ../../lib/transport/SubscriberInstance.h"
3+ #include < atomic>
4+ #include < chrono>
5+ #include < iostream>
6+ #include < memory>
7+ #include < thread>
8+ using namespace sttp ;
9+ using namespace sttp ::transport;
10+ using boost::asio::ip::tcp;
11+ using namespace std ::chrono_literals;
12+
13+ struct State
14+ {
15+ std::atomic<int > connections{0 }, metadata{0 }, errors{0 };
16+ };
17+
18+ class TestSubscriber final : public SubscriberInstance
19+ {
20+ State& m_state;
21+ void ConnectionEstablished () override { ++m_state.connections ; }
22+ void ReceivedMetadata (const std::vector<uint8_t >&) override { ++m_state.metadata ; }
23+ void StatusMessage (const std::string&) override {}
24+ void ErrorMessage (const std::string&) override { ++m_state.errors ; }
25+ public:
26+ explicit TestSubscriber (State& state) : m_state(state) {}
27+ };
28+
29+ void Require (bool value, const char * message)
30+ {
31+ if (!value) throw std::runtime_error (message);
32+ }
33+
34+ template <typename Predicate> void Wait (Predicate predicate)
35+ {
36+ const auto deadline = std::chrono::steady_clock::now () + 5s;
37+ while (!predicate ())
38+ {
39+ Require (std::chrono::steady_clock::now () < deadline, " Timed out waiting for subscriber"
40+ std::this_thread::sleep_for (1ms);
41+ }
42+ }
43+
44+ // Read complete command frames with a deadline, including fragmented TCP reads.
45+ std::vector<uint8_t > ReadCommand (tcp::socket& socket, std::chrono::milliseconds timeout)
46+ {
47+ const auto deadline = std::chrono::steady_clock::now () + timeout;
48+ std::vector<uint8_t > packet;
49+ size_t expected = 4 ;
50+ while (packet.size () < expected)
51+ {
52+ if (socket.available ())
53+ {
54+ uint8_t data[4096 ];
55+ const size_t count = socket.read_some (boost::asio::buffer (data, std::min (sizeof (data), expected - packet.size ())));
56+ packet.insert (packet.end (), data, data + count);
57+ if (packet.size () == 4 )
58+ {
59+ expected = 4 + (uint32_t (packet[0 ]) << 24 | uint32_t (packet[1 ]) << 16 | uint32_t (packet[2 ]) << 8 | packet[3 ]);
60+ Require (expected > 4 && expected <= 65536 , " Invalid command frame length"
61+ }
62+ }
63+ else
64+ {
65+ if (std::chrono::steady_clock::now () >= deadline)
66+ {
67+ Require (packet.empty (), " Incomplete command frame"
68+ return {};
69+ }
70+ std::this_thread::sleep_for (1ms);
71+ }
72+ }
73+ return packet;
74+ }
75+
76+ void Respond (tcp::socket& socket, uint8_t command, bool accepted = true )
77+ {
78+ // Four-byte frame size, response code, command code, four-byte body size.
79+ const uint8_t packet[] = {0 , 0 , 0 , 6 , uint8_t (accepted ? 0x80 : 0x81 ), command, 0 , 0 , 0 , 0 };
80+ boost::asio::write (socket, boost::asio::buffer (packet));
81+ }
82+
83+ void Run (bool reverse, bool parse, int version, const std::string& handshake, bool reconnect)
84+ {
85+ boost::asio::io_context service;
86+ tcp::acceptor acceptor (service, tcp::endpoint (boost::asio::ip::address_v4::loopback (), 0 ));
87+ const uint16_t port = acceptor.local_endpoint ().port ();
88+ State state; // Outlives callbacks and subscriber destruction.
89+ auto subscriber = std::make_unique<TestSubscriber>(state);
90+ subscriber->Initialize (" 127.0.0.1"
91+ subscriber->SetAutoReconnect (reconnect);
92+ subscriber->SetRetryInterval (50 );
93+ subscriber->SetAutoParseMetadata (parse);
94+ subscriber->SetVersion (static_cast <uint8_t >(version));
95+ subscriber->SetOperationalModesResponseTimeout (500 );
96+ if (reverse)
97+ {
98+ acceptor.close ();
99+ Require (subscriber->Listen (" 127.0.0.1" " Reverse listener failed"
100+ }
101+ else
102+ {
103+ acceptor.non_blocking (true );
104+ subscriber->ConnectAsync ();
105+ }
106+
107+ const int attempts = reconnect ? 2 : 1 ;
108+ for (int attempt = 1 ; attempt <= attempts; ++attempt)
109+ {
110+ tcp::socket socket (service);
111+ if (reverse)
112+ socket.connect (tcp::endpoint (boost::asio::ip::address_v4::loopback (), port));
113+ else
114+ Wait ([&]
115+ {
116+ boost::system::error_code error;
117+ acceptor.accept (socket, error);
118+ if (error == boost::asio::error::would_block || error == boost::asio::error::try_again) return false ;
119+ if (error) throw boost::system::system_error (error);
120+ return true ;
121+ });
122+ auto packet = ReadCommand (socket, 5s);
123+ Require (!packet.empty () && packet[4 ] == 0x06 , " Expected operational modes first"
124+ if (version > 2 )
125+ {
126+ Require (ReadCommand (socket, 100ms).empty (), " Startup command precedes handshake acceptance"
127+ Require (state.connections == attempt - 1 , " Connection event precedes handshake acceptance"
128+ if (handshake != " timeout" Respond (socket, 0x06 , handshake != " reject"
129+ if (handshake != " accept"
130+ {
131+ Wait ([&] { return state.errors > 0 ; });
132+ Require (ReadCommand (socket, 200ms).empty (), " Startup command follows failed handshake"
133+ Require (state.connections == 0 && state.metadata == 0 , " Failed handshake starts subscriber"
134+ subscriber.reset ();
135+ break ;
136+ }
137+ }
138+ packet = ReadCommand (socket, 5s);
139+ Require (!packet.empty () && packet[4 ] == (parse ? 0x01 : 0x02 ), " Incorrect initial startup command"
140+ if (parse)
141+ {
142+ Respond (socket, 0x01 );
143+ packet = ReadCommand (socket, 5s);
144+ Require (!packet.empty () && packet[4 ] == 0x02 , " Metadata does not trigger subscription"
145+ }
146+ Require (ReadCommand (socket, 200ms).empty (), " Duplicate metadata request or subscription"
147+ Require (state.connections == attempt, " Duplicate or missing connection event"
148+ Require (state.metadata == (parse ? attempt : 0 ), " Duplicate or missing metadata callback"
149+ if (attempt == attempts) subscriber.reset ();
150+ // Closing the publisher-side socket triggers real automatic reconnect on the next iteration.
151+ }
152+ std::cout << " PASS " " reverse" " outbound" " parse="
153+ << " version=" " handshake=" " reconnect="
154+ }
155+
156+ int main ()
157+ {
158+ try
159+ {
160+ for (bool reverse : {false , true })
161+ {
162+ for (bool parse : {false , true })
163+ for (int version : {2 , 3 })
164+ Run (reverse, parse, version, " accept"
165+ Run (reverse, true , 3 , " reject" false );
166+ Run (reverse, true , 3 , " timeout" false );
167+ }
168+ return 0 ;
169+ }
170+ catch (const std::exception& ex)
171+ {
172+ std::cerr << " FAIL: " what () << std::endl;
173+ return 1 ;
174+ }
175+ }
0 commit comments