Skip to content
This repository has been archived by the owner on Sep 21, 2020. It is now read-only.

Commit

Permalink
Implement keep-alives.
Browse files Browse the repository at this point in the history
  • Loading branch information
PeterJohnson committed Sep 9, 2015
1 parent e1efb73 commit 953a2ce
Show file tree
Hide file tree
Showing 4 changed files with 20 additions and 8 deletions.
2 changes: 1 addition & 1 deletion TODO.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

## Functionality

* Keepalives
* Initial connection flag

## Unit Test Coverage
Expand All @@ -12,6 +11,7 @@
* NetworkCommunication class
* Dispatcher class
* Storage incoming processing
* Keepalives
* Notifiers
* RPC
* C++ API
Expand Down
4 changes: 3 additions & 1 deletion src/Dispatcher.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,9 @@ void DispatcherBase::DispatchThreadMain() {
bool reconnect = false;
for (auto& conn : m_connections) {
// post outgoing messages if connection is active
if (conn->state() == NetworkConnection::kActive) conn->PostOutgoing();
// only send keep-alives on client
if (conn->state() == NetworkConnection::kActive)
conn->PostOutgoing(!m_server);

// if client, reconnect if connection died
if (!m_server && conn->state() == NetworkConnection::kDead)
Expand Down
18 changes: 13 additions & 5 deletions src/NetworkConnection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -250,10 +250,18 @@ void NetworkConnection::QueueOutgoing(std::shared_ptr<Message> msg) {
}
}

void NetworkConnection::PostOutgoing() {
void NetworkConnection::PostOutgoing(bool keep_alive) {
std::lock_guard<std::mutex> lock(m_pending_mutex);
if (m_pending_outgoing.empty()) return;
m_outgoing.emplace(std::move(m_pending_outgoing));
m_pending_outgoing.resize(0);
m_pending_update.resize(0);
auto now = std::chrono::steady_clock::now();
if (m_pending_outgoing.empty()) {
if (!keep_alive) return;
// send keep-alives once a second (if no other messages have been sent)
if ((now - m_last_post) < std::chrono::seconds(1)) return;
m_outgoing.emplace(Outgoing{Message::KeepAlive()});
} else {
m_outgoing.emplace(std::move(m_pending_outgoing));
m_pending_outgoing.resize(0);
m_pending_update.resize(0);
}
m_last_post = now;
}
4 changes: 3 additions & 1 deletion src/NetworkConnection.h
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
#define NT_NETWORKCONNECTION_H_

#include <atomic>
#include <chrono>
#include <memory>
#include <thread>

Expand Down Expand Up @@ -56,7 +57,7 @@ class NetworkConnection {
NetworkStream& stream() { return *m_stream; }

void QueueOutgoing(std::shared_ptr<Message> msg);
void PostOutgoing();
void PostOutgoing(bool keep_alive);

unsigned int uid() const { return m_uid; }

Expand Down Expand Up @@ -95,6 +96,7 @@ class NetworkConnection {
mutable std::mutex m_remote_id_mutex;
std::string m_remote_id;
std::atomic_ullong m_last_update;
std::chrono::steady_clock::time_point m_last_post;

std::mutex m_pending_mutex;
Outgoing m_pending_outgoing;
Expand Down

0 comments on commit 953a2ce

Please sign in to comment.