mirror of https://github.com/Icinga/icinga2.git
401 lines
11 KiB
C++
401 lines
11 KiB
C++
/******************************************************************************
|
|
* Icinga 2 *
|
|
* Copyright (C) 2012 Icinga Development Team (http://www.icinga.org/) *
|
|
* *
|
|
* This program is free software; you can redistribute it and/or *
|
|
* modify it under the terms of the GNU General Public License *
|
|
* as published by the Free Software Foundation; either version 2 *
|
|
* of the License, or (at your option) any later version. *
|
|
* *
|
|
* This program is distributed in the hope that it will be useful, *
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
|
|
* GNU General Public License for more details. *
|
|
* *
|
|
* You should have received a copy of the GNU General Public License *
|
|
* along with this program; if not, write to the Free Software Foundation *
|
|
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. *
|
|
******************************************************************************/
|
|
|
|
#include "i2-icinga.h"
|
|
|
|
using namespace icinga;
|
|
|
|
/**
|
|
* Constructor for the EndpointManager class.
|
|
*/
|
|
EndpointManager::EndpointManager(void)
|
|
: m_NextMessageID(0)
|
|
{
|
|
m_RequestTimer = boost::make_shared<Timer>();
|
|
m_RequestTimer->OnTimerExpired.connect(boost::bind(&EndpointManager::RequestTimerHandler, this));
|
|
m_RequestTimer->SetInterval(5);
|
|
m_RequestTimer->Start();
|
|
}
|
|
|
|
/**
|
|
* Sets the identity of the endpoint manager. This identity is used when
|
|
* connecting to remote peers.
|
|
*
|
|
* @param identity The new identity.
|
|
*/
|
|
void EndpointManager::SetIdentity(const String& identity)
|
|
{
|
|
m_Identity = identity;
|
|
}
|
|
|
|
/**
|
|
* Retrieves the identity for the endpoint manager.
|
|
*
|
|
* @returns The identity.
|
|
*/
|
|
String EndpointManager::GetIdentity(void) const
|
|
{
|
|
return m_Identity;
|
|
}
|
|
|
|
/**
|
|
* Sets the SSL context that is used for remote connections.
|
|
*
|
|
* @param sslContext The new SSL context.
|
|
*/
|
|
void EndpointManager::SetSSLContext(const shared_ptr<SSL_CTX>& sslContext)
|
|
{
|
|
m_SSLContext = sslContext;
|
|
}
|
|
|
|
/**
|
|
* Retrieves the SSL context that is used for remote connections.
|
|
*
|
|
* @returns The SSL context.
|
|
*/
|
|
shared_ptr<SSL_CTX> EndpointManager::GetSSLContext(void) const
|
|
{
|
|
return m_SSLContext;
|
|
}
|
|
|
|
/**
|
|
* Creates a new JSON-RPC listener on the specified port.
|
|
*
|
|
* @param service The port to listen on.
|
|
*/
|
|
void EndpointManager::AddListener(const String& service)
|
|
{
|
|
if (!GetSSLContext())
|
|
throw_exception(logic_error("SSL context is required for AddListener()"));
|
|
|
|
stringstream s;
|
|
s << "Adding new listener: port " << service;
|
|
Logger::Write(LogInformation, "icinga", s.str());
|
|
|
|
JsonRpcServer::Ptr server = boost::make_shared<JsonRpcServer>(m_SSLContext);
|
|
RegisterServer(server);
|
|
|
|
server->Bind(service, AF_INET6);
|
|
server->Listen();
|
|
server->Start();
|
|
}
|
|
|
|
/**
|
|
* Creates a new JSON-RPC client and connects to the specified host and port.
|
|
*
|
|
* @param node The remote host.
|
|
* @param service The remote port.
|
|
*/
|
|
void EndpointManager::AddConnection(const String& node, const String& service)
|
|
{
|
|
stringstream s;
|
|
s << "Adding new endpoint: [" << node << "]:" << service;
|
|
Logger::Write(LogInformation, "icinga", s.str());
|
|
|
|
JsonRpcEndpoint::Ptr endpoint = boost::make_shared<JsonRpcEndpoint>();
|
|
RegisterEndpoint(endpoint);
|
|
endpoint->Connect(node, service, m_SSLContext);
|
|
}
|
|
|
|
/**
|
|
* Registers a new JSON-RPC server with this endpoint manager.
|
|
*
|
|
* @param server The JSON-RPC server.
|
|
*/
|
|
void EndpointManager::RegisterServer(const JsonRpcServer::Ptr& server)
|
|
{
|
|
m_Servers.push_back(server);
|
|
server->OnNewClient.connect(boost::bind(&EndpointManager::NewClientHandler,
|
|
this, _2));
|
|
}
|
|
|
|
/**
|
|
* Processes a new client connection.
|
|
*
|
|
* @param ncea Event arguments.
|
|
*/
|
|
void EndpointManager::NewClientHandler(const TcpClient::Ptr& client)
|
|
{
|
|
Logger::Write(LogInformation, "icinga", "Accepted new client from " + client->GetPeerAddress());
|
|
|
|
JsonRpcEndpoint::Ptr endpoint = boost::make_shared<JsonRpcEndpoint>();
|
|
endpoint->SetClient(static_pointer_cast<JsonRpcClient>(client));
|
|
client->Start();
|
|
RegisterEndpoint(endpoint);
|
|
}
|
|
|
|
/**
|
|
* Unregisters a JSON-RPC server.
|
|
*
|
|
* @param server The JSON-RPC server.
|
|
*/
|
|
void EndpointManager::UnregisterServer(const JsonRpcServer::Ptr& server)
|
|
{
|
|
m_Servers.erase(
|
|
remove(m_Servers.begin(), m_Servers.end(), server),
|
|
m_Servers.end());
|
|
// TODO: unbind event
|
|
}
|
|
|
|
/**
|
|
* Registers a new endpoint with this endpoint manager.
|
|
*
|
|
* @param endpoint The new endpoint.
|
|
*/
|
|
void EndpointManager::RegisterEndpoint(const Endpoint::Ptr& endpoint)
|
|
{
|
|
endpoint->SetEndpointManager(GetSelf());
|
|
|
|
UnregisterEndpoint(endpoint);
|
|
|
|
String identity = endpoint->GetIdentity();
|
|
|
|
if (!identity.IsEmpty()) {
|
|
m_Endpoints[identity] = endpoint;
|
|
OnNewEndpoint(GetSelf(), endpoint);
|
|
} else {
|
|
m_PendingEndpoints.push_back(endpoint);
|
|
}
|
|
|
|
if (endpoint->IsLocal()) {
|
|
/* this endpoint might have introduced new subscriptions
|
|
* or publications which affect remote endpoints, we need
|
|
* to close all fully-connected remote endpoints to make sure
|
|
* these subscriptions/publications are kept up-to-date. */
|
|
Iterator prev, it;
|
|
for (it = m_Endpoints.begin(); it != m_Endpoints.end(); ) {
|
|
prev = it;
|
|
it++;
|
|
|
|
if (!prev->second->IsLocal())
|
|
m_Endpoints.erase(prev);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Unregisters an endpoint.
|
|
*
|
|
* @param endpoint The endpoint.
|
|
*/
|
|
void EndpointManager::UnregisterEndpoint(const Endpoint::Ptr& endpoint)
|
|
{
|
|
m_PendingEndpoints.erase(
|
|
remove(m_PendingEndpoints.begin(), m_PendingEndpoints.end(), endpoint),
|
|
m_PendingEndpoints.end());
|
|
|
|
String identity = endpoint->GetIdentity();
|
|
if (!identity.IsEmpty())
|
|
m_Endpoints.erase(identity);
|
|
}
|
|
|
|
/**
|
|
* Sends a unicast message to the specified recipient.
|
|
*
|
|
* @param sender The sender of the message.
|
|
* @param recipient The recipient of the message.
|
|
* @param message The request.
|
|
*/
|
|
void EndpointManager::SendUnicastMessage(const Endpoint::Ptr& sender,
|
|
const Endpoint::Ptr& recipient, const MessagePart& message)
|
|
{
|
|
/* don't forward messages between non-local endpoints */
|
|
if (!sender->IsLocal() && !recipient->IsLocal())
|
|
return;
|
|
|
|
if (ResponseMessage::IsResponseMessage(message))
|
|
recipient->ProcessResponse(sender, message);
|
|
else
|
|
recipient->ProcessRequest(sender, message);
|
|
}
|
|
|
|
/**
|
|
* Sends a message to exactly one recipient out of all recipients who have a
|
|
* subscription for the message's topic.
|
|
*
|
|
* @param sender The sender of the message.
|
|
* @param message The message.
|
|
*/
|
|
void EndpointManager::SendAnycastMessage(const Endpoint::Ptr& sender,
|
|
const RequestMessage& message)
|
|
{
|
|
String method;
|
|
if (!message.GetMethod(&method))
|
|
throw_exception(invalid_argument("Message is missing the 'method' property."));
|
|
|
|
vector<Endpoint::Ptr> candidates;
|
|
Endpoint::Ptr endpoint;
|
|
BOOST_FOREACH(tie(tuples::ignore, endpoint), m_Endpoints) {
|
|
/* don't forward messages between non-local endpoints */
|
|
if (!sender->IsLocal() && !endpoint->IsLocal())
|
|
continue;
|
|
|
|
if (endpoint->HasSubscription(method))
|
|
candidates.push_back(endpoint);
|
|
}
|
|
|
|
if (candidates.empty())
|
|
return;
|
|
|
|
Endpoint::Ptr recipient = candidates[rand() % candidates.size()];
|
|
SendUnicastMessage(sender, recipient, message);
|
|
}
|
|
|
|
/**
|
|
* Sends a message to all recipients who have a subscription for the
|
|
* message's topic.
|
|
*
|
|
* @param sender The sender of the message.
|
|
* @param message The message.
|
|
*/
|
|
void EndpointManager::SendMulticastMessage(const Endpoint::Ptr& sender,
|
|
const RequestMessage& message)
|
|
{
|
|
String id;
|
|
if (message.GetID(&id))
|
|
throw_exception(invalid_argument("Multicast requests must not have an ID."));
|
|
|
|
String method;
|
|
if (!message.GetMethod(&method))
|
|
throw_exception(invalid_argument("Message is missing the 'method' property."));
|
|
|
|
Endpoint::Ptr recipient;
|
|
BOOST_FOREACH(tie(tuples::ignore, recipient), m_Endpoints) {
|
|
/* don't forward messages back to the sender */
|
|
if (sender == recipient)
|
|
continue;
|
|
|
|
if (recipient->HasSubscription(method))
|
|
SendUnicastMessage(sender, recipient, message);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Calls the specified callback function for each registered endpoint.
|
|
*
|
|
* @param callback The callback function.
|
|
*/
|
|
void EndpointManager::ForEachEndpoint(function<void (const EndpointManager::Ptr&, const Endpoint::Ptr&)> callback)
|
|
{
|
|
map<String, Endpoint::Ptr>::iterator prev, i;
|
|
for (i = m_Endpoints.begin(); i != m_Endpoints.end(); ) {
|
|
prev = i;
|
|
i++;
|
|
|
|
callback(GetSelf(), prev->second);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Retrieves an endpoint that has the specified identity.
|
|
*
|
|
* @param identity The identity of the endpoint.
|
|
*/
|
|
Endpoint::Ptr EndpointManager::GetEndpointByIdentity(const String& identity) const
|
|
{
|
|
map<String, Endpoint::Ptr>::const_iterator i;
|
|
i = m_Endpoints.find(identity);
|
|
if (i != m_Endpoints.end())
|
|
return i->second;
|
|
else
|
|
return Endpoint::Ptr();
|
|
}
|
|
|
|
void EndpointManager::SendAPIMessage(const Endpoint::Ptr& sender, const Endpoint::Ptr& recipient,
|
|
RequestMessage& message,
|
|
function<void(const EndpointManager::Ptr&, const Endpoint::Ptr, const RequestMessage&, const ResponseMessage&, bool TimedOut)> callback, double timeout)
|
|
{
|
|
m_NextMessageID++;
|
|
|
|
stringstream idstream;
|
|
idstream << m_NextMessageID;
|
|
|
|
String id = idstream.str();
|
|
message.SetID(id);
|
|
|
|
PendingRequest pr;
|
|
pr.Request = message;
|
|
pr.Callback = callback;
|
|
pr.Timeout = Utility::GetTime() + timeout;
|
|
|
|
m_Requests[id] = pr;
|
|
|
|
if (!recipient)
|
|
SendAnycastMessage(sender, message);
|
|
else
|
|
SendUnicastMessage(sender, recipient, message);
|
|
}
|
|
|
|
bool EndpointManager::RequestTimeoutLessComparer(const pair<String, PendingRequest>& a,
|
|
const pair<String, PendingRequest>& b)
|
|
{
|
|
return a.second.Timeout < b.second.Timeout;
|
|
}
|
|
|
|
void EndpointManager::RequestTimerHandler(void)
|
|
{
|
|
map<String, PendingRequest>::iterator it;
|
|
for (it = m_Requests.begin(); it != m_Requests.end(); it++) {
|
|
if (it->second.HasTimedOut()) {
|
|
it->second.Callback(GetSelf(), Endpoint::Ptr(), it->second.Request, ResponseMessage(), true);
|
|
|
|
m_Requests.erase(it);
|
|
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
void EndpointManager::ProcessResponseMessage(const Endpoint::Ptr& sender, const ResponseMessage& message)
|
|
{
|
|
String id;
|
|
if (!message.GetID(&id))
|
|
throw_exception(invalid_argument("Response message must have a message ID."));
|
|
|
|
map<String, PendingRequest>::iterator it;
|
|
it = m_Requests.find(id);
|
|
|
|
if (it == m_Requests.end())
|
|
return;
|
|
|
|
it->second.Callback(GetSelf(), sender, it->second.Request, message, false);
|
|
|
|
m_Requests.erase(it);
|
|
}
|
|
|
|
EndpointManager::Iterator EndpointManager::Begin(void)
|
|
{
|
|
return m_Endpoints.begin();
|
|
}
|
|
|
|
EndpointManager::Iterator EndpointManager::End(void)
|
|
{
|
|
return m_Endpoints.end();
|
|
}
|
|
|
|
EndpointManager::Ptr EndpointManager::GetInstance(void)
|
|
{
|
|
static EndpointManager::Ptr instance;
|
|
|
|
if (!instance)
|
|
instance = boost::make_shared<EndpointManager>();
|
|
|
|
return instance;
|
|
}
|