Simplify the supervisor responsibility

4 files changed, 27 insertions, 15 deletions

diff --git a/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.cpp b/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.cpp
index e00164b5a1a..757f3a976ed 100644
--- a/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.cpp
+++ b/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.cpp
@@ -20,7 +20,9 @@ namespace documentapi {
 
 ExternPolicy::ExternPolicy(const string &param) :
     _lock(),
-    _orb(std::make_unique<FRT_Supervisor>()),
+    _threadPool(std::make_unique<FastOS_ThreadPool>(1024*60)),
+    _transport(std::make_unique<FNET_Transport>()),
+    _orb(std::make_unique<FRT_Supervisor>(_transport.get())),
     _mirror(),
     _pattern(),
     _session(),
@@ -56,7 +58,7 @@ ExternPolicy::ExternPolicy(const string &param) :
 
     slobrok::ConfiguratorFactory config(spec);
     _mirror.reset(new MirrorAPI(*_orb, config));
-    _started = _orb->Start();
+    _started = _transport->Start(_threadPool.get());
     if (!_started) {
         _error = "Failed to start FNET supervisor.";
         return;
@@ -81,7 +83,7 @@ ExternPolicy::~ExternPolicy()
 {
     _mirror.reset();
     if (_started) {
-        _orb->ShutDown(true);
+        _transport->ShutDown(true);
     }
 }
 
@@ -133,10 +135,9 @@ ExternPolicy::update()
 
         IMirrorAPI::SpecList entries = _mirror->lookup(_pattern);
         if (!entries.empty()) {
-            for (IMirrorAPI::SpecList::iterator it = entries.begin();
-                 it != entries.end(); ++it)
+            for (const auto & spec : entries)
             {
-                _recipients.push_back(mbus::Hop::parse(it->second + _session));
+                _recipients.push_back(mbus::Hop::parse(spec.second + _session));
             }
         }
     }
diff --git a/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.h b/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.h
index 61182071a54..c56f8f214ec 100644
--- a/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.h
+++ b/documentapi/src/vespa/documentapi/messagebus/policies/externpolicy.h
@@ -9,6 +9,8 @@
 #include <vespa/documentapi/common.h>
 
 class FRT_Supervisor;
+class FNET_Transport;
+class FastOS_ThreadPool;
 
 namespace documentapi {
 
@@ -19,9 +21,11 @@ namespace documentapi {
 class ExternPolicy : public mbus::IRoutingPolicy {
 private:
     using IMirrorAPI = slobrok::api::IMirrorAPI;
-    vespalib::Lock                   _lock;
-    std::unique_ptr<FRT_Supervisor>  _orb;
-    std::unique_ptr<IMirrorAPI>      _mirror;
+    vespalib::Lock                      _lock;
+    std::unique_ptr<FastOS_ThreadPool>  _threadPool;
+    std::unique_ptr<FNET_Transport>     _transport;
+    std::unique_ptr<FRT_Supervisor>     _orb;
+    std::unique_ptr<IMirrorAPI>         _mirror;
     string                           _pattern;
     string                           _session;
     string                           _error;
diff --git a/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.cpp b/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.cpp
index 5bbfa75012d..18dd525b066 100644
--- a/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.cpp
+++ b/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.cpp
@@ -5,6 +5,8 @@
 #include <vespa/messagebus/routing/routingcontext.h>
 #include <vespa/fnet/frt/frt.h>
 #include <vespa/slobrok/sbmirror.h>
+#include <vespa/fnet/transport.h>
+#include <vespa/fastos/thread.h>
 
 using slobrok::api::IMirrorAPI;
 using slobrok::api::MirrorAPI;
@@ -14,7 +16,9 @@ namespace documentapi {
 ExternSlobrokPolicy::ExternSlobrokPolicy(const std::map<string, string>& param)
     : AsyncInitializationPolicy(param),
       _firstTry(true),
-      _orb(std::make_unique<FRT_Supervisor>()),
+      _threadPool(std::make_unique<FastOS_ThreadPool>(1024*60)),
+      _transport(std::make_unique<FNET_Transport>()),
+      _orb(std::make_unique<FRT_Supervisor>(_transport.get())),
       _slobrokConfigId("admin/slobrok.0")
 {
     if (param.find("config") != param.end()) {
@@ -42,10 +46,10 @@ ExternSlobrokPolicy::ExternSlobrokPolicy(const std::map<string, string>& param)
 
 ExternSlobrokPolicy::~ExternSlobrokPolicy()
 {
-    bool started = _mirror.get() != NULL;
+    bool started = (bool)_mirror;
     _mirror.reset();
     if (started) {
-        _orb->ShutDown(true);
+        _transport->ShutDown(true);
     }
 }
 
@@ -56,13 +60,12 @@ string ExternSlobrokPolicy::init() {
     } else if (_configSources.size() != 0) {
         slobrok::ConfiguratorFactory config(
             config::ConfigUri(_slobrokConfigId,
-                              config::IConfigContext::SP(
-                                  new config::ConfigContext(config::ServerSpec(_configSources)))));
+                             std::make_unique<config::ConfigContext>(config::ServerSpec(_configSources))));
         _mirror.reset(new MirrorAPI(*_orb, config));
     }
 
     if (_mirror.get()) {
-        _orb->Start();
+        _transport->Start(_threadPool.get());
     }
 
     return "";
diff --git a/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.h b/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.h
index dc0534095b6..d2966f852d5 100644
--- a/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.h
+++ b/documentapi/src/vespa/documentapi/messagebus/policies/externslobrokpolicy.h
@@ -9,6 +9,8 @@
 #include <vespa/config/subscription/sourcespec.h>
 
 class FRT_Supervisor;
+class FNET_Transport;
+class FastOS_ThreadPool;
 
 namespace documentapi {
 
@@ -22,6 +24,8 @@ protected:
     bool   _firstTry;
     config::ServerSpec::HostSpecList          _configSources;
     vespalib::Lock                            _lock;
+    std::unique_ptr<FastOS_ThreadPool>        _threadPool;
+    std::unique_ptr<FNET_Transport>           _transport;
     std::unique_ptr<FRT_Supervisor>           _orb;
     std::unique_ptr<slobrok::api::IMirrorAPI> _mirror;
     std::vector<std::string>                  _slobroks;