spark/src/lib/config/ConfigurationState.cpp

#include "lib/os/os.h"

#include "ConfigurationState.h"
#include "ConfigurationException.h"
#include "ConfigPropertyEvaluator.h"
#include "ConfigErrorCollector.h"
#include "PreprocessInputStream.h"
#include "ConfigValueQueue.h"
#include "lib/os/os.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/is_unittest_environment.h"
#include "lib/utils/bts_array_size.h"
#include "config.h"

#include "lib/config/protos/system.pb.h"

#include <boost/algorithm/string.hpp> // iequals
#include <boost/filesystem.hpp>
#include <iostream>
#include <fstream>
#include <vector>
#include <map>
#include <tuple>

#define URDL_HEADER_ONLY 1
#define URDL_DISALE_SSL 1
#include <urdl/url.hpp>

#ifdef WIN32
#include <Winsock2.h>
#include <d2d1.h>
#else
#include <netdb.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif

#include "lib/config_proto_files.h"

using namespace std;

namespace bts {
namespace config {
namespace internal {

const char* MESSAGE_STRING_RULE = "rule";
const char* MESSAGE_STRING_VALUE = "value";
const char* MESSAGE_STRING_INCLUDE = "include";
const char* MESSAGE_STRING_INCLUDE_BACKUP = "fallback";
const char* MESSAGE_STRING_MACRO_TYPE = "MacroType";
const char* MESSAGE_STRING_MACRO = "macro";

const char* CONFIG_SERVER_ENV_VAR = "BTSConfig";

inline tuple<string,uint32_t> hostname(const std::string& path);

ConfigurationState::ConfigurationState(void) :
   _is_initialized(false),
   _initialization_mutex(),
   _dynamic_pool(),
   _prototype_factory(),
   _prototype(nullptr),
   _config_chain(),
   _defaults(),
   _properties(new ConfigPropertyEvaluator(*this)),
   _rule_engine(_properties),
   _macros(),
   _deferred_queue(new ConfigValueQueue(*_properties.get())),
   _host(),
   _port(80)
{
}

ConfigurationState::~ConfigurationState(void)
{
}

std::string ConfigurationState::defaults_path()
{
   std::string appdir = os::get_install_path();
   return appdir + "/defaults/";
}

std::string get_command_argument(const char* key)
{
   std::vector<std::string> cmd_line = os::get_command_line();
   std::vector<std::string>::iterator begin = cmd_line.begin(), end = cmd_line.end();
   for (; begin != end; begin++)
   {
      if (boost::iequals(*begin, key))
      {
         begin++;
         if (begin == end) {
            break;
         }
         else {
            os::setenv(CONFIG_SERVER_ENV_VAR, begin->c_str());
            return *begin;
         }
      }
   }
   return "";
}

string ConfigurationState::settings_path()
{
   // Look for the -ConfigServer option on the command line
   std::string cmd_arg = get_command_argument("-ConfigServer");
   if (!cmd_arg.empty()) return cmd_arg;

   // Then the BTSCONFIG environment variable
   const char* env_server = getenv(CONFIG_SERVER_ENV_VAR);
   if (env_server) return env_server;

   // then look in sparkdata
   return os::sparkdata_path("conf");
}

void ConfigurationState::initialize_defaults(IConfigLog& log)
{
   std::lock_guard<std::recursive_mutex> lock(_initialization_mutex);
   if (_is_initialized) return;

   generate_all_config_proto(log);
   
   std::shared_ptr<proto::Message> new_message(_prototype->New());
   ErrorsOnlyLog elog(log);
   load_message(defaults_path(), *new_message.get(), elog, *_deferred_queue.get());
   _defaults = new_message;

   // If defaults has an include process that now
   // FEATURE: Unlike settings loaded from ~/sparkdata/ these includes
   // take precedence over the settings in the file which includes them.
   // While this seems backwards it is always the behavior you want
   std::string include = get_include(*_defaults.get());
   if (!include.empty())
   {
      load(include, "", log);
   }

   _is_initialized = true;
}

void ConfigurationState::initialize_builtins()
{
   System sys;

   sys.set_host_name(os::host_name());
   sys.set_process_name(os::get_application_name());
   sys.set_directory(os::get_install_path());
   sys.set_platform(os::platform_name());
   sys.set_qualifier(os::get_qualifier());
   sys.set_unittest_environment(utils::internal::is_unittest_environment());
   sys.set_data_path(os::bts_data_path());
   sys.set_os_user(os::username());
#ifndef NDEBUG
   sys.set_build_type("debug");
#endif
   sys.set_config_host(_host);
   sys.set_config_port(_port);
   if (!_host.empty()) {
      hostent* host_entry = gethostbyname(_host.c_str());
      bts_assert_msg(host_entry != NULL, "Unable to resolve host name " + _host);
      in_addr* address = (in_addr*) host_entry->h_addr_list[0];
      string ip_address = inet_ntoa(*address);
      sys.set_config_ip(ip_address);
   }
#ifdef _WIN32
   sys.set_is_remote_app(GetSystemMetrics(SM_REMOTESESSION) == 1);

   // Screen dpi, Assumes screen dpi is uniform in x and y
   // Computes 2 different ratios depending - one will be right depending on whether the process calls SetDPIAware()
   HDC screen = GetDC(0);
   int log_pixels_x = GetDeviceCaps(screen, LOGPIXELSX);
   auto ratio2 = log_pixels_x / 96.0;
   int virtual_width = GetDeviceCaps(screen, HORZRES);
   int physical_width = GetDeviceCaps(screen, DESKTOPHORZRES);
   double ratio = (double)physical_width / (double)virtual_width;
   sys.set_dpi(log_pixels_x * ratio);
   sys.set_dpi_ratio(max(ratio, ratio2));
   ReleaseDC(0, screen);

#endif
   
   std::vector<std::string> args = os::get_command_line();
   for (std::vector<std::string>::iterator iter = args.begin(); iter != args.end(); iter++)
   {
      sys.add_cmd_args(*iter);
   }

   // Version string
   std::string app_version_file = os::get_install_path() + "/spark.version";
   if (boost::filesystem::exists(app_version_file))
   {
      std::ifstream input(app_version_file.c_str(), std::ifstream::in);
      if (!input.fail())
      {
         // Read the last line of the file
         std::string version;
         while(!input.eof())
         {
            std::string temp;
            std::getline(input, temp);
            if (!temp.empty()) version = temp;
         }
         
         sys.set_version(version);
         input.close();
      }
   }

   sys.set_num_cores(os::num_cores());

   //
   // Add the System value to the defaults
   const proto::FieldDescriptor* field_desc = _defaults->GetDescriptor()->FindFieldByName("System");
   const proto::Descriptor* message_desc = field_desc->message_type();
   
   // Create a meta message
   proto::Message* new_message = _defaults->GetReflection()->AddMessage(_defaults.get(), field_desc);
   const proto::FieldDescriptor* value_desc = message_desc->FindFieldByName("value");
   new_message->GetReflection()->MutableMessage(new_message, value_desc)->CopyFrom(sys);

   initialize_introspection();
}

void ConfigurationState::initialize_introspection() {
   // With this config can be self describing enough
   // that another language can request "all" of config
   ConfigIntrospection ci;
   proto::DescriptorPool::generated_pool();
   for (size_t x = 0; x < BTS_ARRAYSIZE(config_proto_files); x++) {
      std::string proto_path = "core/bts_core/config/"; 
      proto_path += config_proto_files[x];
      const proto::FileDescriptor* file_desc = proto::DescriptorPool::generated_pool()->FindFileByName(proto_path);
      if (file_desc != nullptr) {
         ConfigIntrospection::ConfigProto* config_proto = ci.add_config();
         bts_assert(config_proto);
         config_proto->set_name(file_desc->name());
         for (int y=0; y < file_desc->message_type_count(); y++) {
            const proto::Descriptor* message_desc = file_desc->message_type(y);
            if (message_desc != nullptr) {
               config_proto->add_type(message_desc->full_name());
            }
         }
      }
   }


   // Add to defaults
   const proto::FieldDescriptor* field_desc = _defaults->GetDescriptor()->FindFieldByName("ConfigIntrospection");
   const proto::Descriptor* message_desc = field_desc->message_type();
   proto::Message* new_message = _defaults->GetReflection()->AddMessage(_defaults.get(), field_desc);
   const proto::FieldDescriptor* value_desc = message_desc->FindFieldByName("value");
   new_message->GetReflection()->MutableMessage(new_message, value_desc)->CopyFrom(ci);
}

void ConfigurationState::initialize_macros()
{
   std::vector<std::shared_ptr<proto::Message> > :: iterator iter;
   for (iter = _config_chain.begin(); iter != _config_chain.end(); ++iter)
   {
      const proto::Message* message = (*iter).get();
      add_macro(message);
   }
   add_macro(_defaults.get());
}

void ConfigurationState::add_macro(const proto::Message* message)
{
   bts_assert(message);
   const proto::FieldDescriptor* field_desc = message->GetDescriptor()->FindFieldByName(MESSAGE_STRING_MACRO);
   if (field_desc)
   {
      int size = message->GetReflection()->FieldSize(*message, field_desc);
      for (int index = 0; index < size; index++)
      {
         // read "name" and "expression" fields
         const proto::Message& meta_message = message->GetReflection()->GetRepeatedMessage(*message, field_desc, index);
         const proto::FieldDescriptor* name_desc = meta_message.GetDescriptor()->FindFieldByName("name");
         const std::string name = '$' + meta_message.GetReflection()->GetString(meta_message, name_desc);

         const proto::FieldDescriptor* exp_desc = meta_message.GetDescriptor()->FindFieldByName("expression");
         const std::string expression = meta_message.GetReflection()->GetString(meta_message, exp_desc);
         
         if (_macros.find(name) == _macros.end())
         {
            // eval expression to Value
            //expression::Value result = _rule_engine.eval_to_value(expression);
            _macros[name] = expression;
         }
      }
   }
}

expression::Value ConfigurationState::get_macro(const std::string& name)
{
   map<string, string>::iterator iter = _macros.find(name);
   if (iter == _macros.end()) {
      std::string error = std::string("Macro '") + name + "' is undefined";
      throw ConfigurationException(error);
   }
   return _rule_engine.eval_to_value(iter->second);
}

std::string ConfigurationState::get_macro_string(const std::string& name) {
   try {
      expression::Value result = get_macro(name);
      return expression::value_to_string(result);
   }
   catch (const ConfigurationException& e) { return e.what(); }
}

void check_rule(proto::Message* message, expression::ExpressionEvaluator& rule_engine) {
   const proto::Reflection* reflection = message->GetReflection();
   const proto::Descriptor* descriptor = message->GetDescriptor();
   for (int findex = 0; findex < descriptor->field_count(); ++findex) {
      const proto::FieldDescriptor* field = descriptor->field(findex);
      if (field->is_repeated()) {
         int field_size = reflection->FieldSize(*message, field);
         for (int index = 0; index < field_size; ++index) {
            const proto::Message& field_message = reflection->GetRepeatedMessage(*message, field, index);
            const proto::FieldDescriptor* rule_descriptor = field_message.GetDescriptor()->FindFieldByName(MESSAGE_STRING_RULE);
            if (rule_descriptor) { // Macros don't have rules
               const std::string rule = field_message.GetReflection()->GetString(field_message, rule_descriptor);
               rule_engine.eval(rule);
            }
         }
      }
   }
}

void ConfigurationState::check_rules() {
   initialize();
   std::vector<std::shared_ptr<proto::Message> > :: iterator iter;
   for (iter = _config_chain.begin(); iter != _config_chain.end(); ++iter)
   {
      proto::Message* message = (*iter).get();
      check_rule(message, _rule_engine);
   }
   check_rule(_defaults.get(), _rule_engine);
}

void ConfigurationState::initialize_settings()
{
   std::lock_guard<std::recursive_mutex> lock(_initialization_mutex);
   if (_is_initialized) return;

   ConfigLog log;
   initialize_defaults(log);

   std::string settings = settings_path();
   std::string fallback = get_command_argument("-ConfigFallback");
   load(settings, fallback, log);

   initialize_builtins();
   initialize_macros();
   _deferred_queue->process();
}

void ConfigurationState::reset()
{
   std::lock_guard<std::recursive_mutex> lock(_initialization_mutex);

   _config_chain.clear();
   _prototype = nullptr;
   _defaults.reset();
   _prototype_factory.reset();
   _dynamic_pool.reset();
   _macros.clear();
   _deferred_queue->clear();
   _is_initialized = false;
   _host.clear();
   _port = 80;
}

int ConfigurationState::load(const std::string& url, const std::string& fallback_url, IConfigLog& log)
{
   std::shared_ptr<proto::Message> new_message(_prototype->New());

   bool loaded = false;
   try {
      loaded = load_message(url, *new_message.get(), log, *_deferred_queue.get());
   }
   catch(ServerUnreachableException&) {
      if (fallback_url.empty()) {
         throw;
      }
   }

   if (!loaded && !fallback_url.empty())
   {
      std::string message = "Failed to load: " + url + " using fallback.";
      log.warning(message);
      loaded = load_message(fallback_url, *new_message.get(), log, *_deferred_queue.get());
      if (_host.empty() || _host == "localhost") { tie(_host, _port) = hostname(fallback_url); }
   }
   if (_host.empty() || _host == "localhost") { tie(_host, _port) = hostname(url); }
   if (!loaded) return 0;


   std::string include = get_include(*new_message.get());
   std::string fallback = get_fallback(*new_message.get());

   // insert at front?
   _config_chain.push_back(new_message);

   // Recursively load includes
   if (!include.empty())
   {
      return 1 + load(include, fallback, log);
   }
   return 1;
}

int ConfigurationState::load_string(const std::string& values, bool print_errors)
{
   ConfigLog log;
   log.set_print_errors(print_errors);
   initialize_defaults(log);

   std::shared_ptr<proto::Message> new_message(_prototype->New());

   proto::io::ArrayInputStream input_stream(values.c_str(), (int)values.length());
   parse_stream("test string", input_stream, *new_message.get(), log, *_deferred_queue.get());

   std::string include = get_include(*new_message.get());
   std::string fallback = get_fallback(*new_message.get());

   _config_chain.push_back(new_message);

   int count = 0;
   if (!include.empty())
   {
      count += load(include, fallback, log);
   }

   if (_host.empty()) {
      _host = "localhost";
   }

   initialize_builtins();
   initialize_macros();
   _deferred_queue->process();
   return 1 + count;
}

void ConfigurationState::add_metadata_field(proto::FileDescriptorProto* file_dp, proto::DescriptorProto* message_dp, const proto::Descriptor* base_field, int index)
{
   // Create the metadata message
   proto::DescriptorProto* meta_message = file_dp->add_message_type();
   std::string meta_name = base_field->name() + "meta";
   meta_message->set_name(meta_name);
   
   // Add 'rule'
   proto::FieldDescriptorProto* rule_field = meta_message->add_field();
   rule_field->set_name(MESSAGE_STRING_RULE);
   rule_field->set_number(1);
   rule_field->set_type(proto::FieldDescriptorProto_Type_TYPE_STRING);
   rule_field->set_label(proto::FieldDescriptorProto_Label_LABEL_OPTIONAL);
   rule_field->set_default_value("true");

   // Add 'value'
   proto::FieldDescriptorProto* value_field = meta_message->add_field();
   value_field->set_name(MESSAGE_STRING_VALUE);
   value_field->set_number(2);
   value_field->set_type_name(base_field->full_name());
   value_field->set_label(proto::FieldDescriptorProto_Label_LABEL_REQUIRED);
   
   // Now add a field to message_dp of that type
   proto::FieldDescriptorProto* field_desc = message_dp->add_field();
   field_desc->set_name(base_field->name());
   field_desc->set_number(index);
   field_desc->set_type_name(meta_name);
   field_desc->set_label(proto::FieldDescriptorProto_Label_LABEL_REPEATED);
}

// Construct a new Message that describes the save file format
// This can then be used with google::protobuf::TextFormat to load and save files
void ConfigurationState::generate_all_config_proto(IConfigLog& log)
{
   const char* bts_config_classname = "__BTSConfig";
   int field_number = 0;

   proto::FileDescriptorProto bts_config_dp;
   bts_config_dp.set_name("config.proto");
   proto::DescriptorProto* config_message = bts_config_dp.add_message_type();
   config_message->set_name(bts_config_classname);

   // Add the 'include' field
   proto::FieldDescriptorProto* include_field = config_message->add_field();
   include_field->set_name(MESSAGE_STRING_INCLUDE);
   include_field->set_number(++field_number);
   include_field->set_type(proto::FieldDescriptorProto_Type_TYPE_STRING);
   include_field->set_label(proto::FieldDescriptorProto_Label_LABEL_OPTIONAL);

   // Add the 'fallback' field
   proto::FieldDescriptorProto* backup_field = config_message->add_field();
   backup_field->set_name(MESSAGE_STRING_INCLUDE_BACKUP);
   backup_field->set_number(++field_number);
   backup_field->set_type(proto::FieldDescriptorProto_Type_TYPE_STRING);
   backup_field->set_label(proto::FieldDescriptorProto_Label_LABEL_OPTIONAL);

   // Add 'macro' message type
   proto::DescriptorProto* macro_message = bts_config_dp.add_message_type();
   macro_message->set_name(MESSAGE_STRING_MACRO_TYPE);
   proto::FieldDescriptorProto* name_field = macro_message->add_field();
   name_field->set_name("name");
   name_field->set_number(1);
   name_field->set_type(proto::FieldDescriptorProto_Type_TYPE_STRING);
   name_field->set_label(proto::FieldDescriptorProto_Label_LABEL_REQUIRED);
   proto::FieldDescriptorProto* expression_field = macro_message->add_field();
   expression_field->set_name("expression");
   expression_field->set_number(2);
   expression_field->set_type(proto::FieldDescriptorProto_Type_TYPE_STRING);
   expression_field->set_label(proto::FieldDescriptorProto_Label_LABEL_REQUIRED);

   // add 'macro' message field
   proto::FieldDescriptorProto* macro_field = config_message->add_field();
   macro_field->set_name(MESSAGE_STRING_MACRO);
   macro_field->set_number(++field_number);
   macro_field->set_type(proto::FieldDescriptorProto_Type_TYPE_MESSAGE);
   macro_field->set_type_name(MESSAGE_STRING_MACRO_TYPE);
   macro_field->set_label(proto::FieldDescriptorProto_Label_LABEL_REPEATED);

   // add a meta type and field for each generated message
   proto::DescriptorPool::generated_pool();
   for (size_t x = 0; x < BTS_ARRAYSIZE(config_proto_files); x++)
   {
      std::string proto_path = "lib/config/protos/"; 
      proto_path += config_proto_files[x];
      const proto::FileDescriptor* file_desc = proto::DescriptorPool::generated_pool()->FindFileByName(proto_path);
      if (file_desc != nullptr)
      {
         bts_config_dp.add_dependency(file_desc->name());
         for (int y=0; y < file_desc->message_type_count(); y++)
         {
            const proto::Descriptor* message_desc = file_desc->message_type(y);
            add_metadata_field(&bts_config_dp, config_message, message_desc, ++field_number);
         }
      }
      else
      {
         #ifdef WIN32
         // on linux the linker might throw away config protos that
         // are not used in sparkd. On windows we build a dll so that
         // should not happen
         throw InvalidDescriptorException(config_proto_files[x]);
         #endif
      }
   }
   
   // Generate a proper FileDesctiptor from the FileDescriptorProto
   // The newly created descriptors live in the pool
   _dynamic_pool.reset(new proto::DescriptorPool(proto::DescriptorPool::generated_pool()));
   ConfigErrorCollector errors(log, "");
   const proto::FileDescriptor* file_desc = _dynamic_pool->BuildFileCollectingErrors(bts_config_dp, &errors);
   if (!file_desc) {
      std::string error_message = errors.ToString();
      if (!error_message.empty()) {
         bts_assert_msg(file_desc, error_message);
      } else {
         bts_assert_msg(false, "Config: error building parser message.  No errors logged by protobuf library.\n This could mean you have defined 2 config keys with the same name.  Unfortunately that's all I know.");
      }
   }
   const proto::Descriptor* bts_config_desc = file_desc->FindMessageTypeByName(bts_config_classname);
   
   _prototype_factory.reset(new proto::DynamicMessageFactory());
   _prototype_factory->SetDelegateToGeneratedFactory(true);
   _prototype = _prototype_factory->GetPrototype(bts_config_desc);
}

bool is_valid_key_descriptor(const proto::Descriptor* property_desc)
{
   std::string file_name = property_desc->file()->name();
   size_t file_name_start = file_name.find_last_of('/') + 1;
   file_name = file_name.substr(file_name_start);
   for (size_t x = 0; x < BTS_ARRAYSIZE(config_proto_files); x++)
   {
      std::string proto_name = config_proto_files[x];
      if (boost::iequals(file_name, proto_name))
      {
         return true;
      }
   }
   return false;
}

const proto::Message* ConfigurationState::try_get_message(const std::string& property_name)
{
   initialize_settings();

   const proto::Message* message;

   // Iterate through the loaded files

   std::vector<std::shared_ptr<proto::Message> > :: iterator iter;
   for (iter = _config_chain.begin(); iter != _config_chain.end(); ++iter)
   {
      message = try_get_value(property_name, **iter);
      if (message)
      {
         return message;
      }
   }

   // Fall back to defaults
   message = try_get_value(property_name, *_defaults.get());
   if (message)
   {
      return message;
   }

   return nullptr;
}

const proto::Message& ConfigurationState::get_message(const proto::Descriptor* property_desc)
{
   if (!is_valid_key_descriptor(property_desc))
   {
      throw InvalidKeyException(property_desc->name().c_str());
   }
   
   const proto::Message* message = try_get_message(property_desc->name());
   if (!message)
   {
      // Get the message prototype - will only have compiled in defaults
      message = proto::MessageFactory::generated_factory()->GetPrototype(property_desc);
   }

   return *message;
}

const proto::Message* ConfigurationState::try_get_value(const std::string& field_name, const proto::Message& message) const
{
   // 1) Get the metadata descriptor
   const proto::Descriptor* message_desc = message.GetDescriptor();
   const proto::FieldDescriptor* meta_desc = message_desc->FindFieldByName(field_name);
   if (meta_desc)
   {
      // 2) Get the repeated metadata
      int meta_size = message.GetReflection()->FieldSize(message, meta_desc);
      for (int index = 0; index < meta_size; index++)
      {
         // 3) Iterate over repeated field and check rule
         const proto::Message& meta_message = message.GetReflection()->GetRepeatedMessage(message, meta_desc, index);
         const proto::FieldDescriptor* rule_desc = meta_message.GetDescriptor()->FindFieldByName(MESSAGE_STRING_RULE);
         const std::string rule = meta_message.GetReflection()->GetString(meta_message, rule_desc);

         if (_rule_engine.eval(rule))
         {
             // 4) return value
             const proto::FieldDescriptor* value_desc = meta_message.GetDescriptor()->FindFieldByName(MESSAGE_STRING_VALUE);
             const proto::Message& message_value = meta_message.GetReflection()->GetMessage(meta_message, value_desc, proto::MessageFactory::generated_factory());
             return &message_value;
         }
         else
         {
            // TODO: Consider a rule result cache
            // pro: faster lookups after first parse
            // con: right now config is read only and perfectly parallel after load
         }
      }
   }
   return nullptr;
}

const google::protobuf::Descriptor* ConfigurationState::get_descriptor_by_name(const std::string& name) {
   const proto::Message* pmsg = _defaults.get();
   if (pmsg) {
      // Get the metadata descriptor (Meta = rule+value fields)
      const proto::Descriptor* desc = pmsg->GetDescriptor();
      const proto::FieldDescriptor* meta_desc = desc->FindFieldByName(name);
      if (meta_desc) {
         // Find the "value" field descriptor
         const proto::Descriptor* pmsg_desc = meta_desc->message_type();
         const proto::FieldDescriptor* value_desc = pmsg_desc->FindFieldByName(MESSAGE_STRING_VALUE);
         if (value_desc) {
            // return it's type
            return value_desc->message_type();
         }
      }
   }
   return nullptr;
}

PathType::type get_path_type(std::string path)
{
   if (path.find("://") != std::string::npos)
   {
      return PathType::directory;
      //const char last_char = path.at(path.length()-1);
      //return last_char == '/' ? PathType::directory : PathType::file;
   }
   else
   {
      if (boost::filesystem::is_directory(path))
      {
         return PathType::directory;
      }
      else if (boost::filesystem::exists(path))
      {
         return PathType::file;
      }
      else
      {
         return PathType::not_found;
      }
   }
}

urdl::url url_from_string(std::string& path, bool required)
{
   boost::system::error_code ec;
   // attempt to parse the path as a URL - if that fails assume it is an absolute file path and append file:///
   urdl::url url = urdl::url::from_string(path, ec);
   if (ec)
   {
      std::string file_prefix = "file:///";
      std::string abs_path = boost::filesystem::absolute(path).string();
      url = urdl::url::from_string(file_prefix + os::normalize_path(abs_path), ec);
   }
   bts_assert_msg(!required || !ec, ec.message().c_str());

   return url;
}

inline tuple<string,uint32_t> hostname(const std::string& path) {
   boost::system::error_code ec;
   string host;
   uint32_t port = 80;
   // Try to parse a url with urdl
   urdl::url url = urdl::url::from_string(path, ec);
   if (!ec) {
      host = url.host();
      port = url.port();
   } else {
      // Check for a UNC path
      string norm_path = path;
      os::normalize_path(norm_path);
      if (norm_path[0] == '/' && norm_path[1] == '/') {
         size_t host_len = norm_path.find_first_of('/', 2) - 2;
         if (host_len > 0) {
            host = norm_path.substr(2, host_len);
         }
      }
   }

   // If we are still empty return "localhost"
   if (host.empty()) host = "localhost";

   return std::make_tuple(host, port);
}

// Loads a single message file or directory
bool load_message(std::string url, proto::Message& out_message, IConfigLog& log, ConfigValueQueue& lookup)
{
   ConfigDownloader downloader;

   bool result = false;
   switch(get_path_type(url))
   {
   case PathType::directory:
      result = load_directory(url, downloader, out_message, log, lookup) > 0;
      break;
   case PathType::file:
      result = load_file(url, downloader, out_message, log, lookup) == ConfigDownloader::SUCCESS;
      break;
   default:
      std::ostringstream error;
      error << "Error '" << url << "' is not a valid configuration path";
      std::string err_str = error.str();
      log.error(err_str);
      throw ConfigurationException(err_str.c_str());
      break;
   }

   return result;
}

int load_directory(std::string url, ConfigDownloader& downloader, proto::Message& out_message, IConfigLog& log, ConfigValueQueue& lookup)
{
   LogAggregator aggregate_log(log);
   aggregate_log.info("URL: " + url);

   const char last_char = url.at(url.length()-1);
   if (last_char != '/') {
      url = url + '/';
   }
   
   // If a file called "all" exists in the config directory we can short circuit ~60 http requests
   ConfigDownloader::Result all_result = load_file(url + ".all", downloader, out_message, aggregate_log, lookup, false);
   if (all_result == ConfigDownloader::SUCCESS) {
      log.info(aggregate_log.text());
      return 1;
   }
   
   // load spark.conf
   int load_count = 0;
   ConfigDownloader::Result result = load_file(url + "spark.conf", downloader, out_message, aggregate_log, lookup, false);
   if (result == ConfigDownloader::SUCCESS)
   {
      load_count++;
   } 
   else if (result == ConfigDownloader::PERMANENT_FAILURE) { 
      // If the server is down it doesn't make sense to suffer though the timeouts for the other files
      throw ServerUnreachableException(url.c_str());
   }

   bool edge_has_include = !get_include(out_message).empty();

   for(const char* proto_file: config_proto_files)
   {
      std::string config_file(proto_file);
      if (config_file == "system.proto") continue;
      if (config_file == "edge.proto") continue;

      const char proto_ext[] = ".proto";
      const char conf_ext[] = ".conf";
      size_t extension_pos = config_file.rfind(proto_ext);
      // path_to_config_proto/foo.proto => foo.conf
      config_file.replace(extension_pos, sizeof(proto_ext), conf_ext, sizeof(conf_ext));
      boost::filesystem::path p (config_file);
      config_file = p.filename ().string ();
      if (load_file(url + config_file, downloader, out_message, aggregate_log, lookup, false))
      {
         load_count++;
         if (!edge_has_include) {
            bts_assert_msg(get_include(out_message).empty(), "Includes may only be specified in spark.conf");
         }

         // NOTE: This code allows any config key to go in any .conf file
         // If this becomse a problem we could iterate over the loaded messages
         // and ensure they came from the correct location.
      }
   }

   log.info(aggregate_log.text());

   return load_count;
}

ConfigDownloader::Result load_file(std::string path, ConfigDownloader& downloader, proto::Message& out_message, IConfigLog& log, ConfigValueQueue& lookup, bool required)
{
   stringstream url_stream;
   string url = url_from_string(path, required).to_string();
   ConfigDownloader::Result result = downloader.get(url, url_stream);
   if (!(result == ConfigDownloader::SUCCESS)) {
      bts_assert_msg(!required || url_stream, (path + " not found.").c_str());
      return result;
   }

   log.info("Loaded: " + path);

   proto::io::IstreamInputStream proto_stream(&url_stream);
   parse_stream(path, proto_stream, out_message, log, lookup);

   return result;
}

void parse_stream(const std::string& file, proto::io::ZeroCopyInputStream& input_stream, proto::Message& out_message, IConfigLog& log, ConfigValueQueue& lookup)
{
   ConfigErrorCollector error_collector(log, file.c_str());
   bts::google::protobuf::TextFormat::Parser parser;
   parser.RecordErrorsTo(&error_collector);
   parser.SetValueLookup(&lookup);
   lookup.set_filename(file);
   parser.AllowPartialMessage(true);
   bool success = parser.Parse(&input_stream, &out_message);
   if (!success)
   {
      std::string error_message = error_collector.ToString();
      log.error(error_message);
      throw ConfigurationException(error_message);
   }
}

std::string get_field(const proto::Message& message, const char* name)
{
   const proto::Descriptor* message_descriptor = message.GetDescriptor();
   const proto::FieldDescriptor* include_field = message_descriptor->FindFieldByName(name);
   const std::string include = message.GetReflection()->GetString(message, include_field);

   return include;
}

std::string get_include(const proto::Message& message)
{
   return get_field(message, MESSAGE_STRING_INCLUDE);
}

std::string get_fallback(const proto::Message& message)
{
   return get_field(message, MESSAGE_STRING_INCLUDE_BACKUP);
}

}}}//end namespaces