* Initialize topic * Producer must be initialized and connected prior to calling this method. * Topic map will be updated. * * \param [in] topic_var MSGBUS_TOPIC_VAR_ * \param [in] router_group Router group - empty/NULL means no router group * \param [in] peer_group Peer group - empty/NULL means no peer group * \param [in] peer_asn Peer asn (remote a
| 321 | * \return (RdKafka::Topic *) pointer or NULL if error |
| 322 | */ |
| 323 | RdKafka::Topic * KafkaTopicSelector::initTopic(const std::string &topic_var, |
| 324 | const std::string *router_group, const std::string *peer_group, |
| 325 | uint32_t peer_asn) { |
| 326 | std::string errstr; |
| 327 | char uint32_str[12]; |
| 328 | |
| 329 | // Get the actual topic name based on var |
| 330 | std::string topic_name = this->cfg->topic_names_map[topic_var]; |
| 331 | |
| 332 | /* |
| 333 | * topics that contain the peer asn need to have the key include the peer asn |
| 334 | */ |
| 335 | if (topic_name.find("{peer_asn}") != std::string::npos) { |
| 336 | topic_flags_map[topic_var].include_peerAsn = true; |
| 337 | SELF_DEBUG("peer_asn found in topic %s, setting topic flag to include peer ASN", topic_name.c_str()); |
| 338 | } else { |
| 339 | topic_flags_map[topic_var].include_peerAsn = false; |
| 340 | } |
| 341 | |
| 342 | // Update the topic key based on the peer_group/router_group |
| 343 | std::string topic_key = getTopicKey(topic_var, router_group, peer_group, peer_asn); |
| 344 | |
| 345 | // Update the topic name based on app variables |
| 346 | if (topic_var.compare(MSGBUS_TOPIC_VAR_COLLECTOR)) { // if not collector topic |
| 347 | if (router_group != NULL and router_group->size() > 0) { |
| 348 | boost::replace_all(topic_name, "{router_group}", *router_group); |
| 349 | } else |
| 350 | boost::replace_all(topic_name, "{router_group}", "default"); |
| 351 | |
| 352 | if (topic_var.compare(MSGBUS_TOPIC_VAR_ROUTER)) { // if not router topic |
| 353 | if (peer_group != NULL and peer_group->size() > 0) { |
| 354 | boost::replace_all(topic_name, "{peer_group}", *peer_group); |
| 355 | } else |
| 356 | boost::replace_all(topic_name, "{peer_group}", "default"); |
| 357 | |
| 358 | if (peer_asn > 0) { |
| 359 | snprintf(uint32_str, sizeof(uint32_str), "%u", peer_asn); |
| 360 | boost::replace_all(topic_name, "{peer_asn}", (const char *)uint32_str); |
| 361 | } else |
| 362 | boost::replace_all(topic_name, "{peer_asn}", "default"); |
| 363 | } |
| 364 | } |
| 365 | |
| 366 | SELF_DEBUG("Creating topic %s (map key=%s)" , topic_name.c_str(), topic_key.c_str()); |
| 367 | |
| 368 | // Delete topic if it already exists |
| 369 | topic_map::iterator t_it; |
| 370 | |
| 371 | if ( (t_it=topic.find(topic_key)) != topic.end() and t_it->second != NULL) { |
| 372 | delete t_it->second; |
| 373 | } |
| 374 | |
| 375 | /* |
| 376 | * Topic configuration |
| 377 | */ |
| 378 | if (tconf->set("partitioner_cb", peer_partitioner_callback, errstr) != RdKafka::Conf::CONF_OK) { |
| 379 | LOG_ERR("Failed to configure kafka partitioner callback: %s", errstr.c_str()); |
| 380 | throw "ERROR: Failed to configure kafka partitioner callback"; |