dunedaq::kafkaopmon::OpMonPublisher Class Reference

#include <OpMonPublisher.hpp>

Public Member Functions
	OpMonPublisher (const nlohmann::json &conf)
	OpMonPublisher ()=delete
	OpMonPublisher (const OpMonPublisher &)=delete
OpMonPublisher &	operator= (const OpMonPublisher &)=delete
	OpMonPublisher (OpMonPublisher &&)=delete
OpMonPublisher &	operator= (OpMonPublisher &&)=delete
	~OpMonPublisher ()
void	publish (dunedaq::opmon::OpMonEntry &&) const

Protected Member Functions
std::string	extract_topic (const dunedaq::opmon::OpMonEntry &) const noexcept
std::string	extract_key (const dunedaq::opmon::OpMonEntry &e) const noexcept

Private Attributes
std::unique_ptr< RdKafka::Producer >	m_producer
std::string	m_default_topic = "monitoring.opmon_stream"

Detailed Description

Definition at line 65 of file OpMonPublisher.hpp.

Constructor & Destructor Documentation

OpMonPublisher() [1/4]

OpMonPublisher::OpMonPublisher

(

const nlohmann::json &

conf

)

Definition at line 14 of file OpMonPublisher.cpp.

                                                        {
 
  RdKafka::Conf * k_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  std::string errstr;
 
  // Good observations on threadsafety here
  // https://www.confluent.io/blog/modern-cpp-kafka-api-for-safe-easy-messaging/
  
  auto it = conf.find("bootstrap");
  if ( it == conf.end() ) {
    throw MissingParameter(ERS_HERE,
               "bootstrap",
               nlohmann::to_string(conf) );
  }
  
  k_conf->set("bootstrap.servers", *it, errstr);
  if( ! errstr.empty() ) {
    throw FailedConfiguration(ERS_HERE,
                  "bootstrap.servers",
                  errstr);
  }
  
  std::string client_id;
  it = conf.find( "cliend_id" );
  if ( it != conf.end() )
    client_id = *it;
  else
    client_id = "kafkaopmon_default_producer";
  
  k_conf->set("client.id", client_id, errstr);    
  if ( ! errstr.empty() ) {
    ers::error( FailedConfiguration(ERS_HERE, "client.id", errstr ) );
  }
 
  // Create producer instance
  m_producer.reset(RdKafka::Producer::create(k_conf, errstr));
    
  if( ! m_producer ){
    throw FailedProducerCreation(ERS_HERE, errstr);
  }
    
  it = conf.find("default_topic");
  if (it != conf.end()) m_default_topic = "monitoring." + it->get<std::string>();
  
}

OpMonPublisher() [2/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( )

delete

OpMonPublisher() [3/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( const OpMonPublisher & )

delete

OpMonPublisher() [4/4]

dunedaq::kafkaopmon::OpMonPublisher::OpMonPublisher ( OpMonPublisher && )

delete

~OpMonPublisher()

OpMonPublisher::~OpMonPublisher

(

)

Definition at line 61 of file OpMonPublisher.cpp.

                                {
 
  int timeout_ms = 500;
  RdKafka::ErrorCode err = m_producer -> flush( timeout_ms );
 
  if ( err == RdKafka::ERR__TIMED_OUT ) {
    ers::warning( TimeoutReachedWhileFlushing( ERS_HERE, timeout_ms ) );
  }
  
}

Member Function Documentation

extract_key()

std::string dunedaq::kafkaopmon::OpMonPublisher::extract_key ( const dunedaq::opmon::OpMonEntry & e ) const

inlineprotectednoexcept

Definition at line 84 of file OpMonPublisher.hpp.

                                                                            {
      return dunedaq::opmonlib::to_string(e.origin()) + '/' + e.measurement() ;
    }

extract_topic()

std::string dunedaq::kafkaopmon::OpMonPublisher::extract_topic ( const dunedaq::opmon::OpMonEntry & ) const

inlineprotectednoexcept

Definition at line 81 of file OpMonPublisher.hpp.

                                                                             {
      return m_default_topic; 
    }

operator=() [1/2]

OpMonPublisher & dunedaq::kafkaopmon::OpMonPublisher::operator= ( const OpMonPublisher & )

delete

operator=() [2/2]

OpMonPublisher & dunedaq::kafkaopmon::OpMonPublisher::operator= ( OpMonPublisher && )

delete

publish()

void OpMonPublisher::publish

(

dunedaq::opmon::OpMonEntry &&

entry

)

const

Definition at line 73 of file OpMonPublisher.cpp.

                                                                    {
 
  std::string binary;
  entry.SerializeToString( & binary );
 
  auto topic = extract_topic( entry );
  auto key   = extract_key( entry );
 
  RdKafka::ErrorCode err = m_producer -> produce( topic,
                          RdKafka::Topic::PARTITION_UA,
                          RdKafka::Producer::RK_MSG_COPY,
                          const_cast<char *>(binary.c_str()), binary.size(),
                          key.c_str(), key.size(),
                          0,
                          nullptr
                          );
 
  if ( err == RdKafka::ERR_NO_ERROR ) return ;
 
  std::string err_cause;
  
  switch( err ) {
  case RdKafka::ERR__QUEUE_FULL :
    err_cause = "maximum number of outstanding messages reached";
    break;
  case RdKafka::ERR_MSG_SIZE_TOO_LARGE :
    err_cause = "message too large";
    break;
  case RdKafka::ERR__UNKNOWN_PARTITION :
    err_cause = "Unknown partition";
    break;
  case RdKafka::ERR__UNKNOWN_TOPIC :
    err_cause = "Unknown topic (" ;
    err_cause += topic;
    err_cause += ')';
    break;
  default:
    err_cause = "unknown";
    break;
  }
 
  throw FailedProduce(ERS_HERE, key, err_cause) ;
 
}

Member Data Documentation

m_default_topic

std::string dunedaq::kafkaopmon::OpMonPublisher::m_default_topic = "monitoring.opmon_stream"

private

Definition at line 90 of file OpMonPublisher.hpp.

m_producer

std::unique_ptr<RdKafka::Producer> dunedaq::kafkaopmon::OpMonPublisher::m_producer

private

Definition at line 89 of file OpMonPublisher.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/kafkaopmon/include/kafkaopmon/OpMonPublisher.hpp
• /github/workspace/dunedaq/sourcecode/kafkaopmon/src/OpMonPublisher.cpp