class UrlExtractParameter {

  @Udf
  public String extractParam(
      @UdfParameter(description = "a valid URL") final String input,
      @UdfParameter(description = "the parameter key") final String paramToFind) {
    final String query = UrlParser.extract(input, URI::getQuery);
    if (query == null) {
      return null;
    }

    for (final String param : PARAM_SPLITTER.split(query)) {
      final List<String> kvParam = KV_SPLITTER.splitToList(param);
      if (kvParam.size() == 1 && kvParam.get(0).equals(paramToFind)) {
        return "";
      } else if (kvParam.size() == 2 && kvParam.get(0).equals(paramToFind)) {
        return kvParam.get(1);
      }
    }
    return null;
  }

}