Bug fix for coral MessageService wrapper

CondCore/CondDB/interface/ConnectionPool.h

@@ -64,14 +64,13 @@ namespace cond {
       std::string m_authPath = std::string("");
       int m_authSys = 0;
       coral::MsgLevel m_messageLevel = coral::Error;
-      std::unique_ptr<CoralMsgReporter> m_msgReporter;
+      CoralMsgReporter* m_msgReporter = nullptr;
       bool m_loggingEnabled = false;
       //The frontier security option is turned on for all sessions
       //usig this wrapper of the CORAL connection setup for configuring the server access
       std::string m_frontierSecurity = std::string("");
       // this one has to be moved!
       cond::CoralServiceManager* m_pluginManager = nullptr;
-      std::map<std::string, int> m_dbTypes;
     };
   }  // namespace persistency
 }  // namespace cond

CondCore/CondDB/interface/Logger.h

@@ -14,13 +14,14 @@
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
 //
 #include <sstream>
+#include <memory>
 #include <boost/date_time/posix_time/posix_time.hpp>
 
 namespace cond {
 
   namespace persistency {
 
-    class ConnectionPool;
+    class MsgDispatcher;
 
     template <typename EdmLogger>
     class EchoedLogStream {
@@ -93,7 +94,10 @@ namespace cond {
       virtual ~Logger();
 
       //
-      void setDbDestination(const std::string& connectionString, ConnectionPool& connectionPool);
+      void subscribeCoralMessages(const std::weak_ptr<MsgDispatcher>& dispatcher);
+
+      //
+      void setDbDestination(const std::string& connectionString);
 
       //
       void start();
@@ -124,12 +128,12 @@ namespace cond {
     private:
       std::string m_jobName;
       std::string m_connectionString;
-      ConnectionPool* m_sharedConnectionPool;
       bool m_started;
       boost::posix_time::ptime m_startTime;
       boost::posix_time::ptime m_endTime;
       int m_retCode;
       std::stringstream m_log;
+      std::weak_ptr<MsgDispatcher> m_dispatcher;
     };
 
   }  // namespace persistency

CondCore/CondDB/src/ConnectionPool.cc

@@ -23,6 +23,9 @@ namespace cond {
 
     ConnectionPool::ConnectionPool() {
       m_pluginManager = new cond::CoralServiceManager;
+      m_msgReporter = new CoralMsgReporter;
+      // ownership is acquired ( instrinsically unsafe, need to change coral )
+      coral::MessageStream::installMsgReporter(m_msgReporter);
       configure();
     }
 
@@ -73,10 +76,8 @@ namespace cond {
       coralConfig.disableConnectionSharing();
       // message streaming
       coral::MessageStream::setMsgVerbosity(m_messageLevel);
-      if (m_msgReporter.get() != nullptr) {
-        m_msgReporter->setOutputLevel(m_messageLevel);
-        coral::MessageStream::installMsgReporter(static_cast<coral::IMsgReporter*>(m_msgReporter.get()));
-      }
+      m_msgReporter->setOutputLevel(m_messageLevel);
+
       // authentication
       std::string authServiceName("CORAL/Services/EnvironmentAuthenticationService");
       std::string authPath = m_authPath;
@@ -171,9 +172,7 @@ namespace cond {
 
     void ConnectionPool::setMessageVerbosity(coral::MsgLevel level) { m_messageLevel = level; }
 
-    void ConnectionPool::setLogDestination(Logger& logger) {
-      m_msgReporter = std::make_unique<CoralMsgReporter>(logger);
-    }
+    void ConnectionPool::setLogDestination(Logger& logger) { m_msgReporter->subscribe(logger); }
 
   }  // namespace persistency
 }  // namespace cond

CondCore/CondDB/src/CoralMsgReporter.cc

@@ -7,9 +7,17 @@
 #include "CondCore/CondDB/interface/Logger.h"
 #include "CoralMsgReporter.h"
 
+cond::persistency::MsgDispatcher::MsgDispatcher(Logger& logger) { m_recipient = &logger; }
+
+void cond::persistency::MsgDispatcher::unsubscribe() { m_recipient = nullptr; }
+
+bool cond::persistency::MsgDispatcher::hasRecipient() { return m_recipient != nullptr; }
+
+cond::persistency::Logger& cond::persistency::MsgDispatcher::recipient() { return *m_recipient; }
+
 /// Default constructor
-cond::persistency::CoralMsgReporter::CoralMsgReporter(Logger& logger)
-    : m_logger(logger), m_level(coral::Error), m_format(0), m_mutex() {
+cond::persistency::CoralMsgReporter::CoralMsgReporter()
+    : m_dispatcher(), m_level(coral::Error), m_format(0), m_mutex() {
   // Use a non-default format?
   //char* msgformat = getenv ( "CORAL_MSGFORMAT" );
   if (getenv("CORAL_MESSAGEREPORTER_FORMATTED"))
@@ -79,58 +87,61 @@ coral::MsgLevel cond::persistency::CoralMsgReporter::outputLevel() const { retur
 /// Modify output level
 void cond::persistency::CoralMsgReporter::setOutputLevel(coral::MsgLevel lvl) { m_level = lvl; }
 
+void reportToRecipient(const std::string& msg, int lvl, cond::persistency::Logger& recipient) {
+  switch (lvl) {
+    case coral::Nil:
+    case coral::Verbose:
+    case coral::Debug:
+      recipient.logDebug() << "CORAL: " << msg;
+      break;
+    case coral::Info:
+      recipient.logInfo() << "CORAL: " << msg;
+      break;
+    case coral::Warning:
+      recipient.logWarning() << "CORAL: " << msg;
+      break;
+    case coral::Error:
+      recipient.logError() << "CORAL: " << msg;
+      break;
+  }
+}
+
 /// Report message to stdout
-void cond::persistency::CoralMsgReporter::report(int lvl, const std::string& src, const std::string& msg) {
+void cond::persistency::CoralMsgReporter::report(int lvl, const std::string&, const std::string& msg) {
   if (lvl < m_level)
     return;
   std::lock_guard<std::recursive_mutex> lock(m_mutex);
 
-  std::stringstream out;
-  /**
-  if ( m_format == 1 ) // COOL format
-  {
-    // Formatted CORAL reporter (as in COOL)
-    //std::ostream& out = std::cout;
-    const std::string::size_type src_name_maxsize = 36;
-    if ( src.size() <= src_name_maxsize )
-    {
-      out << src << std::string( src_name_maxsize-src.size(), ' ' );
-    }
-    else
-    {
-      out << src.substr( 0, src_name_maxsize-3 ) << "...";
-    }
-    switch ( lvl )
-    {
-    case 0:  out << " Nil      "; break;
-    case 1:  out << " Verbose  "; break;
-    case 2:  out << " Debug    "; break;
-    case 3:  out << " Info     "; break;
-    case 4:  out << " Warning  "; break;
-    case 5:  out << " Error    "; break;
-    case 6:  out << " Fatal    "; break;
-    case 7:  out << " Always   "; break;
-    default: out << " Unknown  "; break;
-    }
-    out << msg << std::endl;
+  if (m_dispatcher.get() && m_dispatcher->hasRecipient()) {
+    reportToRecipient(msg, lvl, m_dispatcher->recipient());
   }
-  else{
-  **/
   // Default CORAL reporter
+  std::string level("");
   switch (lvl) {
     case coral::Nil:
+      level = "Nil";
+      break;
     case coral::Verbose:
+      level = "Verbose";
+      break;
     case coral::Debug:
-      m_logger.logDebug() << "CORAL: " << msg;
+      level = "Debug";
       break;
     case coral::Info:
-      m_logger.logInfo() << "CORAL: " << msg;
+      level = "Info";
       break;
     case coral::Warning:
-      m_logger.logWarning() << "CORAL: " << msg;
+      level = "Warning";
       break;
     case coral::Error:
-      m_logger.logError() << "CORAL: " << msg;
+      level = "Error";
       break;
   }
+  std::cout << msg << " " << level << " " << msg << std::endl;
+}
+
+void cond::persistency::CoralMsgReporter::subscribe(Logger& logger) {
+  m_dispatcher.reset(new MsgDispatcher(logger));
+  std::weak_ptr<MsgDispatcher> callBack(m_dispatcher);
+  logger.subscribeCoralMessages(callBack);
 }

CondCore/CondDB/src/CoralMsgReporter.h

@@ -4,6 +4,7 @@
 //#include "CondCore/CondDB/interface/Logger.h"
 
 #include <mutex>
+#include <set>
 #include "CoralBase/MessageStream.h"
 
 namespace cond {
@@ -12,16 +13,28 @@ namespace cond {
 
     class Logger;
 
-    class CoralMsgReporter : public coral::IMsgReporter {
+    class MsgDispatcher {
     public:
-      // Empty ctr is suppressed
-      CoralMsgReporter() = delete;
+      MsgDispatcher() = delete;
+      explicit MsgDispatcher(Logger& logger);
+      virtual ~MsgDispatcher() {}
+
+      void unsubscribe();
+
+      bool hasRecipient();
+      Logger& recipient();
 
+    private:
+      Logger* m_recipient = nullptr;
+    };
+
+    class CoralMsgReporter : public coral::IMsgReporter {
+    public:
       /// Default constructor
-      explicit CoralMsgReporter(Logger& logger);
+      CoralMsgReporter();
 
       /// Destructor
-      ~CoralMsgReporter() override {}
+      ~CoralMsgReporter() override{};
 
       /// Release reference to reporter
       void release() override { delete this; }
@@ -35,9 +48,11 @@ namespace cond {
       /// Report a message
       void report(int lvl, const std::string& src, const std::string& msg) override;
 
+      void subscribe(Logger& logger);
+
     private:
       // the destination of the streams...
-      Logger& m_logger;
+      std::shared_ptr<MsgDispatcher> m_dispatcher;
 
       /// The current message level threshold
       coral::MsgLevel m_level;

CondCore/CondDB/src/Logger.cc

@@ -1,10 +1,13 @@
 #include "CondCore/CondDB/interface/Logger.h"
-#include "CondCore/CondDB/interface/ConnectionPool.h"
+#include "CondCore/CondDB/interface/Auth.h"
 #include "CondCore/CondDB/interface/Exception.h"
 //
 #include "DbCore.h"
+#include "CoralMsgReporter.h"
 #include "RelationalAccess/ITransaction.h"
 //
+#include "RelationalAccess/ConnectionService.h"
+#include "RelationalAccess/ISessionProxy.h"
 #include <boost/date_time/posix_time/posix_time_io.hpp>
 #include <fstream>
 //
@@ -45,21 +48,23 @@ namespace cond {
     Logger::Logger(const std::string& jobName)
         : m_jobName(jobName),
           m_connectionString(""),
-          m_sharedConnectionPool(nullptr),
           m_started(false),
           m_startTime(),
           m_endTime(),
           m_retCode(0),
           m_log() {}
 
     //
-    Logger::~Logger() {}
-
-    void Logger::setDbDestination(const std::string& connectionString, ConnectionPool& connectionPool) {
-      m_connectionString = connectionString;
-      m_sharedConnectionPool = &connectionPool;
+    Logger::~Logger() {
+      auto dispatcher = m_dispatcher.lock();
+      if (dispatcher.get())
+        dispatcher->unsubscribe();
     }
 
+    void Logger::subscribeCoralMessages(const std::weak_ptr<MsgDispatcher>& dispatcher) { m_dispatcher = dispatcher; }
+
+    void Logger::setDbDestination(const std::string& connectionString) { m_connectionString = connectionString; }
+
     //
     void Logger::start() {
       if (!m_started) {
@@ -114,13 +119,12 @@ namespace cond {
     //
     void Logger::saveOnDb() {
       if (!m_log.str().empty()) {
-        if (m_sharedConnectionPool == nullptr) {
-          throwException("Connection pool handle has not been provided.", "Logger::saveOnDb");
-        }
         if (m_connectionString.empty()) {
           throwException("Connection string for destination database has not been provided.", "Logger::saveOnDb");
         }
-        auto coralSession = m_sharedConnectionPool->createCoralSession(m_connectionString, true);
+        coral::ConnectionService connServ;
+        std::unique_ptr<coral::ISessionProxy> coralSession(
+            connServ.connect(m_connectionString, auth::COND_WRITER_ROLE, coral::Update));
         coralSession->transaction().start(false);
         try {
           O2O_RUN::Table destinationTable(coralSession->nominalSchema());

CondCore/DBOutputService/src/PoolDBOutputService.cc

@@ -55,7 +55,7 @@ cond::service::PoolDBOutputService::PoolDBOutputService(const edm::ParameterSet&
   m_session = m_connection.createSession(connectionString, true);
   bool saveLogsOnDb = iConfig.getUntrackedParameter<bool>("saveLogsOnDB", false);
   if (saveLogsOnDb)
-    m_logger.setDbDestination(connectionString, m_connection);
+    m_logger.setDbDestination(connectionString);
   // implicit start
   doStartTransaction();