transport: replace a bunch of bools with a proper state machine

Change-Id: I42f74aebfd714cd8b3be7fcb3068228935482366
This commit is contained in:
Davide Pesavento committed 2022-12-15 04:00:22 -05:00
1 parent 67090837ab
commit 15fa994bf6
5 files changed
+40 -43

No files matched your search

+1 -1
View File
@@ -292,7 +292,7 @@ Face::shutdown()
{
IO_CAPTURE_WEAK_IMPL(post) {
impl->shutdown();
if (m_transport->isConnected())
if (m_transport->getState() != Transport::State::CLOSED)
m_transport->close();
} IO_CAPTURE_WEAK_IMPL_END
}
+2 -2
View File
@@ -304,12 +304,12 @@ public: // IO routine
void
ensureConnected(bool wantResume)
{
if (!m_face.m_transport->isConnected()) {
if (m_face.m_transport->getState() == Transport::State::CLOSED) {
m_face.m_transport->connect(m_face.getIoService(),
[this] (const Block& wire) { m_face.onReceiveElement(wire); });
}
if (wantResume && !m_face.m_transport->isReceiving()) {
if (wantResume) {
m_face.m_transport->resume();
}
}
@@ -55,10 +55,10 @@ public:
void
connect(const typename Protocol::endpoint& endpoint)
{
if (m_isConnecting) {
if (m_transport.getState() == Transport::State::CONNECTING) {
return;
}
m_isConnecting = true;
m_transport.setState(Transport::State::CONNECTING);
// Wait at most 4 seconds to connect
/// @todo Decide whether this number should be configurable
@@ -76,38 +76,30 @@ public:
void
close()
{
m_isConnecting = false;
m_transport.setState(Transport::State::CLOSED);
boost::system::error_code error; // to silently ignore all errors
m_connectTimer.cancel(error);
m_socket.cancel(error);
m_socket.close(error);
m_transport.m_isConnected = false;
m_transport.m_isReceiving = false;
TransmissionQueue{}.swap(m_transmissionQueue); // clear the queue
}
void
pause()
{
if (m_isConnecting)
return;
if (m_transport.m_isReceiving) {
m_transport.m_isReceiving = false;
if (m_transport.getState() == Transport::State::RUNNING) {
m_socket.cancel();
m_transport.setState(Transport::State::PAUSED);
}
}
void
resume()
{
if (m_isConnecting)
return;
if (!m_transport.m_isReceiving) {
m_transport.m_isReceiving = true;
if (m_transport.getState() == Transport::State::PAUSED) {
m_transport.setState(Transport::State::RUNNING);
m_inputBufferSize = 0;
asyncReceive();
}
@@ -118,7 +110,9 @@ public:
{
m_transmissionQueue.push(block);
if (m_transport.m_isConnected && m_transmissionQueue.size() == 1) {
if (m_transport.getState() != Transport::State::CLOSED &&
m_transport.getState() != Transport::State::CONNECTING &&
m_transmissionQueue.size() == 1) {
asyncWrite();
}
// if not connected or there's another transmission in progress (m_transmissionQueue.size() > 1),
@@ -129,16 +123,14 @@ protected:
void
connectHandler(const boost::system::error_code& error)
{
m_isConnecting = false;
m_connectTimer.cancel();
if (error) {
m_transport.m_isConnected = false;
m_transport.close();
NDN_THROW(Transport::Error(error, "error while connecting to the forwarder"));
}
m_transport.m_isConnected = true;
m_transport.setState(Transport::State::PAUSED);
if (!m_transmissionQueue.empty()) {
resume();
@@ -172,8 +164,8 @@ protected:
NDN_THROW(Transport::Error(error, "error while writing data to socket"));
}
if (!m_transport.m_isConnected) {
return; // queue has been already cleared
if (m_transport.getState() == Transport::State::CLOSED) {
return; // queue has already been cleared
}
BOOST_ASSERT(!m_transmissionQueue.empty());
@@ -249,7 +241,6 @@ protected:
size_t m_inputBufferSize = 0;
TransmissionQueue m_transmissionQueue;
boost::asio::steady_timer m_connectTimer;
bool m_isConnecting = false;
};
} // namespace detail
@@ -42,10 +42,10 @@ public:
void
connect(const typename Protocol::resolver::query& query)
{
if (this->m_isConnecting) {
if (this->m_transport.getState() == Transport::State::CONNECTING) {
return;
}
this->m_isConnecting = true;
this->m_transport.setState(Transport::State::CONNECTING);
// Wait at most 4 seconds to connect
/// @todo Decide whether this number should be configurable
+22 -16
View File
@@ -30,7 +30,8 @@
namespace ndn {
/** \brief Provides TLV-block delivery service.
/**
* \brief Provides a "TLV-oriented" delivery service.
*/
class Transport : noncopyable
{
@@ -43,9 +44,16 @@ public:
Error(const boost::system::error_code& code, const std::string& msg);
};
using ReceiveCallback = std::function<void(const Block& wire)>;
using ErrorCallback = std::function<void()>;
enum class State {
CLOSED,
CONNECTING,
RUNNING,
PAUSED,
};
using ReceiveCallback = std::function<void(const Block&)>;
public:
virtual
~Transport() = default;
@@ -89,29 +97,27 @@ public:
resume() = 0;
/**
* \brief Return whether the transport is connected.
* \brief Return the current state of the transport.
*/
bool
isConnected() const noexcept
State
getState() const noexcept
{
return m_isConnected;
return m_state;
}
/**
* \retval true incoming packets are expected, the receive callback will be invoked
* \retval false incoming packets are not expected, the receive callback will not be invoked
*/
bool
isReceiving() const noexcept
protected:
void
setState(State state) noexcept
{
return m_isReceiving;
m_state = state;
}
protected:
boost::asio::io_service* m_ioService = nullptr;
ReceiveCallback m_receiveCallback;
bool m_isConnected = false;
bool m_isReceiving = false;
private:
State m_state = State::CLOSED;
};
} // namespace ndn