Status RDB::saveRdbObject()

in src/storage/rdb/rdb.cc [459:521]
62 lines of code
26 McCabe index (conditional complexity)

Status RDB::saveRdbObject(engine::Context &ctx, int type, const std::string &key, const RedisObjValue &obj,
                          uint64_t ttl_ms) {
  rocksdb::Status db_status;
  if (type == RDBTypeString) {
    const auto &value = std::get<std::string>(obj);
    redis::String string_db(storage_, ns_);
    uint64_t expire_ms = 0;
    if (ttl_ms > 0) {
      expire_ms = ttl_ms + util::GetTimeStampMS();
    }
    db_status = string_db.SetEX(ctx, key, value, expire_ms);
  } else if (type == RDBTypeSet || type == RDBTypeSetIntSet || type == RDBTypeSetListPack) {
    const auto &members = std::get<std::vector<std::string>>(obj);
    redis::Set set_db(storage_, ns_);
    uint64_t count = 0;
    std::vector<Slice> insert_members;
    insert_members.reserve(members.size());
    for (const auto &member : members) {
      insert_members.emplace_back(member);
    }
    db_status = set_db.Add(ctx, key, insert_members, &count);
  } else if (type == RDBTypeZSet || type == RDBTypeZSet2 || type == RDBTypeZSetListPack || type == RDBTypeZSetZipList) {
    const auto &member_scores = std::get<std::vector<MemberScore>>(obj);
    redis::ZSet zset_db(storage_, ns_);
    uint64_t count = 0;
    db_status = zset_db.Add(ctx, key, ZAddFlags(0), const_cast<std::vector<MemberScore> *>(&member_scores), &count);
  } else if (type == RDBTypeHash || type == RDBTypeHashListPack || type == RDBTypeHashZipList ||
             type == RDBTypeHashZipMap) {
    const auto &entries = std::get<std::map<std::string, std::string>>(obj);
    std::vector<FieldValue> field_values;
    field_values.reserve(entries.size());
    for (const auto &entry : entries) {
      field_values.emplace_back(entry.first, entry.second);
    }
    redis::Hash hash_db(storage_, ns_);
    uint64_t count = 0;
    db_status = hash_db.MSet(ctx, key, field_values, false /*nx*/, &count);
  } else if (type == RDBTypeList || type == RDBTypeListZipList || type == RDBTypeListQuickList ||
             type == RDBTypeListQuickList2) {
    const auto &elements = std::get<std::vector<std::string>>(obj);
    if (!elements.empty()) {
      std::vector<Slice> insert_elements;
      insert_elements.reserve(elements.size());
      for (const auto &element : elements) {
        insert_elements.emplace_back(element);
      }
      redis::List list_db(storage_, ns_);
      uint64_t list_size = 0;
      db_status = list_db.Push(ctx, key, insert_elements, false, &list_size);
    }
  } else {
    return {Status::RedisExecErr, fmt::format("unsupported save type: {}", type)};
  }
  if (!db_status.ok()) {
    return {Status::RedisExecErr, db_status.ToString()};
  }
  // String type will use the SETEX, so just only set the ttl for other types
  if (ttl_ms > 0 && type != RDBTypeString) {
    redis::Database db(storage_, ns_);
    db_status = db.Expire(ctx, key, ttl_ms + util::GetTimeStampMS());
  }
  return db_status.ok() ? Status::OK() : Status{Status::RedisExecErr, db_status.ToString()};
}