if (!GetBindPort().IsEmpty())
AddListener(GetBindPort());
- m_ReconnectTimer = boost::make_shared<Timer>();
- m_ReconnectTimer->OnTimerExpired.connect(boost::bind(&ClusterComponent::ReconnectTimerHandler, this));
- m_ReconnectTimer->SetInterval(5);
- m_ReconnectTimer->Start();
+ m_ClusterTimer = boost::make_shared<Timer>();
+ m_ClusterTimer->OnTimerExpired.connect(boost::bind(&ClusterComponent::ClusterTimerHandler, this));
+ m_ClusterTimer->SetInterval(5);
+ m_ClusterTimer->Start();
Service::OnNewCheckResult.connect(bind(&ClusterComponent::CheckResultHandler, this, _1, _2, _3));
Service::OnNextCheckChanged.connect(bind(&ClusterComponent::NextCheckChangedHandler, this, _1, _2, _3));
endpoint->SetClient(tlsStream);
}
-void ClusterComponent::ReconnectTimerHandler(void)
+void ClusterComponent::ClusterTimerHandler(void)
{
+ /* broadcast a heartbeat message */
+ Dictionary::Ptr message = boost::make_shared<Dictionary>();
+ message->Set("jsonrpc", "2.0");
+ message->Set("method", "cluster::HeartBeat");
+
+ BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
+ endpoint->SendMessage(message);
+ }
+
Array::Ptr peers = GetPeers();
if (!peers)
endpoint->SendMessage(message);
}
+ if (message->Get("method") == "cluster::HeartBeat") {
+ sender->SetSeen(Utility::GetTime());
+ return;
+ }
+
Dictionary::Ptr params = message->Get("params");
if (!params)
return m_Port;
}
+double Endpoint::GetSeen(void) const
+{
+ return m_Seen;
+}
+
+void Endpoint::SetSeen(double ts)
+{
+ m_Seen = ts;
+}
+
void Endpoint::InternalSerialize(const Dictionary::Ptr& bag, int attributeTypes) const
{
DynamicObject::InternalSerialize(bag, attributeTypes);
bag->Set("host", m_Host);
bag->Set("port", m_Port);
}
+
+ if (attributeTypes & Attribute_State)
+ bag->Set("seen", m_Seen);
}
void Endpoint::InternalDeserialize(const Dictionary::Ptr& bag, int attributeTypes)
m_Host = bag->Get("host");
m_Port = bag->Get("port");
}
+
+ if (attributeTypes & Attribute_State)
+ m_Seen = bag->Get("seen");
}
String GetHost(void) const;
String GetPort(void) const;
+ double GetSeen(void) const;
+ void SetSeen(double ts);
+
protected:
virtual void InternalSerialize(const Dictionary::Ptr& bag, int attributeTypes) const;
virtual void InternalDeserialize(const Dictionary::Ptr& bag, int attributeTypes);
String m_Port;
Stream::Ptr m_Client;
+ double m_Seen;
void MessageThreadProc(const Stream::Ptr& stream);
};