| 479 | } |
| 480 | |
| 481 | void Statestore::Topic::BuildDelta(const SubscriberId& subscriber_id, |
| 482 | TopicEntry::Version last_processed_version, |
| 483 | const string& filter_prefix, TTopicDelta* delta) { |
| 484 | // If the subscriber version is > 0, send this update as a delta. Otherwise, this is |
| 485 | // a new subscriber so send them a non-delta update that includes all entries in the |
| 486 | // topic. |
| 487 | delta->is_delta = last_processed_version > Subscriber::TOPIC_INITIAL_VERSION; |
| 488 | delta->__set_from_version(last_processed_version); |
| 489 | { |
| 490 | // Acquire shared lock - we are not modifying the topic. |
| 491 | shared_lock<shared_mutex> read_lock(lock_); |
| 492 | TopicUpdateLog::const_iterator next_update = |
| 493 | topic_update_log_.upper_bound(last_processed_version); |
| 494 | |
| 495 | uint64_t topic_size = 0; |
| 496 | for (; next_update != topic_update_log_.end(); ++next_update) { |
| 497 | TopicEntryMap::const_iterator itr = entries_.find(next_update->second); |
| 498 | DCHECK(itr != entries_.end()); |
| 499 | const TopicEntry& topic_entry = itr->second; |
| 500 | // Don't send deleted entries for non-delta updates. |
| 501 | if (!delta->is_delta && topic_entry.is_deleted()) { |
| 502 | continue; |
| 503 | } |
| 504 | // Skip any entries that don't match the requested prefix. |
| 505 | if (!HasPrefixString(itr->first, filter_prefix)) continue; |
| 506 | |
| 507 | delta->topic_entries.push_back(TTopicItem()); |
| 508 | TTopicItem& delta_entry = delta->topic_entries.back(); |
| 509 | delta_entry.key = itr->first; |
| 510 | delta_entry.value = topic_entry.value(); |
| 511 | delta_entry.deleted = topic_entry.is_deleted(); |
| 512 | topic_size += delta_entry.key.size() + delta_entry.value.size(); |
| 513 | } |
| 514 | |
| 515 | if (!delta->is_delta && |
| 516 | last_version_ > Subscriber::TOPIC_INITIAL_VERSION) { |
| 517 | VLOG_QUERY << "Preparing initial " << delta->topic_name |
| 518 | << " topic update for " << subscriber_id << ". Size = " |
| 519 | << PrettyPrinter::Print(topic_size, TUnit::BYTES); |
| 520 | } |
| 521 | |
| 522 | if (topic_update_log_.size() > 0) { |
| 523 | // The largest version for this topic will be the last entry in the version history |
| 524 | // map. |
| 525 | delta->__set_to_version(topic_update_log_.rbegin()->first); |
| 526 | } else { |
| 527 | // There are no updates in the version history |
| 528 | delta->__set_to_version(Subscriber::TOPIC_INITIAL_VERSION); |
| 529 | } |
| 530 | } |
| 531 | } |
| 532 | void Statestore::Topic::ToJson(Document* document, Value* topic_json) { |
| 533 | // Acquire shared lock - we are not modifying the topic. |
| 534 | shared_lock<shared_mutex> read_lock(lock_); |
no test coverage detected