int MCRouterTestClient::del(const dynamic &keys, bool local,
                            dynamic &results) {

  std::vector<mcrouter_msg_t> msgs(keys.size());
  dynamic raw_results = dynamic::object;

  for (int i = 0; i < keys.size(); i ++) {
    auto key = keys[i].asString().toStdString();
    if (!local) {
      key = "/*/*/" + key;
    }
    msgs[i] = make_delete_request(key);
  }

  int ret = 0;
  bool res = issueRequests(msgs.data(), keys.size(), raw_results);
  if (res) {
    for ( auto& raw_reply : raw_results.items() ) {
      bool found = (raw_reply.second["result"] == (int) mc_res_deleted);
      results[raw_reply.first] = found;
      ret += (int) found;
    }
  }
  return ret;
}

static void bserEncode(dynamic const& dyn,
                       QueueAppender& appender,
                       const serialization_opts& opts) {
  switch (dyn.type()) {
    case dynamic::Type::NULLT:
      appender.write((int8_t)BserType::Null);
      return;
    case dynamic::Type::BOOL:
      appender.write(
          (int8_t)(dyn.getBool() ? BserType::True : BserType::False));
      return;
    case dynamic::Type::DOUBLE: {
      double dval = dyn.getDouble();
      appender.write((int8_t)BserType::Real);
      appender.write(dval);
      return;
    }
    case dynamic::Type::INT64:
      bserEncodeInt(dyn.getInt(), appender);
      return;
    case dynamic::Type::OBJECT:
      bserEncodeObject(dyn, appender, opts);
      return;
    case dynamic::Type::ARRAY:
      bserEncodeArray(dyn, appender, opts);
      return;
    case dynamic::Type::STRING:
      bserEncodeString(dyn.getString(), appender);
      return;
  }
}

 Optional<SchemaError> validate(ValidationContext& vc,
                                const dynamic& value) const override {
   if (!value.isArray()) {
     return none;
   }
   if (itemsValidator_) {
     for (const auto& v : value) {
       if (auto se = vc.validate(itemsValidator_.get(), v)) {
         return se;
       }
     }
     return none;
   }
   size_t pos = 0;
   for (; pos < value.size() && pos < itemsValidators_.size(); ++pos) {
     if (auto se = vc.validate(itemsValidators_[pos].get(), value[pos])) {
       return se;
     }
   }
   if (!allowAdditionalItems_ && pos < value.size()) {
     return makeError("no more additional items", value);
   }
   if (additionalItemsValidator_) {
     for (; pos < value.size(); ++pos) {
       if (auto se =
               vc.validate(additionalItemsValidator_.get(), value[pos])) {
         return se;
       }
     }
   }
   return none;
 }

SettingsMap::SettingsMap(const dynamic& d) {
  if (!d.isObject()) {
    throw BistroException("Can only create settings map from an object");
  }
  for (const auto& pair : d.items()) {
    const string& name = pair.first.asString().toStdString();
    set(name, pair.second);
  }
}

CrontabSelector::CrontabSelector(
  const dynamic &d,
  int64_t min_val,
  int64_t max_val,
  function<int64_t(const string& lc_str)> str_to_value
) : start_(min_val),
    end_(max_val),
    period_(1),
    minVal_(min_val),
    maxVal_(max_val) {

  switch (d.type()) {
    case dynamic::Type::INT64:
    case dynamic::Type::STRING:
      sortedValues_.emplace_back(parseValue(d, str_to_value));
      break;
    case dynamic::Type::ARRAY:
      for (const auto& val : d) {
        sortedValues_.emplace_back(parseValue(val, str_to_value));
      }
      // If somebody specifies [] for a selector, we have to silently
      // accept it, since PHP's JSON library can morph {} into [], and {}
      // must mean "default selector accepting all values."
      break;
    case dynamic::Type::OBJECT:
      for (const auto& pair : d.items()) {
        // Interval is first so that it doesn't accept strings like "jan"
        if (pair.first == "period") {
          period_ = pair.second.asInt();
          if (period_ < 1 || period_ >= maxVal_ - minVal_) {
            throw runtime_error(format(
              "period not in [1, {}]: {}", maxVal_ - minVal_, period_
            ).str());
          }
          continue;
        }
        // For start & end, we are happy to accept string names
        auto val = parseValue(pair.second, str_to_value);
        if (pair.first == "start") {
          start_ = val;
        } else if (pair.first == "end") {
          end_ = val;
        } else {
          throw runtime_error(format("Unknown key: {}", pair.first).str());
        }
      }
      // If we got an empty object, no problem -- this selector will
      // follow the default of "match everything".
      break;
    default:
      throw runtime_error(format(
        "Bad type for crontab selector: {}", d.typeName()
      ).str());
  }
  sort(sortedValues_.begin(), sortedValues_.end());
}

dynamic dynamic::merge_diff(const dynamic& source, const dynamic& target) {
  if (!source.isObject() || source.type() != target.type()) {
    return target;
  }

  dynamic diff = object;

  // added/modified keys
  for (const auto& pair : target.items()) {
    auto it = source.find(pair.first);
    if (it == source.items().end()) {
      diff[pair.first] = pair.second;
    } else {
      diff[pair.first] = merge_diff(source[pair.first], target[pair.first]);
    }
  }

  // removed keys
  for (const auto& pair : source.items()) {
    auto it = target.find(pair.first);
    if (it == target.items().end()) {
      diff[pair.first] = nullptr;
    }
  }

  return diff;
}

 explicit TypeValidator(const dynamic& schema) {
   if (schema.isString()) {
     addType(schema.stringPiece());
   } else if (schema.isArray()) {
     for (const auto& item : schema) {
       if (item.isString()) {
         addType(item.stringPiece());
       }
     }
   }
 }

EpochCrontabItem::EpochCrontabItem(
  const dynamic& d, boost::local_time::time_zone_ptr tz
) : CrontabItem(tz),
    epoch_sel_(d.getDefault("epoch"), 0, std::numeric_limits<time_t>::max()) {

  if (d.size() != 1) {
    throw runtime_error(
      "Can only have the 'epoch' key: " + folly::toJson(d).toStdString()
    );
  }
}

void checkItemParseError(
    // real_k can differ from err_k, which is typically coerced to string
    dynamic d, Fn fn, dynamic real_k, dynamic err_k, std::string re) {
  DynamicParser p(DynamicParser::OnError::RECORD, &d);
  fn(p);
  auto errors = p.releaseErrors();
  auto error = errors.at("nested").at(err_k);
  EXPECT_EQ(d.at(real_k), error.at("value"));
  EXPECT_PCRE_MATCH(re, error.at("error").getString());
  EXPECT_EQ(dynamic(dynamic::object("nested", dynamic::object(
    err_k, dynamic::object("value", d.at(real_k))("error", error.at("error"))
  ))), errors);
}

bool dynamic::operator==(dynamic const& o) const {
  if (type() != o.type()) {
    if (isNumber() && o.isNumber()) {
      auto& integ = isInt() ? *this : o;
      auto& doubl = isInt() ? o     : *this;
      return integ.asInt() == doubl.asDouble();
    }
    return false;
  }

#define FB_X(T) return *getAddress<T>() == *o.getAddress<T>();
  FB_DYNAMIC_APPLY(type_, FB_X);
#undef FB_X
}

TEST(Dynamic, At) {
  const dynamic cd = dynamic::object("key1", make_long_string());
  dynamic dd = dynamic::object("key1", make_long_string());
  dynamic md = dynamic::object("key1", make_long_string());

  dynamic ds(make_long_string());
  EXPECT_EQ(ds, cd.at("key1"));
  EXPECT_EQ(ds, cd.at("key1"));

  dd.at("key1").getString() += " hello";
  EXPECT_EQ(dynamic(make_long_string() + " hello"), dd.at("key1"));
  EXPECT_EQ(dynamic(make_long_string() + " hello"), dd.at("key1"));

  EXPECT_EQ(ds, std::move(md).at("key1"));
  EXPECT_NE(ds, md.at("key1"));
}

 AllOfValidator(SchemaValidatorContext& context, const dynamic& schema) {
   if (schema.isArray()) {
     for (const auto& item : schema) {
       validators_.emplace_back(SchemaValidator::make(context, item));
     }
   }
 }

 explicit RequiredValidator(const dynamic& schema) {
   if (schema.isArray()) {
     for (const auto& item : schema) {
       if (item.isString()) {
         properties_.emplace_back(item.getString());
       }
     }
   }
 }

static void bserEncodeArraySimple(dynamic const& dyn,
                                  QueueAppender& appender,
                                  const serialization_opts& opts) {
  appender.write((int8_t)BserType::Array);
  bserEncodeInt(dyn.size(), appender);
  for (const auto& ele : dyn) {
    bserEncode(ele, appender, opts);
  }
}

TEST(Dynamic, At) {
  const dynamic cd = dynamic::object("key1", make_long_string());
  dynamic dd = dynamic::object("key1", make_long_string());
  dynamic md = dynamic::object("key1", make_long_string());

  dynamic ds(make_long_string());
  EXPECT_EQ(ds, cd.at("key1"));
  EXPECT_EQ(ds, cd.at("key1"));

  dd.at("key1").getString() += " hello";
  EXPECT_EQ(dynamic(make_long_string() + " hello"), dd.at("key1"));
  EXPECT_EQ(dynamic(make_long_string() + " hello"), dd.at("key1"));

  EXPECT_EQ(ds, std::move(md).at("key1")); // move available, but not performed
  EXPECT_EQ(ds, md.at("key1"));
  dynamic moved = std::move(md).at("key1"); // move performed
  EXPECT_EQ(ds, moved);
  EXPECT_NE(ds, md.at("key1"));
}

static void bserEncodeObject(dynamic const& dyn,
                             QueueAppender& appender,
                             const serialization_opts& opts) {
  appender.write((int8_t)BserType::Object);
  bserEncodeInt(dyn.size(), appender);

  if (opts.sort_keys) {
    std::vector<std::pair<dynamic, dynamic>> sorted(dyn.items().begin(),
                                                    dyn.items().end());
    std::sort(sorted.begin(), sorted.end());
    for (const auto& item : sorted) {
      bserEncode(item.first, appender, opts);
      bserEncode(item.second, appender, opts);
    }
  } else {
    for (const auto& item : dyn.items()) {
      bserEncode(item.first, appender, opts);
      bserEncode(item.second, appender, opts);
    }
  }
}

JobBackoffSettings::JobBackoffSettings(const dynamic& d) : repeat_(false) {
  if (!d.isArray()) {
    throw BistroException("Expected array; got ", folly::toJson(d));
  }
  if (d.empty()) {
    throw BistroException("Backoff setting is empty");
  }
  for (const auto& item : d) {
    if (item.isInt()) {
      int val = item.asInt();
      if (val <= 0) {
        throw BistroException("Backoff times must be positive: ", val);
      }
      if (!values_.empty()) {
        if (values_.back() == val) {
          throw BistroException("Duplicate backoff time: ", val);
        }
        if (values_.back() > val) {
          throw BistroException("Backoff times must be in increasing order");
        }
      }
      values_.push_back(val);
    } else if (item.isString()) {
      const auto& s = item.asString();
      if (s == "repeat") {
        if (values_.empty()) {
          throw BistroException("No backoff interval given before 'repeat'");
        }
        repeat_ = true;
      } else if  (s != "fail") {
        throw BistroException("Unknown string in backoff settings: ", s);
      }
      break;
    } else {
      throw BistroException("Invalid backoff value: ", folly::toJson(item));
    }
  }
}

int64_t CrontabSelector::parseValue(
    const dynamic& d,
    function<int64_t(const string& lc_str)> str_to_value) {

  int64_t res;
  if (d.isInt()) {
    res = d.asInt();
  } else if (d.isString()) {
    auto s = d.asString();
    if (str_to_value == nullptr) {
      throw runtime_error("Cannot parse string " + s);
    }
    transform(s.begin(), s.end(), s.begin(), ::tolower);
    res = str_to_value(s);
  } else {
    throw runtime_error(format("Cannot parse {}", folly::toJson(d)).str());
  }
  if (res < minVal_ || res > maxVal_) {
    throw runtime_error(format(
      "Value {} out of range [{}, {}]", res, minVal_, maxVal_
    ).str());
  }
  return res;
}

 DependencyValidator(SchemaValidatorContext& context, const dynamic& schema) {
   if (!schema.isObject()) {
     return;
   }
   for (const auto& pair : schema.items()) {
     if (!pair.first.isString()) {
       continue;
     }
     if (pair.second.isArray()) {
       auto p = make_pair(pair.first.getString(), std::vector<fbstring>());
       for (const auto& item : pair.second) {
         if (item.isString()) {
           p.second.push_back(item.getString());
         }
       }
       propertyDep_.emplace_back(std::move(p));
     }
     if (pair.second.isObject()) {
       schemaDep_.emplace_back(
           make_pair(pair.first.getString(),
                     SchemaValidator::make(context, pair.second)));
     }
   }
 }

static void GenStat(Stat& stat, const dynamic& v) {
    switch (v.type()) {
    case dynamic::ARRAY:
        {
            for (auto &i : v)
                GenStat(stat, i);
            stat.arrayCount++;
            stat.elementCount += v.size();
        }
        break;

    case dynamic::OBJECT:
        {
            auto p = v.items();
            for (auto& i : p) {
                GenStat(stat, i.second);
                stat.stringLength += i.first.size();
            }
            stat.objectCount++;
            stat.memberCount += v.size();
            stat.stringCount += v.size();
        }
        break;

    case dynamic::STRING: 
        stat.stringCount++;
        stat.stringLength += v.size();
        break;

    case dynamic::INT64:
    case dynamic::DOUBLE:
        stat.numberCount++;
        break;

    case dynamic::BOOL:
        if (v.getBool())
            stat.trueCount++;
        else
            stat.falseCount++;
        break;

    case dynamic::NULLT:
        stat.nullCount++;
        break;
    }
}