Don't send heartbeats during log replay

fixes #8461
refs #8485

diff --git a/lib/base/tlsstream.cpp b/lib/base/tlsstream.cpp
index 65630a828147041d9313b64df35ed2eb6cec7dde..c4fb0d72d78367cf6497dc7f357084b3e12a29c3 100644 (file)
--- a/lib/base/tlsstream.cpp
+++ b/lib/base/tlsstream.cpp
@@ -211,6 +211,7 @@ void TlsStream::OnEvent(int revents)
 
                        m_SSL.reset();
                        m_Socket->Close();
+                       m_Socket.reset();
 
                        m_Eof = true;
 
@@ -222,6 +223,7 @@ void TlsStream::OnEvent(int revents)
 
                        m_SSL.reset();
                        m_Socket->Close();
+                       m_Socket.reset();
 
                        m_ErrorCode = ERR_peek_error();
                        m_ErrorOccurred = true;
@@ -284,6 +286,8 @@ void TlsStream::Write(const void *buffer, size_t count)
  */
 void TlsStream::Close(void)
 {
+       SocketEvents::Unregister();
+
        boost::mutex::scoped_lock lock(m_Mutex);
 
        if (!m_SSL)
@@ -292,9 +296,8 @@ void TlsStream::Close(void)
        (void) SSL_shutdown(m_SSL.get());
        m_SSL.reset();
 
-       SocketEvents::Unregister();
-
        m_Socket->Close();
+       m_Socket.reset();
 
        m_Eof = true;
 }

diff --git a/lib/remote/apiclient-heartbeat.cpp b/lib/remote/apiclient-heartbeat.cpp
index 618dbbfdf435263d96621b651c5e0de327e66394..066a6102616b1600eb8e77599206a363f36c8ffe 100644 (file)
--- a/lib/remote/apiclient-heartbeat.cpp
+++ b/lib/remote/apiclient-heartbeat.cpp
@@ -45,6 +45,12 @@ void ApiClient::HeartbeatTimerHandler(void)
 {
        BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjectsByType<Endpoint>()) {
                BOOST_FOREACH(const ApiClient::Ptr& client, endpoint->GetClients()) {
+                       if (endpoint->GetSyncing()) {
+                               Log(LogInformation, "ApiClient")
+                                   << "Not sending heartbeat for endpoint '" << endpoint->GetName() << "' because we're replaying the log for it.";
+                               continue;
+                       }
+
                        if (client->m_NextHeartbeat != 0 && client->m_NextHeartbeat < Utility::GetTime()) {
                                Log(LogWarning, "ApiClient")
                                    << "Client for endpoint '" << endpoint->GetName() << "' has requested "

diff --git a/lib/remote/apiclient.cpp b/lib/remote/apiclient.cpp
index e28148ff4752026ed19d74f0bd99c7976d347837..f3a8b01035171030223f44c61823d768c032abc3 100644 (file)
--- a/lib/remote/apiclient.cpp
+++ b/lib/remote/apiclient.cpp
@@ -259,7 +259,7 @@ Value RequestCertificateHandler(const MessageOrigin& origin, const Dictionary::P
 
 void ApiClient::TimeoutTimerHandler(void)
 {
-       if (m_Seen < Utility::GetTime() - 60) {
+       if (m_Seen < Utility::GetTime() - 60 && !m_Endpoint->GetSyncing()) {
                /* Obtain a strong reference to ourselves because Disconnect otherwise removes the last reference */
                ApiClient::Ptr self = this;
 

diff --git a/lib/remote/apilistener.cpp b/lib/remote/apilistener.cpp
index a62bc7edf14180b29494ad389f926fe5be6e24e6..d4c7284976b258cd6c376f755bc31c9a21e5b94e 100644 (file)
--- a/lib/remote/apilistener.cpp
+++ b/lib/remote/apilistener.cpp
@@ -20,6 +20,7 @@
 #include "remote/apilistener.hpp"
 #include "remote/apiclient.hpp"
 #include "remote/endpoint.hpp"
+#include "remote/jsonrpc.hpp"
 #include "base/convert.hpp"
 #include "base/netstring.hpp"
 #include "base/json.hpp"
@@ -659,6 +660,7 @@ void ApiListener::ReplayLog(const ApiClient::Ptr& client)
 
        int count = -1;
        double peer_ts = endpoint->GetLocalLogPosition();
+       double logpos_ts = peer_ts;
        bool last_sync = false;
        
        Endpoint::Ptr target_endpoint = client->GetEndpoint();
@@ -697,7 +699,7 @@ void ApiListener::ReplayLog(const ApiClient::Ptr& client)
                        Log(LogNotice, "ApiListener")
                            << "Replaying log: " << path;
 
-                       std::fstream *fp = new std::fstream(path.CStr(), std::fstream::in);
+                       std::fstream *fp = new std::fstream(path.CStr(), std::fstream::in | std::fstream::binary);
                        StdioStream::Ptr logStream = new StdioStream(fp, true);
 
                        String message;
@@ -747,6 +749,20 @@ void ApiListener::ReplayLog(const ApiClient::Ptr& client)
                                count++;
 
                                peer_ts = pmessage->Get("timestamp");
+
+                               if (ts > logpos_ts + 10) {
+                                       logpos_ts = ts;
+
+                                       Dictionary::Ptr lparams = new Dictionary();
+                                       lparams->Set("log_position", logpos_ts);
+
+                                       Dictionary::Ptr lmessage = new Dictionary();
+                                       lmessage->Set("jsonrpc", "2.0");
+                                       lmessage->Set("method", "log::SetLogPosition");
+                                       lmessage->Set("params", lparams);
+
+                                       JsonRpc::SendMessage(client->GetStream(), lmessage);
+                               }
                        }
 
                        logStream->Close();