mirror of https://github.com/Icinga/icinga2.git
Refactored subscription code in preparation for a centralized broker/discovery component.
This commit is contained in:
Gunnar Beutner
parent
0c59c675be
commit
c8a5343fe5
|
|
@ -113,3 +113,45 @@ DictionaryIterator Dictionary::End(void)
|
|||
{
|
||||
return m_Data.end();
|
||||
}
|
||||
|
||||
long Dictionary::GetLength(void) const
|
||||
{
|
||||
return m_Data.size();
|
||||
}
|
||||
|
||||
void Dictionary::AddUnnamedProperty(const Variant& value)
|
||||
{
|
||||
map<string, Variant>::const_iterator it;
|
||||
string key;
|
||||
do {
|
||||
long index = GetLength();
|
||||
|
||||
stringstream s;
|
||||
s << "_" << GetLength();
|
||||
|
||||
key = s.str();
|
||||
it = m_Data.find(key);
|
||||
} while (it != m_Data.end());
|
||||
|
||||
m_Data[key] = value;
|
||||
}
|
||||
|
||||
void Dictionary::AddUnnamedPropertyString(const string& value)
|
||||
{
|
||||
AddUnnamedProperty(Variant(value));
|
||||
}
|
||||
|
||||
void Dictionary::AddUnnamedPropertyInteger(long value)
|
||||
{
|
||||
AddUnnamedProperty(Variant(value));
|
||||
}
|
||||
|
||||
void Dictionary::AddUnnamedPropertyDictionary(const Dictionary::Ptr& value)
|
||||
{
|
||||
AddUnnamedProperty(Variant(value));
|
||||
}
|
||||
|
||||
void Dictionary::AddUnnamedPropertyObject(const Object::Ptr& value)
|
||||
{
|
||||
AddUnnamedProperty(Variant(value));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -41,6 +41,14 @@ public:
|
|||
DictionaryIterator Begin(void);
|
||||
DictionaryIterator End(void);
|
||||
|
||||
void AddUnnamedProperty(const Variant& value);
|
||||
void AddUnnamedPropertyString(const string& value);
|
||||
void AddUnnamedPropertyInteger(long value);
|
||||
void AddUnnamedPropertyDictionary(const Dictionary::Ptr& value);
|
||||
void AddUnnamedPropertyObject(const Object::Ptr& value);
|
||||
|
||||
long GetLength(void) const;
|
||||
|
||||
Event<PropertyChangedEventArgs> OnPropertyChanged;
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -3,4 +3,5 @@
|
|||
|
||||
SUBDIRS = configfile \
|
||||
configrpc \
|
||||
demo
|
||||
demo \
|
||||
discovery
|
||||
|
|
|
|||
|
|
@ -17,9 +17,6 @@ void DemoComponent::Start(void)
|
|||
EndpointManager::Ptr endpointManager = GetIcingaApplication()->GetEndpointManager();
|
||||
endpointManager->RegisterEndpoint(m_DemoEndpoint);
|
||||
|
||||
endpointManager->OnNewEndpoint += bind_weak(&DemoComponent::NewEndpointHandler, shared_from_this());
|
||||
endpointManager->ForeachEndpoint(bind(&DemoComponent::NewEndpointHandler, this, _1));
|
||||
|
||||
m_DemoTimer = make_shared<Timer>();
|
||||
m_DemoTimer->SetInterval(5);
|
||||
m_DemoTimer->OnTimerExpired += bind_weak(&DemoComponent::DemoTimerHandler, shared_from_this());
|
||||
|
|
@ -36,15 +33,6 @@ void DemoComponent::Stop(void)
|
|||
}
|
||||
}
|
||||
|
||||
int DemoComponent::NewEndpointHandler(const NewEndpointEventArgs& neea)
|
||||
{
|
||||
/* Allow sending/receiving demo messages without authentication */
|
||||
neea.Endpoint->AddAllowedMethodSinkPrefix("demo::");
|
||||
neea.Endpoint->AddAllowedMethodSourcePrefix("demo::");
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DemoComponent::DemoTimerHandler(const TimerEventArgs& tea)
|
||||
{
|
||||
Application::Log("Sending multicast 'hello world' message.");
|
||||
|
|
|
|||
|
|
@ -11,7 +11,6 @@ private:
|
|||
VirtualEndpoint::Ptr m_DemoEndpoint;
|
||||
|
||||
int DemoTimerHandler(const TimerEventArgs& tea);
|
||||
int NewEndpointHandler(const NewEndpointEventArgs& neea);
|
||||
int HelloWorldRequestHandler(const NewRequestEventArgs& nrea);
|
||||
|
||||
public:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,22 @@
|
|||
## Process this file with automake to produce Makefile.in
|
||||
|
||||
pkglib_LTLIBRARIES = \
|
||||
discovery.la
|
||||
|
||||
discovery_la_SOURCES = \
|
||||
discoverycomponent.cpp \
|
||||
discoverycomponent.h \
|
||||
discoverymessage.cpp \
|
||||
discoverymessage.h \
|
||||
i2-discovery.h
|
||||
|
||||
discovery_la_CXXFLAGS = -I${top_srcdir}/base \
|
||||
-I${top_srcdir}/jsonrpc \
|
||||
-I${top_srcdir}/cJSON \
|
||||
-I${top_srcdir}/icinga
|
||||
|
||||
discovery_la_LDFLAGS = -module -version-info 0:0:0 -no-undefined -pthread
|
||||
discovery_la_LIBADD = ${top_builddir}/base/libbase.la \
|
||||
${top_builddir}/jsonrpc/libjsonrpc.la \
|
||||
${top_builddir}/cJSON/libcJSON.la \
|
||||
${top_builddir}/icinga/libicinga.la
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
|
||||
<ItemGroup Label="ProjectConfigurations">
|
||||
<ProjectConfiguration Include="Debug|Win32">
|
||||
<Configuration>Debug</Configuration>
|
||||
<Platform>Win32</Platform>
|
||||
</ProjectConfiguration>
|
||||
<ProjectConfiguration Include="Release|Win32">
|
||||
<Configuration>Release</Configuration>
|
||||
<Platform>Win32</Platform>
|
||||
</ProjectConfiguration>
|
||||
</ItemGroup>
|
||||
<PropertyGroup Label="Globals">
|
||||
<ProjectGuid>{EAD41628-BB96-4F99-9070-8A9676801295}</ProjectGuid>
|
||||
<Keyword>Win32Proj</Keyword>
|
||||
<RootNamespace>discovery</RootNamespace>
|
||||
</PropertyGroup>
|
||||
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.Default.props" />
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'" Label="Configuration">
|
||||
<ConfigurationType>DynamicLibrary</ConfigurationType>
|
||||
<UseDebugLibraries>true</UseDebugLibraries>
|
||||
<CharacterSet>Unicode</CharacterSet>
|
||||
</PropertyGroup>
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'" Label="Configuration">
|
||||
<ConfigurationType>DynamicLibrary</ConfigurationType>
|
||||
<UseDebugLibraries>false</UseDebugLibraries>
|
||||
<WholeProgramOptimization>true</WholeProgramOptimization>
|
||||
<CharacterSet>Unicode</CharacterSet>
|
||||
</PropertyGroup>
|
||||
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.props" />
|
||||
<ImportGroup Label="ExtensionSettings">
|
||||
</ImportGroup>
|
||||
<ImportGroup Label="PropertySheets" Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
|
||||
<Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
|
||||
</ImportGroup>
|
||||
<ImportGroup Label="PropertySheets" Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
|
||||
<Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
|
||||
</ImportGroup>
|
||||
<PropertyGroup Label="UserMacros" />
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
|
||||
<LinkIncremental>true</LinkIncremental>
|
||||
<IncludePath>$(SolutionDir)\base;$(SolutionDir)\jsonrpc;$(SolutionDir)\icinga;$(IncludePath)</IncludePath>
|
||||
<LibraryPath>$(OutDir);$(LibraryPath)</LibraryPath>
|
||||
</PropertyGroup>
|
||||
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
|
||||
<LinkIncremental>false</LinkIncremental>
|
||||
<IncludePath>$(SolutionDir)\base;$(SolutionDir)\jsonrpc;$(SolutionDir)\icinga;$(IncludePath)</IncludePath>
|
||||
<LibraryPath>$(OutDir);$(LibraryPath)</LibraryPath>
|
||||
</PropertyGroup>
|
||||
<ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
|
||||
<ClCompile>
|
||||
<PrecompiledHeader>
|
||||
</PrecompiledHeader>
|
||||
<WarningLevel>Level3</WarningLevel>
|
||||
<Optimization>Disabled</Optimization>
|
||||
<PreprocessorDefinitions>WIN32;_DEBUG;_WINDOWS;_USRDLL;DISCOVERY_EXPORTS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
|
||||
</ClCompile>
|
||||
<Link>
|
||||
<SubSystem>Windows</SubSystem>
|
||||
<GenerateDebugInformation>true</GenerateDebugInformation>
|
||||
<AdditionalDependencies>base.lib;jsonrpc.lib;icinga.lib;cJSON.lib;%(AdditionalDependencies)</AdditionalDependencies>
|
||||
</Link>
|
||||
</ItemDefinitionGroup>
|
||||
<ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
|
||||
<ClCompile>
|
||||
<WarningLevel>Level3</WarningLevel>
|
||||
<PrecompiledHeader>
|
||||
</PrecompiledHeader>
|
||||
<Optimization>MaxSpeed</Optimization>
|
||||
<FunctionLevelLinking>true</FunctionLevelLinking>
|
||||
<IntrinsicFunctions>true</IntrinsicFunctions>
|
||||
<PreprocessorDefinitions>WIN32;NDEBUG;_WINDOWS;_USRDLL;DISCOVERY_EXPORTS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
|
||||
</ClCompile>
|
||||
<Link>
|
||||
<SubSystem>Windows</SubSystem>
|
||||
<GenerateDebugInformation>true</GenerateDebugInformation>
|
||||
<EnableCOMDATFolding>true</EnableCOMDATFolding>
|
||||
<OptimizeReferences>true</OptimizeReferences>
|
||||
<AdditionalDependencies>base.lib;jsonrpc.lib;icinga.lib;cJSON.lib;%(AdditionalDependencies)</AdditionalDependencies>
|
||||
</Link>
|
||||
</ItemDefinitionGroup>
|
||||
<ItemGroup>
|
||||
<ClCompile Include="discoverycomponent.cpp" />
|
||||
<ClCompile Include="discoverymessage.cpp" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ClInclude Include="discoverycomponent.h" />
|
||||
<ClInclude Include="discoverymessage.h" />
|
||||
<ClInclude Include="i2-discovery.h" />
|
||||
</ItemGroup>
|
||||
<Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
|
||||
<ImportGroup Label="ExtensionTargets">
|
||||
</ImportGroup>
|
||||
</Project>
|
||||
|
|
@ -0,0 +1,287 @@
|
|||
#include "i2-discovery.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
||||
string DiscoveryComponent::GetName(void) const
|
||||
{
|
||||
return "discoverycomponent";
|
||||
}
|
||||
|
||||
void DiscoveryComponent::Start(void)
|
||||
{
|
||||
m_DiscoveryEndpoint = make_shared<VirtualEndpoint>();
|
||||
|
||||
long isBroker = 0;
|
||||
GetConfig()->GetPropertyInteger("broker", &isBroker);
|
||||
m_Broker = (isBroker != 0);
|
||||
|
||||
if (IsBroker()) {
|
||||
m_DiscoveryEndpoint->RegisterMethodSource("discovery::NewComponent");
|
||||
m_DiscoveryEndpoint->RegisterMethodHandler("discovery::RegisterComponent",
|
||||
bind_weak(&DiscoveryComponent::RegisterComponentMessageHandler, shared_from_this()));
|
||||
}
|
||||
|
||||
m_DiscoveryEndpoint->RegisterMethodSource("discovery::RegisterComponent");
|
||||
m_DiscoveryEndpoint->RegisterMethodHandler("discovery::NewComponent",
|
||||
bind_weak(&DiscoveryComponent::NewComponentMessageHandler, shared_from_this()));
|
||||
|
||||
GetEndpointManager()->ForeachEndpoint(bind(&DiscoveryComponent::NewEndpointHandler, this, _1));
|
||||
GetEndpointManager()->OnNewEndpoint += bind_weak(&DiscoveryComponent::NewEndpointHandler, shared_from_this());
|
||||
|
||||
GetEndpointManager()->RegisterEndpoint(m_DiscoveryEndpoint);
|
||||
}
|
||||
|
||||
void DiscoveryComponent::Stop(void)
|
||||
{
|
||||
EndpointManager::Ptr mgr = GetEndpointManager();
|
||||
|
||||
if (mgr)
|
||||
mgr->UnregisterEndpoint(m_DiscoveryEndpoint);
|
||||
}
|
||||
|
||||
int DiscoveryComponent::CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea)
|
||||
{
|
||||
if (endpoint == neea.Endpoint)
|
||||
return 0;
|
||||
|
||||
if (!neea.Endpoint->IsConnected())
|
||||
return 0;
|
||||
|
||||
if (endpoint->GetIdentity() == neea.Endpoint->GetIdentity()) {
|
||||
Application::Log("Detected duplicate identity (" + endpoint->GetIdentity() + " - Disconnecting old endpoint.");
|
||||
|
||||
neea.Endpoint->Stop();
|
||||
GetEndpointManager()->UnregisterEndpoint(neea.Endpoint);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::NewEndpointHandler(const NewEndpointEventArgs& neea)
|
||||
{
|
||||
neea.Endpoint->OnIdentityChanged += bind_weak(&DiscoveryComponent::NewIdentityHandler, shared_from_this());
|
||||
|
||||
/* accept discovery::RegisterComponent messages from any endpoint */
|
||||
neea.Endpoint->RegisterMethodSource("discovery::RegisterComponent");
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::DiscoverySinkHandler(const NewMethodEventArgs& nmea, ComponentDiscoveryInfo::Ptr info) const
|
||||
{
|
||||
info->SubscribedMethods.insert(nmea.Method);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::DiscoverySourceHandler(const NewMethodEventArgs& nmea, ComponentDiscoveryInfo::Ptr info) const
|
||||
{
|
||||
info->PublishedMethods.insert(nmea.Method);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::DiscoveryEndpointHandler(const NewEndpointEventArgs& neea, ComponentDiscoveryInfo::Ptr info) const
|
||||
{
|
||||
neea.Endpoint->ForeachMethodSink(bind(&DiscoveryComponent::DiscoverySinkHandler, this, _1, info));
|
||||
neea.Endpoint->ForeachMethodSource(bind(&DiscoveryComponent::DiscoverySourceHandler, this, _1, info));
|
||||
return 0;
|
||||
}
|
||||
|
||||
bool DiscoveryComponent::GetComponentDiscoveryInfo(string component, ComponentDiscoveryInfo::Ptr *info) const
|
||||
{
|
||||
if (component == GetEndpointManager()->GetIdentity()) {
|
||||
/* Build fake discovery info for ourselves */
|
||||
*info = make_shared<ComponentDiscoveryInfo>();
|
||||
GetEndpointManager()->ForeachEndpoint(bind(&DiscoveryComponent::DiscoveryEndpointHandler, this, _1, *info));
|
||||
|
||||
(*info)->Node = GetIcingaApplication()->GetNode();
|
||||
(*info)->Service = GetIcingaApplication()->GetService();
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
map<string, ComponentDiscoveryInfo::Ptr>::const_iterator i;
|
||||
|
||||
i = m_Components.find(component);
|
||||
|
||||
if (i == m_Components.end())
|
||||
return false;
|
||||
|
||||
*info = i->second;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool DiscoveryComponent::IsBroker(void) const
|
||||
{
|
||||
return m_Broker;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::NewIdentityHandler(const EventArgs& ea)
|
||||
{
|
||||
Endpoint::Ptr endpoint = static_pointer_cast<Endpoint>(ea.Source);
|
||||
string identity = endpoint->GetIdentity();
|
||||
|
||||
if (identity == GetEndpointManager()->GetIdentity()) {
|
||||
Application::Log("Detected loop-back connection - Disconnecting endpoint.");
|
||||
|
||||
endpoint->Stop();
|
||||
GetEndpointManager()->UnregisterEndpoint(endpoint);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
GetEndpointManager()->ForeachEndpoint(bind(&DiscoveryComponent::CheckExistingEndpoint, this, endpoint, _1));
|
||||
|
||||
// we assume the other component _always_ wants
|
||||
// discovery::RegisterComponent messages from us
|
||||
endpoint->RegisterMethodSink("discovery::RegisterComponent");
|
||||
|
||||
// send a discovery::RegisterComponent message, if the
|
||||
// other component is a broker this makes sure
|
||||
// the broker knows about our message types
|
||||
SendDiscoveryMessage("discovery::RegisterComponent", GetEndpointManager()->GetIdentity(), endpoint);
|
||||
|
||||
map<string, ComponentDiscoveryInfo::Ptr>::iterator i;
|
||||
|
||||
if (IsBroker()) {
|
||||
// we assume the other component _always_ wants
|
||||
// discovery::NewComponent messages from us
|
||||
endpoint->RegisterMethodSink("discovery::NewComponent");
|
||||
|
||||
// send discovery::NewComponent message for ourselves
|
||||
SendDiscoveryMessage("discovery::NewComponent", GetEndpointManager()->GetIdentity(), endpoint);
|
||||
|
||||
// send discovery::NewComponent messages for all components
|
||||
// we know about
|
||||
for (i = m_Components.begin(); i != m_Components.end(); i++) {
|
||||
SendDiscoveryMessage("discovery::NewComponent", i->first, endpoint);
|
||||
}
|
||||
}
|
||||
|
||||
// check if we already know the other component
|
||||
i = m_Components.find(endpoint->GetIdentity());
|
||||
|
||||
if (i == m_Components.end()) {
|
||||
// we don't know the other component yet, so
|
||||
// wait until we get a discovery::NewComponent message
|
||||
// from a broker
|
||||
return 0;
|
||||
}
|
||||
|
||||
// TODO: send discovery::Welcome message
|
||||
// TODO: add subscriptions/provides to this endpoint
|
||||
return 0;
|
||||
}
|
||||
|
||||
void DiscoveryComponent::SendDiscoveryMessage(string method, string identity, Endpoint::Ptr recipient)
|
||||
{
|
||||
JsonRpcRequest request;
|
||||
request.SetMethod(method);
|
||||
|
||||
DiscoveryMessage params;
|
||||
request.SetParams(params);
|
||||
|
||||
params.SetIdentity(identity);
|
||||
|
||||
Message subscriptions;
|
||||
params.SetSubscribes(subscriptions);
|
||||
|
||||
Message publications;
|
||||
params.SetProvides(publications);
|
||||
|
||||
ComponentDiscoveryInfo::Ptr info;
|
||||
|
||||
if (!GetComponentDiscoveryInfo(identity, &info))
|
||||
return;
|
||||
|
||||
if (!info->Node.empty() && !info->Service.empty()) {
|
||||
params.SetPropertyString("node", info->Node);
|
||||
params.SetPropertyString("service", info->Service);
|
||||
}
|
||||
|
||||
set<string>::iterator i;
|
||||
for (i = info->PublishedMethods.begin(); i != info->PublishedMethods.end(); i++)
|
||||
publications.AddUnnamedPropertyString(*i);
|
||||
|
||||
for (i = info->SubscribedMethods.begin(); i != info->SubscribedMethods.end(); i++)
|
||||
subscriptions.AddUnnamedPropertyString(*i);
|
||||
|
||||
if (recipient)
|
||||
GetEndpointManager()->SendUnicastRequest(m_DiscoveryEndpoint, recipient, request);
|
||||
else
|
||||
GetEndpointManager()->SendMulticastRequest(m_DiscoveryEndpoint, request);
|
||||
}
|
||||
|
||||
void DiscoveryComponent::ProcessDiscoveryMessage(string identity, DiscoveryMessage message)
|
||||
{
|
||||
ComponentDiscoveryInfo::Ptr info = make_shared<ComponentDiscoveryInfo>();
|
||||
}
|
||||
|
||||
int DiscoveryComponent::NewComponentMessageHandler(const NewRequestEventArgs& nrea)
|
||||
{
|
||||
/*Message message;
|
||||
nrea.Request.GetParams(&message);
|
||||
ProcessDiscoveryMessage(message.GetPropertyString(, DiscoveryMessage(message));*/
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::RegisterComponentMessageHandler(const NewRequestEventArgs& nrea)
|
||||
{
|
||||
Message message;
|
||||
nrea.Request.GetParams(&message);
|
||||
ProcessDiscoveryMessage(nrea.Sender->GetIdentity(), DiscoveryMessage(message));
|
||||
return 0;
|
||||
}
|
||||
|
||||
void DiscoveryComponent::AddSubscribedMethod(string identity, string method)
|
||||
{
|
||||
ComponentDiscoveryInfo::Ptr info;
|
||||
|
||||
if (!GetComponentDiscoveryInfo(identity, &info))
|
||||
return;
|
||||
|
||||
info->SubscribedMethods.insert(method);
|
||||
}
|
||||
|
||||
bool DiscoveryComponent::IsSubscribedMethod(string identity, string method) const
|
||||
{
|
||||
if (GetEndpointManager()->GetIdentity() == identity)
|
||||
return true;
|
||||
|
||||
ComponentDiscoveryInfo::Ptr info;
|
||||
|
||||
if (!GetComponentDiscoveryInfo(identity, &info))
|
||||
return false;
|
||||
|
||||
set<string>::const_iterator i;
|
||||
i = info->SubscribedMethods.find(method);
|
||||
|
||||
return (i != info->SubscribedMethods.end());
|
||||
}
|
||||
|
||||
void DiscoveryComponent::AddPublishedMethod(string identity, string method)
|
||||
{
|
||||
ComponentDiscoveryInfo::Ptr info;
|
||||
|
||||
if (!GetComponentDiscoveryInfo(identity, &info))
|
||||
return;
|
||||
|
||||
info->PublishedMethods.insert(method);
|
||||
}
|
||||
|
||||
bool DiscoveryComponent::IsPublishedMethod(string identity, string method) const
|
||||
{
|
||||
if (GetEndpointManager()->GetIdentity() == identity)
|
||||
return true;
|
||||
|
||||
ComponentDiscoveryInfo::Ptr info;
|
||||
|
||||
if (!GetComponentDiscoveryInfo(identity, &info))
|
||||
return false;
|
||||
|
||||
set<string>::const_iterator i;
|
||||
i = info->PublishedMethods.find(method);
|
||||
|
||||
return (i != info->PublishedMethods.end());
|
||||
}
|
||||
|
||||
EXPORT_COMPONENT(DiscoveryComponent);
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
#ifndef DISCOVERYCOMPONENT_H
|
||||
#define DISCOVERYCOMPONENT_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class ComponentDiscoveryInfo : public Object
|
||||
{
|
||||
public:
|
||||
typedef shared_ptr<ComponentDiscoveryInfo> Ptr;
|
||||
typedef weak_ptr<ComponentDiscoveryInfo> WeakPtr;
|
||||
|
||||
string Node;
|
||||
string Service;
|
||||
|
||||
set<string> SubscribedMethods;
|
||||
set<string> PublishedMethods;
|
||||
};
|
||||
|
||||
class DiscoveryComponent : public IcingaComponent
|
||||
{
|
||||
private:
|
||||
VirtualEndpoint::Ptr m_DiscoveryEndpoint;
|
||||
map<string, ComponentDiscoveryInfo::Ptr> m_Components;
|
||||
|
||||
bool m_Broker;
|
||||
|
||||
int NewEndpointHandler(const NewEndpointEventArgs& neea);
|
||||
int NewIdentityHandler(const EventArgs& ea);
|
||||
|
||||
int NewComponentMessageHandler(const NewRequestEventArgs& nrea);
|
||||
int RegisterComponentMessageHandler(const NewRequestEventArgs& nrea);
|
||||
|
||||
void SendDiscoveryMessage(string method, string identity, Endpoint::Ptr recipient);
|
||||
void ProcessDiscoveryMessage(string identity, DiscoveryMessage message);
|
||||
|
||||
bool GetComponentDiscoveryInfo(string component, ComponentDiscoveryInfo::Ptr *info) const;
|
||||
|
||||
int CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea);
|
||||
int DiscoveryEndpointHandler(const NewEndpointEventArgs& neea, ComponentDiscoveryInfo::Ptr info) const;
|
||||
int DiscoverySinkHandler(const NewMethodEventArgs& nmea, ComponentDiscoveryInfo::Ptr info) const;
|
||||
int DiscoverySourceHandler(const NewMethodEventArgs& nmea, ComponentDiscoveryInfo::Ptr info) const;
|
||||
|
||||
bool IsBroker(void) const;
|
||||
|
||||
public:
|
||||
virtual string GetName(void) const;
|
||||
virtual void Start(void);
|
||||
virtual void Stop(void);
|
||||
|
||||
void AddSubscribedMethod(string identity, string method);
|
||||
bool IsSubscribedMethod(string identity, string method) const;
|
||||
|
||||
void AddPublishedMethod(string identity, string method);
|
||||
bool IsPublishedMethod(string identity, string method) const;
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* DISCOVERYCOMPONENT_H */
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
#include "i2-discovery.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
#ifndef DISCOVERYMESSAGE_H
|
||||
#define DISCOVERYMESSAGE_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class DiscoveryMessage : public Message
|
||||
{
|
||||
|
||||
public:
|
||||
DiscoveryMessage(void) : Message() { }
|
||||
DiscoveryMessage(const Message& message) : Message(message) { }
|
||||
|
||||
inline bool GetIdentity(string *value) const
|
||||
{
|
||||
return GetPropertyString("identity", value);
|
||||
}
|
||||
|
||||
inline void SetIdentity(const string& value)
|
||||
{
|
||||
SetPropertyString("identity", value);
|
||||
}
|
||||
|
||||
inline bool GetSubscribes(Message *value) const
|
||||
{
|
||||
return GetPropertyMessage("subscribes", value);
|
||||
}
|
||||
|
||||
inline void SetSubscribes(Message value)
|
||||
{
|
||||
SetPropertyMessage("subscribes", value);
|
||||
}
|
||||
|
||||
inline bool GetProvides(Message *value) const
|
||||
{
|
||||
return GetPropertyMessage("provides", value);
|
||||
}
|
||||
|
||||
inline void SetProvides(Message value)
|
||||
{
|
||||
SetPropertyMessage("provides", value);
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* SUBSCRIPTIONMESSAGE_H */
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
#ifndef I2DISCOVERY_H
|
||||
#define I2DISCOVERY_H
|
||||
|
||||
#include <i2-base.h>
|
||||
#include <i2-jsonrpc.h>
|
||||
#include <i2-icinga.h>
|
||||
|
||||
#include "discoverymessage.h"
|
||||
#include "discoverycomponent.h"
|
||||
|
||||
#endif /* I2DISCOVERY_H */
|
||||
|
|
@ -52,6 +52,7 @@ components/Makefile
|
|||
components/configfile/Makefile
|
||||
components/configrpc/Makefile
|
||||
components/demo/Makefile
|
||||
components/discovery/Makefile
|
||||
icinga/Makefile
|
||||
icinga-app/Makefile
|
||||
jsonrpc/Makefile
|
||||
|
|
|
|||
|
|
@ -3,16 +3,22 @@
|
|||
"icinga": {
|
||||
"privkey": "icinga-c1.key",
|
||||
"pubkey": "icinga-c1.crt",
|
||||
"cakey": "ca.crt"
|
||||
"cakey": "ca.crt",
|
||||
"node": "10.0.10.3",
|
||||
"service": "7777"
|
||||
}
|
||||
},
|
||||
"component": {
|
||||
"configrpc": { "replicate": "0", "configSource": "1" },
|
||||
"demo": { "replicate": "0" }
|
||||
"demo": { "replicate": "0" },
|
||||
"discovery": { "replicate": "0", "broker": "1" }
|
||||
},
|
||||
"rpclistener": {
|
||||
"kekslistener": { "replicate": "0", "port": "7777" }
|
||||
},
|
||||
"rpcconnection": {
|
||||
"foo": { "replicate": "0", "hostname": "10.0.10.3", "port": "7777" }
|
||||
},
|
||||
"host": {
|
||||
"localhost": { "ipaddr": "127.0.0.1" }
|
||||
}
|
||||
|
|
|
|||
10
icinga.sln
10
icinga.sln
|
|
@ -19,6 +19,7 @@ Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "icinga", "icinga\icinga.vcx
|
|||
EndProject
|
||||
Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "icinga-app", "icinga-app\icinga-app.vcxproj", "{BE412865-FEBA-4259-AD41-58950D1F5432}"
|
||||
ProjectSection(ProjectDependencies) = postProject
|
||||
{EAD41628-BB96-4F99-9070-8A9676801295} = {EAD41628-BB96-4F99-9070-8A9676801295}
|
||||
{2E6C1133-730F-4875-A72C-B455B1DD4C5C} = {2E6C1133-730F-4875-A72C-B455B1DD4C5C}
|
||||
{697C6D7E-3109-484C-A7AF-384D28711610} = {697C6D7E-3109-484C-A7AF-384D28711610}
|
||||
{E58F1DA7-B723-412B-B2B7-7FF58E2A944E} = {E58F1DA7-B723-412B-B2B7-7FF58E2A944E}
|
||||
|
|
@ -40,6 +41,11 @@ Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "demo", "components\demo\dem
|
|||
{C1FC77E1-04A4-481B-A78B-2F7AF489C2F8} = {C1FC77E1-04A4-481B-A78B-2F7AF489C2F8}
|
||||
EndProjectSection
|
||||
EndProject
|
||||
Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "discovery", "components\discovery\discovery.vcxproj", "{EAD41628-BB96-4F99-9070-8A9676801295}"
|
||||
ProjectSection(ProjectDependencies) = postProject
|
||||
{C1FC77E1-04A4-481B-A78B-2F7AF489C2F8} = {C1FC77E1-04A4-481B-A78B-2F7AF489C2F8}
|
||||
EndProjectSection
|
||||
EndProject
|
||||
Global
|
||||
GlobalSection(SolutionConfigurationPlatforms) = preSolution
|
||||
Debug|Win32 = Debug|Win32
|
||||
|
|
@ -78,6 +84,10 @@ Global
|
|||
{2E6C1133-730F-4875-A72C-B455B1DD4C5C}.Debug|Win32.Build.0 = Debug|Win32
|
||||
{2E6C1133-730F-4875-A72C-B455B1DD4C5C}.Release|Win32.ActiveCfg = Release|Win32
|
||||
{2E6C1133-730F-4875-A72C-B455B1DD4C5C}.Release|Win32.Build.0 = Release|Win32
|
||||
{EAD41628-BB96-4F99-9070-8A9676801295}.Debug|Win32.ActiveCfg = Debug|Win32
|
||||
{EAD41628-BB96-4F99-9070-8A9676801295}.Debug|Win32.Build.0 = Debug|Win32
|
||||
{EAD41628-BB96-4F99-9070-8A9676801295}.Release|Win32.ActiveCfg = Release|Win32
|
||||
{EAD41628-BB96-4F99-9070-8A9676801295}.Release|Win32.Build.0 = Release|Win32
|
||||
EndGlobalSection
|
||||
GlobalSection(SolutionProperties) = preSolution
|
||||
HideSolutionNode = FALSE
|
||||
|
|
|
|||
|
|
@ -5,8 +5,6 @@ pkglib_LTLIBRARIES = \
|
|||
libicinga.la
|
||||
|
||||
libicinga_la_SOURCES = \
|
||||
discoverycomponent.cpp \
|
||||
discoverycomponent.h \
|
||||
endpoint.cpp \
|
||||
endpoint.h \
|
||||
endpointmanager.cpp \
|
||||
|
|
@ -15,15 +13,9 @@ libicinga_la_SOURCES = \
|
|||
icingaapplication.h \
|
||||
icingacomponent.cpp \
|
||||
icingacomponent.h \
|
||||
identitymessage.cpp \
|
||||
identitymessage.h \
|
||||
i2-icinga.h \
|
||||
jsonrpcendpoint.cpp \
|
||||
jsonrpcendpoint.h \
|
||||
subscriptioncomponent.cpp \
|
||||
subscriptioncomponent.h \
|
||||
subscriptionmessage.cpp \
|
||||
subscriptionmessage.h \
|
||||
virtualendpoint.cpp \
|
||||
virtualendpoint.h
|
||||
|
||||
|
|
|
|||
|
|
@ -1,77 +0,0 @@
|
|||
#include "i2-icinga.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
||||
string DiscoveryComponent::GetName(void) const
|
||||
{
|
||||
return "discoverycomponent";
|
||||
}
|
||||
|
||||
void DiscoveryComponent::Start(void)
|
||||
{
|
||||
m_DiscoveryEndpoint = make_shared<VirtualEndpoint>();
|
||||
m_DiscoveryEndpoint->RegisterMethodHandler("message::Welcome",
|
||||
bind_weak(&DiscoveryComponent::WelcomeMessageHandler, shared_from_this()));
|
||||
|
||||
m_DiscoveryEndpoint->RegisterMethodSource("discovery::PeerAvailable");
|
||||
m_DiscoveryEndpoint->RegisterMethodHandler("discovery::GetPeers",
|
||||
bind_weak(&DiscoveryComponent::GetPeersMessageHandler, shared_from_this()));
|
||||
|
||||
GetEndpointManager()->RegisterEndpoint(m_DiscoveryEndpoint);
|
||||
}
|
||||
|
||||
void DiscoveryComponent::Stop(void)
|
||||
{
|
||||
EndpointManager::Ptr mgr = GetEndpointManager();
|
||||
|
||||
if (mgr)
|
||||
mgr->UnregisterEndpoint(m_DiscoveryEndpoint);
|
||||
}
|
||||
|
||||
int DiscoveryComponent::CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea)
|
||||
{
|
||||
if (endpoint == neea.Endpoint)
|
||||
return 0;
|
||||
|
||||
if (!neea.Endpoint->IsConnected())
|
||||
return 0;
|
||||
|
||||
if (endpoint->GetIdentity() == neea.Endpoint->GetIdentity()) {
|
||||
Application::Log("Detected duplicate identity (" + endpoint->GetIdentity() + " - Disconnecting old endpoint.");
|
||||
|
||||
neea.Endpoint->Stop();
|
||||
GetEndpointManager()->UnregisterEndpoint(neea.Endpoint);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::WelcomeMessageHandler(const NewRequestEventArgs& neea)
|
||||
{
|
||||
if (neea.Sender->GetIdentity() == GetEndpointManager()->GetIdentity()) {
|
||||
Application::Log("Detected loop-back connection - Disconnecting endpoint.");
|
||||
|
||||
neea.Sender->Stop();
|
||||
GetEndpointManager()->UnregisterEndpoint(neea.Sender);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
GetEndpointManager()->ForeachEndpoint(bind(&DiscoveryComponent::CheckExistingEndpoint, this, neea.Sender, _1));
|
||||
|
||||
JsonRpcRequest request;
|
||||
request.SetMethod("discovery::GetPeers");
|
||||
GetEndpointManager()->SendUnicastRequest(m_DiscoveryEndpoint, neea.Sender, request);
|
||||
|
||||
/* TODO: send information about this client to all other clients */
|
||||
/* TODO: send stored events for this client */
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int DiscoveryComponent::GetPeersMessageHandler(const NewRequestEventArgs& nrea)
|
||||
{
|
||||
/* TODO: send information about all available clients to this client */
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -1,27 +0,0 @@
|
|||
#ifndef DISCOVERYCOMPONENT_H
|
||||
#define DISCOVERYCOMPONENT_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class DiscoveryComponent : public IcingaComponent
|
||||
{
|
||||
private:
|
||||
VirtualEndpoint::Ptr m_DiscoveryEndpoint;
|
||||
|
||||
IcingaApplication::Ptr GetIcingaApplication(void) const;
|
||||
|
||||
int WelcomeMessageHandler(const NewRequestEventArgs& neea);
|
||||
int GetPeersMessageHandler(const NewRequestEventArgs& nrea);
|
||||
|
||||
int CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea);
|
||||
|
||||
public:
|
||||
virtual string GetName(void) const;
|
||||
virtual void Start(void);
|
||||
virtual void Stop(void);
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* DISCOVERYCOMPONENT_H */
|
||||
|
|
@ -68,7 +68,7 @@ void Endpoint::UnregisterMethodSource(string method)
|
|||
|
||||
bool Endpoint::IsMethodSource(string method) const
|
||||
{
|
||||
return (m_MethodSources.find(method) != m_MethodSinks.end());
|
||||
return (m_MethodSources.find(method) != m_MethodSources.end());
|
||||
}
|
||||
|
||||
void Endpoint::ForeachMethodSource(function<int (const NewMethodEventArgs&)> callback)
|
||||
|
|
@ -100,3 +100,23 @@ int Endpoint::CountMethodSources(void) const
|
|||
{
|
||||
return m_MethodSources.size();
|
||||
}
|
||||
|
||||
set<string>::const_iterator Endpoint::BeginSinks(void) const
|
||||
{
|
||||
return m_MethodSinks.begin();
|
||||
}
|
||||
|
||||
set<string>::const_iterator Endpoint::EndSinks(void) const
|
||||
{
|
||||
return m_MethodSinks.end();
|
||||
}
|
||||
|
||||
set<string>::const_iterator Endpoint::BeginSources(void) const
|
||||
{
|
||||
return m_MethodSources.begin();
|
||||
}
|
||||
|
||||
set<string>::const_iterator Endpoint::EndSources(void) const
|
||||
{
|
||||
return m_MethodSources.end();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,13 +40,6 @@ public:
|
|||
void RegisterMethodSource(string method);
|
||||
void UnregisterMethodSource(string method);
|
||||
bool IsMethodSource(string method) const;
|
||||
|
||||
virtual void AddAllowedMethodSinkPrefix(string method) = 0;
|
||||
virtual void RemoveAllowedMethodSinkPrefix(string method) = 0;
|
||||
virtual bool IsAllowedMethodSink(string method) const = 0;
|
||||
virtual void AddAllowedMethodSourcePrefix(string method) = 0;
|
||||
virtual void RemoveAllowedMethodSourcePrefix(string method) = 0;
|
||||
virtual bool IsAllowedMethodSource(string method) const = 0;
|
||||
|
||||
virtual bool IsLocal(void) const = 0;
|
||||
virtual bool IsConnected(void) const = 0;
|
||||
|
|
@ -68,7 +61,14 @@ public:
|
|||
int CountMethodSinks(void) const;
|
||||
int CountMethodSources(void) const;
|
||||
|
||||
set<string>::const_iterator BeginSinks(void) const;
|
||||
set<string>::const_iterator EndSinks(void) const;
|
||||
|
||||
set<string>::const_iterator BeginSources(void) const;
|
||||
set<string>::const_iterator EndSources(void) const;
|
||||
|
||||
Event<EventArgs> OnIdentityChanged;
|
||||
Event<EventArgs> OnSessionEstablished;
|
||||
};
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -79,12 +79,6 @@ void EndpointManager::RegisterEndpoint(Endpoint::Ptr endpoint)
|
|||
endpoint->SetEndpointManager(static_pointer_cast<EndpointManager>(shared_from_this()));
|
||||
m_Endpoints.push_front(endpoint);
|
||||
|
||||
endpoint->OnNewMethodSink += bind_weak(&EndpointManager::NewMethodSinkHandler, shared_from_this());
|
||||
endpoint->ForeachMethodSink(bind(&EndpointManager::NewMethodSinkHandler, this, _1));
|
||||
|
||||
endpoint->OnNewMethodSource += bind_weak(&EndpointManager::NewMethodSourceHandler, shared_from_this());
|
||||
endpoint->ForeachMethodSource(bind(&EndpointManager::NewMethodSourceHandler, this, _1));
|
||||
|
||||
NewEndpointEventArgs neea;
|
||||
neea.Source = shared_from_this();
|
||||
neea.Endpoint = endpoint;
|
||||
|
|
@ -109,7 +103,7 @@ void EndpointManager::SendUnicastRequest(Endpoint::Ptr sender, Endpoint::Ptr rec
|
|||
if (!request.GetMethod(&method))
|
||||
throw InvalidArgumentException("Missing 'method' parameter.");
|
||||
|
||||
if (recipient->IsMethodSink(method) && recipient->IsAllowedMethodSink(method))
|
||||
if (recipient->IsMethodSink(method))
|
||||
recipient->ProcessRequest(sender, request);
|
||||
}
|
||||
|
||||
|
|
@ -136,44 +130,6 @@ void EndpointManager::SendMulticastRequest(Endpoint::Ptr sender, const JsonRpcRe
|
|||
}
|
||||
}
|
||||
|
||||
int EndpointManager::NewMethodSinkHandler(const NewMethodEventArgs& ea)
|
||||
{
|
||||
Endpoint::Ptr sender = static_pointer_cast<Endpoint>(ea.Source);
|
||||
|
||||
if (!sender->IsLocal())
|
||||
return 0;
|
||||
|
||||
JsonRpcRequest request;
|
||||
request.SetMethod("message::Subscribe");
|
||||
|
||||
SubscriptionMessage subscriptionMessage;
|
||||
subscriptionMessage.SetMethod(ea.Method);
|
||||
request.SetParams(subscriptionMessage);
|
||||
|
||||
SendMulticastRequest(sender, request);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int EndpointManager::NewMethodSourceHandler(const NewMethodEventArgs& ea)
|
||||
{
|
||||
Endpoint::Ptr sender = static_pointer_cast<Endpoint>(ea.Source);
|
||||
|
||||
if (!sender->IsLocal())
|
||||
return 0;
|
||||
|
||||
JsonRpcRequest request;
|
||||
request.SetMethod("message::Provide");
|
||||
|
||||
SubscriptionMessage subscriptionMessage;
|
||||
subscriptionMessage.SetMethod(ea.Method);
|
||||
request.SetParams(subscriptionMessage);
|
||||
|
||||
SendMulticastRequest(sender, request);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void EndpointManager::ForeachEndpoint(function<int (const NewEndpointEventArgs&)> callback)
|
||||
{
|
||||
NewEndpointEventArgs neea;
|
||||
|
|
|
|||
|
|
@ -22,9 +22,6 @@ class I2_ICINGA_API EndpointManager : public Object
|
|||
|
||||
int NewClientHandler(const NewClientEventArgs& ncea);
|
||||
|
||||
int NewMethodSinkHandler(const NewMethodEventArgs& ea);
|
||||
int NewMethodSourceHandler(const NewMethodEventArgs& ea);
|
||||
|
||||
public:
|
||||
typedef shared_ptr<EndpointManager> Ptr;
|
||||
typedef weak_ptr<EndpointManager> WeakPtr;
|
||||
|
|
|
|||
|
|
@ -17,9 +17,5 @@
|
|||
#include "endpointmanager.h"
|
||||
#include "icingaapplication.h"
|
||||
#include "icingacomponent.h"
|
||||
#include "subscriptioncomponent.h"
|
||||
#include "subscriptionmessage.h"
|
||||
#include "identitymessage.h"
|
||||
#include "discoverycomponent.h"
|
||||
|
||||
#endif /* I2ICINGA_H */
|
||||
|
|
|
|||
|
|
@ -11,28 +11,20 @@
|
|||
</ProjectConfiguration>
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ClCompile Include="discoverycomponent.cpp" />
|
||||
<ClCompile Include="endpoint.cpp" />
|
||||
<ClCompile Include="endpointmanager.cpp" />
|
||||
<ClCompile Include="icingaapplication.cpp" />
|
||||
<ClCompile Include="icingacomponent.cpp" />
|
||||
<ClCompile Include="identitymessage.cpp" />
|
||||
<ClCompile Include="jsonrpcendpoint.cpp" />
|
||||
<ClCompile Include="subscriptioncomponent.cpp" />
|
||||
<ClCompile Include="subscriptionmessage.cpp" />
|
||||
<ClCompile Include="virtualendpoint.cpp" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ClInclude Include="discoverycomponent.h" />
|
||||
<ClInclude Include="endpoint.h" />
|
||||
<ClInclude Include="endpointmanager.h" />
|
||||
<ClInclude Include="i2-icinga.h" />
|
||||
<ClInclude Include="icingaapplication.h" />
|
||||
<ClInclude Include="icingacomponent.h" />
|
||||
<ClInclude Include="identitymessage.h" />
|
||||
<ClInclude Include="jsonrpcendpoint.h" />
|
||||
<ClInclude Include="subscriptioncomponent.h" />
|
||||
<ClInclude Include="subscriptionmessage.h" />
|
||||
<ClInclude Include="virtualendpoint.h" />
|
||||
</ItemGroup>
|
||||
<PropertyGroup Label="Globals">
|
||||
|
|
|
|||
|
|
@ -79,14 +79,6 @@ int IcingaApplication::Main(const vector<string>& args)
|
|||
connectionCollection->ForEachObject(NewRpcConnectionHandler);
|
||||
connectionCollection->OnObjectRemoved += bind_weak(&IcingaApplication::DeletedRpcConnectionHandler, shared_from_this());
|
||||
|
||||
/* load the subscription component */
|
||||
SubscriptionComponent::Ptr subscriptionComponent = make_shared<SubscriptionComponent>();
|
||||
RegisterComponent(subscriptionComponent);
|
||||
|
||||
/* load the discovery component */
|
||||
DiscoveryComponent::Ptr discoveryComponent = make_shared<DiscoveryComponent>();
|
||||
RegisterComponent(discoveryComponent);
|
||||
|
||||
RunEventLoop();
|
||||
|
||||
return EXIT_SUCCESS;
|
||||
|
|
@ -154,6 +146,14 @@ int IcingaApplication::NewIcingaConfigHandler(const EventArgs& ea)
|
|||
if (object->GetPropertyString("cakey", &cakey))
|
||||
SetCAKeyFile(cakey);
|
||||
|
||||
string node;
|
||||
if (object->GetPropertyString("node", &node))
|
||||
SetNode(node);
|
||||
|
||||
string service;
|
||||
if (object->GetPropertyString("service", &service))
|
||||
SetService(service);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -256,3 +256,23 @@ string IcingaApplication::GetCAKeyFile(void) const
|
|||
{
|
||||
return m_CAKeyFile;
|
||||
}
|
||||
|
||||
void IcingaApplication::SetNode(string node)
|
||||
{
|
||||
m_Node = node;
|
||||
}
|
||||
|
||||
string IcingaApplication::GetNode(void) const
|
||||
{
|
||||
return m_Node;
|
||||
}
|
||||
|
||||
void IcingaApplication::SetService(string service)
|
||||
{
|
||||
m_Service = service;
|
||||
}
|
||||
|
||||
string IcingaApplication::GetService(void) const
|
||||
{
|
||||
return m_Service;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ private:
|
|||
string m_PrivateKeyFile;
|
||||
string m_PublicKeyFile;
|
||||
string m_CAKeyFile;
|
||||
string m_Node;
|
||||
string m_Service;
|
||||
|
||||
int NewComponentHandler(const EventArgs& ea);
|
||||
int DeletedComponentHandler(const EventArgs& ea);
|
||||
|
|
@ -45,6 +47,12 @@ public:
|
|||
|
||||
void SetCAKeyFile(string cakey);
|
||||
string GetCAKeyFile(void) const;
|
||||
|
||||
void SetNode(string node);
|
||||
string GetNode(void) const;
|
||||
|
||||
void SetService(string service);
|
||||
string GetService(void) const;
|
||||
};
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,3 +0,0 @@
|
|||
#include "i2-icinga.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
|
@ -1,27 +0,0 @@
|
|||
#ifndef IDENTITYMESSAGE_H
|
||||
#define IDENTITYMESSAGE_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class I2_ICINGA_API IdentityMessage : public Message
|
||||
{
|
||||
|
||||
public:
|
||||
IdentityMessage(void) : Message() { }
|
||||
IdentityMessage(const Message& message) : Message(message) { }
|
||||
|
||||
inline bool GetIdentity(string *value) const
|
||||
{
|
||||
return GetPropertyString("identity", value);
|
||||
}
|
||||
|
||||
inline void SetIdentity(const string& value)
|
||||
{
|
||||
SetPropertyString("identity", value);
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* IDENTITYMESSAGE_H */
|
||||
|
|
@ -15,48 +15,6 @@ JsonRpcClient::Ptr JsonRpcEndpoint::GetClient(void)
|
|||
return m_Client;
|
||||
}
|
||||
|
||||
void JsonRpcEndpoint::AddAllowedMethodSinkPrefix(string method)
|
||||
{
|
||||
m_AllowedMethodSinkPrefixes.insert(method);
|
||||
}
|
||||
|
||||
void JsonRpcEndpoint::RemoveAllowedMethodSinkPrefix(string method)
|
||||
{
|
||||
m_AllowedMethodSinkPrefixes.erase(method);
|
||||
}
|
||||
|
||||
bool JsonRpcEndpoint::IsAllowedMethodSink(string method) const
|
||||
{
|
||||
set<string>::iterator i;
|
||||
for (i = m_AllowedMethodSinkPrefixes.begin(); i != m_AllowedMethodSinkPrefixes.end(); i++) {
|
||||
if (method.compare(0, method.length(), method) == 0)
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
void JsonRpcEndpoint::AddAllowedMethodSourcePrefix(string method)
|
||||
{
|
||||
m_AllowedMethodSourcePrefixes.insert(method);
|
||||
}
|
||||
|
||||
void JsonRpcEndpoint::RemoveAllowedMethodSourcePrefix(string method)
|
||||
{
|
||||
m_AllowedMethodSourcePrefixes.erase(method);
|
||||
}
|
||||
|
||||
bool JsonRpcEndpoint::IsAllowedMethodSource(string method) const
|
||||
{
|
||||
set<string>::iterator i;
|
||||
for (i = m_AllowedMethodSourcePrefixes.begin(); i != m_AllowedMethodSourcePrefixes.end(); i++) {
|
||||
if (method.compare(0, method.length(), method) == 0)
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
void JsonRpcEndpoint::Connect(string host, unsigned short port, shared_ptr<SSL_CTX> sslContext)
|
||||
{
|
||||
m_PeerHostname = host;
|
||||
|
|
@ -115,7 +73,7 @@ int JsonRpcEndpoint::NewMessageHandler(const NewMessageEventArgs& nmea)
|
|||
|
||||
string method;
|
||||
if (message.GetPropertyString("method", &method)) {
|
||||
if (!IsAllowedMethodSource(method))
|
||||
if (!IsMethodSource(method))
|
||||
return 0;
|
||||
|
||||
JsonRpcRequest request = message;
|
||||
|
|
|
|||
|
|
@ -12,8 +12,6 @@ private:
|
|||
JsonRpcClient::Ptr m_Client;
|
||||
map<string, Endpoint::Ptr> m_PendingCalls;
|
||||
Timer::Ptr m_ReconnectTimer;
|
||||
set<string> m_AllowedMethodSinkPrefixes;
|
||||
set<string> m_AllowedMethodSourcePrefixes;
|
||||
|
||||
string m_PeerHostname;
|
||||
unsigned short m_PeerPort;
|
||||
|
|
@ -37,13 +35,6 @@ public:
|
|||
void SetAddress(string address);
|
||||
virtual string GetAddress(void) const;
|
||||
|
||||
virtual void AddAllowedMethodSinkPrefix(string method);
|
||||
virtual void RemoveAllowedMethodSinkPrefix(string method);
|
||||
virtual bool IsAllowedMethodSink(string method) const;
|
||||
virtual void AddAllowedMethodSourcePrefix(string method);
|
||||
virtual void RemoveAllowedMethodSourcePrefix(string method);
|
||||
virtual bool IsAllowedMethodSource(string method) const;
|
||||
|
||||
virtual bool IsLocal(void) const;
|
||||
virtual bool IsConnected(void) const;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,117 +0,0 @@
|
|||
#include "i2-icinga.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
||||
string SubscriptionComponent::GetName(void) const
|
||||
{
|
||||
return "subscriptioncomponent";
|
||||
}
|
||||
|
||||
void SubscriptionComponent::Start(void)
|
||||
{
|
||||
m_SubscriptionEndpoint = make_shared<VirtualEndpoint>();
|
||||
m_SubscriptionEndpoint->RegisterMethodHandler("message::Subscribe", bind_weak(&SubscriptionComponent::SubscribeMessageHandler, shared_from_this()));
|
||||
m_SubscriptionEndpoint->RegisterMethodHandler("message::Provide", bind_weak(&SubscriptionComponent::ProvideMessageHandler, shared_from_this()));
|
||||
m_SubscriptionEndpoint->RegisterMethodSource("message::Welcome");
|
||||
m_SubscriptionEndpoint->RegisterMethodSource("message::Subscribe");
|
||||
m_SubscriptionEndpoint->RegisterMethodSource("message::Provide");
|
||||
|
||||
EndpointManager::Ptr mgr = GetEndpointManager();
|
||||
mgr->OnNewEndpoint += bind_weak(&SubscriptionComponent::NewEndpointHandler, shared_from_this());
|
||||
mgr->ForeachEndpoint(bind(&SubscriptionComponent::NewEndpointHandler, this, _1));
|
||||
mgr->RegisterEndpoint(m_SubscriptionEndpoint);
|
||||
}
|
||||
|
||||
void SubscriptionComponent::Stop(void)
|
||||
{
|
||||
EndpointManager::Ptr mgr = GetEndpointManager();
|
||||
|
||||
if (mgr)
|
||||
mgr->UnregisterEndpoint(m_SubscriptionEndpoint);
|
||||
}
|
||||
|
||||
int SubscriptionComponent::SyncSubscription(Endpoint::Ptr target, string type, const NewMethodEventArgs& nmea)
|
||||
{
|
||||
JsonRpcRequest request;
|
||||
request.SetVersion("2.0");
|
||||
request.SetMethod(type);
|
||||
|
||||
SubscriptionMessage subscriptionMessage;
|
||||
subscriptionMessage.SetMethod(nmea.Method);
|
||||
request.SetParams(subscriptionMessage);
|
||||
GetEndpointManager()->SendUnicastRequest(m_SubscriptionEndpoint, target, request);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int SubscriptionComponent::SyncSubscriptions(Endpoint::Ptr target, const NewEndpointEventArgs& neea)
|
||||
{
|
||||
Endpoint::Ptr source = neea.Endpoint;
|
||||
|
||||
if (!source->IsLocal())
|
||||
return 0;
|
||||
|
||||
source->ForeachMethodSink(bind(&SubscriptionComponent::SyncSubscription, this, target, "message::Subscribe", _1));
|
||||
source->ForeachMethodSource(bind(&SubscriptionComponent::SyncSubscription, this, target, "message::Provide", _1));
|
||||
|
||||
// TODO: bind to endpoint's events
|
||||
//endpoint->OnNewMethodSink...
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int SubscriptionComponent::NewEndpointHandler(const NewEndpointEventArgs& neea)
|
||||
{
|
||||
if (neea.Endpoint->IsLocal())
|
||||
return 0;
|
||||
|
||||
neea.Endpoint->AddAllowedMethodSinkPrefix("message::");
|
||||
neea.Endpoint->AddAllowedMethodSourcePrefix("message::");
|
||||
|
||||
/* we just assume the peer wants those messages */
|
||||
neea.Endpoint->RegisterMethodSink("message::Welcome");
|
||||
neea.Endpoint->RegisterMethodSink("message::Subscribe");
|
||||
neea.Endpoint->RegisterMethodSink("message::Provide");
|
||||
|
||||
GetEndpointManager()->ForeachEndpoint(bind(&SubscriptionComponent::SyncSubscriptions, this, neea.Endpoint, _1));
|
||||
|
||||
/* signal the peer that we're done syncing subscriptions and are now
|
||||
* ready to accept messages. */
|
||||
JsonRpcRequest request;
|
||||
request.SetMethod("message::Welcome");
|
||||
GetEndpointManager()->SendUnicastRequest(m_SubscriptionEndpoint, neea.Endpoint, request);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int SubscriptionComponent::SubscribeMessageHandler(const NewRequestEventArgs& nrea)
|
||||
{
|
||||
Message params;
|
||||
if (!nrea.Request.GetParams(¶ms))
|
||||
return 0;
|
||||
|
||||
SubscriptionMessage subscriptionMessage = params;
|
||||
|
||||
string method;
|
||||
if (!subscriptionMessage.GetMethod(&method))
|
||||
return 0;
|
||||
|
||||
nrea.Sender->RegisterMethodSink(method);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int SubscriptionComponent::ProvideMessageHandler(const NewRequestEventArgs& nrea)
|
||||
{
|
||||
Message params;
|
||||
if (!nrea.Request.GetParams(¶ms))
|
||||
return 0;
|
||||
|
||||
SubscriptionMessage subscriptionMessage = params;
|
||||
|
||||
string method;
|
||||
if (!subscriptionMessage.GetMethod(&method))
|
||||
return 0;
|
||||
|
||||
nrea.Sender->RegisterMethodSource(method);
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
#ifndef I2_SUBSCRIPTIONCOMPONENT_H
|
||||
#define I2_SUBSCRIPTIONCOMPONENT_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class SubscriptionComponent : public IcingaComponent
|
||||
{
|
||||
private:
|
||||
VirtualEndpoint::Ptr m_SubscriptionEndpoint;
|
||||
|
||||
int NewEndpointHandler(const NewEndpointEventArgs& neea);
|
||||
int SubscribeMessageHandler(const NewRequestEventArgs& nrea);
|
||||
int ProvideMessageHandler(const NewRequestEventArgs& nrea);
|
||||
int IdentityMessageHandler(const NewRequestEventArgs& nrea);
|
||||
|
||||
int SyncSubscription(Endpoint::Ptr target, string type, const NewMethodEventArgs& nmea);
|
||||
int SyncSubscriptions(Endpoint::Ptr target, const NewEndpointEventArgs& neea);
|
||||
|
||||
public:
|
||||
virtual string GetName(void) const;
|
||||
virtual void Start(void);
|
||||
virtual void Stop(void);
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* I2_SUBSCRIPTIONCOMPONENT_H */
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
#include "i2-icinga.h"
|
||||
|
||||
using namespace icinga;
|
||||
|
|
@ -1,27 +0,0 @@
|
|||
#ifndef SUBSCRIPTIONMESSAGE_H
|
||||
#define SUBSCRIPTIONMESSAGE_H
|
||||
|
||||
namespace icinga
|
||||
{
|
||||
|
||||
class I2_ICINGA_API SubscriptionMessage : public Message
|
||||
{
|
||||
|
||||
public:
|
||||
SubscriptionMessage(void) : Message() { }
|
||||
SubscriptionMessage(const Message& message) : Message(message) { }
|
||||
|
||||
inline bool GetMethod(string *value) const
|
||||
{
|
||||
return GetPropertyString("method", value);
|
||||
}
|
||||
|
||||
inline void SetMethod(const string& value)
|
||||
{
|
||||
SetPropertyString("method", value);
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* SUBSCRIPTIONMESSAGE_H */
|
||||
|
|
@ -59,36 +59,6 @@ void VirtualEndpoint::ProcessResponse(Endpoint::Ptr sender, const JsonRpcRespons
|
|||
throw NotImplementedException();
|
||||
}
|
||||
|
||||
void VirtualEndpoint::AddAllowedMethodSinkPrefix(string method)
|
||||
{
|
||||
/* Nothing to do here. */
|
||||
}
|
||||
|
||||
void VirtualEndpoint::RemoveAllowedMethodSinkPrefix(string method)
|
||||
{
|
||||
/* Nothing to do here. */
|
||||
}
|
||||
|
||||
void VirtualEndpoint::AddAllowedMethodSourcePrefix(string method)
|
||||
{
|
||||
/* Nothing to do here. */
|
||||
}
|
||||
|
||||
void VirtualEndpoint::RemoveAllowedMethodSourcePrefix(string method)
|
||||
{
|
||||
/* Nothing to do here. */
|
||||
}
|
||||
|
||||
bool VirtualEndpoint::IsAllowedMethodSink(string method) const
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
bool VirtualEndpoint::IsAllowedMethodSource(string method) const
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
void VirtualEndpoint::Stop(void)
|
||||
{
|
||||
/* Nothing to do here. */
|
||||
|
|
|
|||
|
|
@ -25,13 +25,6 @@ public:
|
|||
void RegisterMethodHandler(string method, function<int (const NewRequestEventArgs&)> callback);
|
||||
void UnregisterMethodHandler(string method, function<int (const NewRequestEventArgs&)> callback);
|
||||
|
||||
virtual void AddAllowedMethodSinkPrefix(string method);
|
||||
virtual void RemoveAllowedMethodSinkPrefix(string method);
|
||||
virtual bool IsAllowedMethodSink(string method) const;
|
||||
virtual void AddAllowedMethodSourcePrefix(string method);
|
||||
virtual void RemoveAllowedMethodSourcePrefix(string method);
|
||||
virtual bool IsAllowedMethodSource(string method) const;
|
||||
|
||||
virtual string GetAddress(void) const;
|
||||
|
||||
virtual bool IsLocal(void) const;
|
||||
|
|
|
|||
|
|
@ -56,3 +56,18 @@ void Message::SetPropertyMessage(string key, const Message& value)
|
|||
{
|
||||
GetDictionary()->SetProperty(key, Variant(value.GetDictionary()));
|
||||
}
|
||||
|
||||
void Message::AddUnnamedPropertyString(const string& value)
|
||||
{
|
||||
GetDictionary()->AddUnnamedPropertyString(value);
|
||||
}
|
||||
|
||||
void Message::AddUnnamedPropertyInteger(long value)
|
||||
{
|
||||
GetDictionary()->AddUnnamedPropertyInteger(value);
|
||||
}
|
||||
|
||||
void Message::AddUnnamedPropertyMessage(const Message& value)
|
||||
{
|
||||
GetDictionary()->AddUnnamedPropertyDictionary(value.GetDictionary());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,6 +24,10 @@ public:
|
|||
|
||||
bool GetPropertyMessage(string key, Message *value) const;
|
||||
void SetPropertyMessage(string key, const Message& value);
|
||||
|
||||
void AddUnnamedPropertyString(const string& value);
|
||||
void AddUnnamedPropertyInteger(long value);
|
||||
void AddUnnamedPropertyMessage(const Message& value);
|
||||
};
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue