class EmbeddedAgentConfiguration {

  @InterfaceAudience.Private
  @InterfaceStability.Unstable
  static Map<String, String> configure(String name,
      Map<String, String> properties) throws FlumeException {
    validate(name, properties);
    // we are going to modify the properties as we parse the config
    properties = new HashMap<String, String>(properties);

    if (!properties.containsKey(SOURCE_TYPE) ||
        SOURCE_TYPE_EMBEDDED_ALIAS.equalsIgnoreCase(properties.get(SOURCE_TYPE))) {
      properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED);
    }
    String sinkNames = properties.remove(SINKS);

    String strippedName = name.replaceAll("\\s+","");
    String sourceName = "source-" + strippedName;
    String channelName = "channel-" + strippedName;
    String sinkGroupName = "sink-group-" + strippedName;

    /*
     * Now we are going to process the user supplied configuration
     * and generate an agent configuration. This is only to supply
     * a simpler client api than passing in an entire agent configuration.
     */
    // user supplied config -> agent configuration
    Map<String, String> result = Maps.newHashMap();

    /*
     * First we are going to setup all the root level pointers. I.E
     * point the agent at the components, sink group at sinks, and
     * source at the channel.
     */
    // point agent at source
    result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES),
        sourceName);
    // point agent at channel
    result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS),
        channelName);
    // point agent at sinks
    result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS),
        sinkNames);
    // points the agent at the sinkgroup
    result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS),
        sinkGroupName);
    // points the sinkgroup at the sinks
    result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS,
            sinkGroupName, SINKS), sinkNames);
    // points the source at the channel
    result.put(join(name,
        BasicConfigurationConstants.CONFIG_SOURCES, sourceName,
        BasicConfigurationConstants.CONFIG_CHANNELS), channelName);

    // Properties will be modified during iteration so we need a
    // copy of the keys.
    Set<String> userProvidedKeys = new HashSet<String>(properties.keySet());

    /*
     * Second process the sink configuration and point the sinks
     * at the channel.
     */
    for (String sink :  sinkNames.split("\\s+")) {
      for (String key : userProvidedKeys) {
        String value = properties.get(key);
        if (key.startsWith(sink + SEPERATOR)) {
          properties.remove(key);
          result.put(join(name,
              BasicConfigurationConstants.CONFIG_SINKS, key), value);
        }
      }
      // point the sink at the channel
      result.put(join(name,
          BasicConfigurationConstants.CONFIG_SINKS, sink,
          BasicConfigurationConstants.CONFIG_CHANNEL), channelName);
    }
    /*
     * Third, process all remaining configuration items, prefixing them
     * correctly and then passing them on to the agent.
     */
    userProvidedKeys = new HashSet<String>(properties.keySet());
    for (String key : userProvidedKeys) {
      String value = properties.get(key);
      if (key.startsWith(SOURCE_PREFIX)) {
        // users use `source' but agent needs the actual source name
        key = key.replaceFirst(SOURCE, sourceName);
        result.put(join(name,
            BasicConfigurationConstants.CONFIG_SOURCES, key), value);
      } else if (key.startsWith(CHANNEL_PREFIX)) {
        // users use `channel' but agent needs the actual channel name
        key = key.replaceFirst(CHANNEL, channelName);
        result.put(join(name,
            BasicConfigurationConstants.CONFIG_CHANNELS, key), value);
      } else if (key.startsWith(SINK_PROCESSOR_PREFIX)) {
        // agent.sinkgroups.sinkgroup.processor.*
        result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS,
                sinkGroupName, key), value);
      } else {
        // XXX should we simply ignore this?
        throw new FlumeException("Unknown configuration " + key);
      }
    }
    return result;
  }

  private  EmbeddedAgentConfiguration();

}