public class FITSAdaptor extends AbstractAdaptor {

  /**
   * A default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( FITSAdaptor.class );

  /**
   * The apache digester to process the fits xml meta data.
   */
  private Digester digester;

  /**
   * A default constructor that initialises the digester and sets up the parsing
   * rules.
   */
  public FITSAdaptor() {
    this.digester = new Digester(); // not thread safe
    this.digester.setRules( new RegexRules( new SimpleRegexMatcher() ) );
    this.createParsingRules();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void configure() {
    // nothing to do...
  }

  /**
   * Parses the meta data and retrieves it. This method makes use of a
   * {@link DigesterContext} object that is pushed upon the digester stack. This
   * helper object has some methods for handling some special cases.
   */
  @Override
  public Element parseElement( String name, String data ) {
    Element element = null;

    if ( data == null ) {
      return element;

    }
    try {

      DigesterContext context = new DigesterContext( this.getCache(), this.getPreProcessingRules() );
      this.digester.push( context );

      context = (DigesterContext) this.digester.parse( new StringReader( data ) );
      element = context.getElement();
      List<MetadataRecord> values = context.getValues();

      if ( element != null ) {
        element.setMetadata( values );
      }

    } catch ( IOException e ) {
      LOG.warn( "An exception occurred while processing {}: {}", name, e.getMessage() );
    } catch ( SAXException e ) {
      LOG.warn( "An exception occurred while parsing {}: {}", name, e.getMessage() );
    } catch ( Exception e ) {
      LOG.warn( "An exception occurred while parsing {}: {}", name, e.getMessage() );
    }

    return element;
  }

  /**
   * Returns the prefix of this adaptor ('fits').
   */
  @Override
  public String getAdaptorPrefix() {
    return "fits";
  }

  /**
   * Creates SAX based rules.
   */
  private void createParsingRules() {
    this.createElementRules();
    this.createIdentityRules();
    this.createFileInfoRules();
    this.createRepresentationInfoRules();
    this.createFileStatusRules();
    this.createMetaDataRules();
  }

  /**
   * Creates rules for the creation of the Element object.
   */
  private void createElementRules() {
    this.digester.addCallMethod( "fits", "createElement", 2 );
    this.digester.addCallParam( "fits/fileinfo/filename", 0 );
    this.digester.addCallParam( "fits/fileinfo/filepath", 1 );
  }

  /**
   * Creates rules for the parsing of the identification data within a FITS
   * file.
   */
  private void createIdentityRules() {
    this.createIdentityStatusRules();

    this.createFormatRule( "fits/identification/identity" );
    this.createFormatVersionRule( "fits/identification/identity/version" );
    this.createPuidRule( "fits/identification/identity/externalIdentifier" );

  }

  /**
   * Creates rules for parsing the identity status data within a FITS file.
   */
  private void createIdentityStatusRules() {
    this.digester.addCallMethod( "fits/identification", "setIdentityStatus", 1 );
    this.digester.addCallParam( "fits/identification", 0, "status" );
  }

  /**
   * Creates rules for parsing the format data within a FITS file.
   * 
   * @param pattern
   *          the xpath to the format identity.
   */
  private void createFormatRule( String pattern ) {
    this.digester.addCallMethod( pattern, "createIdentity", 2 );
    this.digester.addCallParam( pattern, 0, "format" );
    this.digester.addCallParam( pattern, 1, "mimetype" );

    this.digester.addCallMethod( pattern + "/tool", "addIdentityTool", 2 );
    this.digester.addCallParam( pattern + "/tool", 0, "toolname" );
    this.digester.addCallParam( pattern + "/tool", 1, "toolversion" );

  }

  /**
   * Creates rules for parsing the format version data within a FITS file.
   * 
   * @param pattern
   *          the xpath to the format version identity.
   */
  private void createFormatVersionRule( String pattern ) {
    this.digester.addCallMethod( pattern, "createFormatVersion", 4 );
    this.digester.addCallParam( pattern, 0 );
    this.digester.addCallParam( pattern, 1, "status" );
    this.digester.addCallParam( pattern, 2, "toolname" );
    this.digester.addCallParam( pattern, 3, "toolversion" );
  }

  /**
   * Creates rules for parsing the pronom identifier data within a FITS file.
   * 
   * @param pattern
   *          the xpath to the external identifier.
   */
  private void createPuidRule( String pattern ) {
    this.digester.addCallMethod( pattern, "createPuid", 3 );
    this.digester.addCallParam( pattern, 0 );
    this.digester.addCallParam( pattern, 1, "toolname" );
    this.digester.addCallParam( pattern, 2, "toolversion" );

  }

  /**
   * Creates rules for the parsing of the file information data within a FITS
   * file.
   */
  private void createFileInfoRules() {
    this.createValueRule( "fits/fileinfo/size" );
    this.createValueRule( "fits/fileinfo/md5checksum" );
    this.createValueRule( "fits/fileinfo/lastmodified" );
    this.createValueRule( "fits/fileinfo/fslastmodified" );
    this.createValueRule( "fits/fileinfo/created" );
    this.createValueRule( "fits/fileinfo/creatingApplicationName" );
    this.createValueRule( "fits/fileinfo/creatingApplicationVersion" );
    this.createValueRule( "fits/fileinfo/inhibitorType" );
    this.createValueRule( "fits/fileinfo/inhibitorTarget" );
    this.createValueRule( "fits/fileinfo/rightsBasis" );
    this.createValueRule( "fits/fileinfo/copyrightBasis" );
    this.createValueRule( "fits/fileinfo/copyrightNote" );
    this.createValueRule( "fits/fileinfo/creatingos" );
  }

  /*
   * Experimental
   */
  /**
   * This is not part of the original FITS specification, but it is reading out
   * representation information out of RODA, if the FITS was provided by RODA.
   */
  private void createRepresentationInfoRules() {
    this.createValueRule( "fits/representationinfo/original" );
  }

  /**
   * Creates rules for parsing the file status data within a FITS file.
   */
  private void createFileStatusRules() {
    this.createValueRule( "fits/filestatus/well-formed" );
    this.createValueRule( "fits/filestatus/valid" );
    this.createValueRule( "fits/filestatus/message" );
  }

  /**
   * Creates rules for parsing the meta data section of a FITS file.
   */
  private void createMetaDataRules() {
    this.createValueRule( "fits/metadata/image/*" );
    this.createValueRule( "fits/metadata/text/*" );
    this.createValueRule( "fits/metadata/document/*" );
    this.createValueRule( "fits/metadata/audio/*" );
    this.createValueRule( "fits/metadata/video/*" );
  }

  /**
   * Creates rule for parsing generic values from FITS files.
   * 
   * @param pattern
   *          the xpath to the generic meta data node.
   */
  private void createValueRule( String pattern ) {
    this.digester.addCallMethod( pattern, "createValue", 5 );
    this.digester.addCallParam( pattern, 0 );
    this.digester.addCallParam( pattern, 1, "status" );
    this.digester.addCallParam( pattern, 2, "toolname" );
    this.digester.addCallParam( pattern, 3, "toolversion" );
    this.digester.addCallParamPath( pattern, 4 );
  }

} public class AssignCollectionToElementRule implements PostProcessingRule {

  /**
   * The collection to apply.
   */
  private String collectionName;

  /**
   * Creates the rule.
   * 
   * @param name
   *          the name of the collection.
   */
  public AssignCollectionToElementRule(String name) {
    this.collectionName = name;
  }

  /**
   * Has a very high priority.
   */
  @Override
  public int getPriority() {
    return 990;
  }

  /**
   * Sets the collection of the given element to the colleciton name and returns
   * it.
   */
  @Override
  public Element process( Element e ) {
    if ( e != null ) {
      e.setCollection( this.collectionName );
    }

    return e;
  }

} public class CreateElementIdentifierRule implements PostProcessingRule {

  /**
   * This rule has a low priority.
   */
  @Override
  public int getPriority() {
    return 10;
  }

  /**
   * Sets a random uuid if the element has no uid defined.
   */
  @Override
  public Element process( Element e ) {
    if ( e != null && e.getUid() == null ) {
      e.setUid( UUID.randomUUID().toString() );
    }

    return e;
  }

} public class EmptyValueProcessingRule implements PreProcessingRule {

  /**
   * Skips the empty values.
   */
  @Override
  public boolean shouldSkip( String property, String value, String status, String tool, String version ) {
    if ( value == null || value.equals( "" ) ) {
      return true;
    }
    return false;
  }

  /**
   * Has a high priority.
   */
  @Override
  public int getPriority() {
    return 990;
  }

} public class FormatVersionResolutionRule implements PostProcessingRule {

  /**
   * Has a high priotity.
   */
  @Override
  public int getPriority() {
    return 900;
  }

  /**
   * Make sure that all format_version have the correct status. Especially when
   * there is a CONFLICT
   */
  @Override
  public Element process( Element e ) {
    List<MetadataRecord> formatVersionRecords = e.removeMetadata( "format_version" );
    if ( formatVersionRecords.size() > 1 ) {
      for ( MetadataRecord mr : formatVersionRecords ) {
        mr.setStatus( Status.CONFLICT.name() );
      }
    }

    e.getMetadata().addAll( formatVersionRecords );

    return e;
  }

} public class HtmlInfoProcessingRule implements PreProcessingRule {

  /**
   * A default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( HtmlInfoProcessingRule.class );

  /**
   * A set of valid html tags.
   */
  private Set<String> tags;

  public HtmlInfoProcessingRule() {
    this.tags = new HashSet<String>();
    this.readTags();
  }

  /**
   * Reads the valid html tags into memory.
   */
  private void readTags() {
    try {
      BufferedReader reader = new BufferedReader( new InputStreamReader( HtmlInfoProcessingRule.class.getClassLoader()
          .getResourceAsStream( "adaptors/htmltags" ) ) );
      String line = reader.readLine();
      while ( line != null ) {

        this.tags.add( line );

        line = reader.readLine();
      }
    } catch ( FileNotFoundException e ) {
      e.printStackTrace();
    } catch ( IOException e ) {
      e.printStackTrace();
    }

  }

  /**
   * Has the lowest possible priority.
   */
  @Override
  public int getPriority() {
    return 1;
  }

  /**
   * Skips the value if the property is not a valid html tag.
   */
  @Override
  public boolean shouldSkip( String property, String value, String status, String tool, String version ) {

    if ( tool != null && tool.equalsIgnoreCase( "HtmlInfo" ) ) {

      if ( property.endsWith( "rences" ) ) {
        int tagIndex = property.indexOf( "Tag" );

        if ( tagIndex == -1 ) {
          return true;
        }

        String tag = property.substring( 0, tagIndex );

        if ( !this.tags.contains( tag ) ) {
          LOG.debug( "Property {} seems to be faulty, skip", property );
          return true;
        }
      }
    }

    return false;
  }

} public class InferDateFromFileNameRule implements PostProcessingRule {

  /**
   * The cache to use.
   */
  private ReadOnlyCache cache;

  /**
   * Creates the rule.
   */
  public InferDateFromFileNameRule() {

  }

  /**
   * Creates the rule with the given cache.
   * 
   * @param cache
   */
  public InferDateFromFileNameRule(ReadOnlyCache cache) {
    this.cache = cache;
  }

  public void setReadOnlyCache( ReadOnlyCache cache ) {
    this.cache = cache;
  }

  /**
   * Has a low priority.
   */
  @Override
  public int getPriority() {
    return 10;
  }

  /**
   * Tries to extract the meta data for the created property and sets it if
   * successful, otherwise it returns the unmodified element.
   */
  @Override
  public Element process( Element e ) {
    if ( e != null ) {
      this.extractCreatedMetadataRecord( e, this.cache.getProperty( "created" ) );
    }

    return e;
  }

  /*
   * experimental only for the SB archive
   */
  /**
   * Tries to parse a creation date out of the name of the current element. Some
   * sources include a timestamp in the name.
   * 
   * If the name is not set or it does not include a timestamp, the method does
   * nothing.
   * 
   * @param created
   *          the property for the creation date.
   */
  private void extractCreatedMetadataRecord( Element e, Property created ) {
    String name = e.getName();
    if ( name != null ) {

      String[] split = name.split( "-" );

      if ( split.length > 2 ) {
        String date = split[2];

        try {
          Long.valueOf( date );

          MetadataRecord c = new MetadataRecord( created, date );
          e.getMetadata().add( c );

        } catch ( NumberFormatException nfe ) {
          // if the value is not a number then it is something else and not a
          // year, skip the inference.

        }
      }
    }
  }

} public class TIKAAdaptor extends AbstractAdaptor {

  private static final Logger LOG = LoggerFactory.getLogger( TIKAAdaptor.class );

  private static final String OPT_TIKA_VERSION = "c3po.adaptor.tika.version";

  private String version = "";

  /**
   * {@inheritDoc}
   */
  @Override
  public void configure() {
    this.version = this.getStringConfig( OPT_TIKA_VERSION, "" );
  }

  /**
   * Parses the TIKA data.
   */
  @Override
  public Element parseElement( String name, String data ) {
    try {
      Map<String, String> metadata = TIKAResultParser.getKeyValueMap( data );
      Element element = this.createElement( metadata );
      return element;
    } catch ( IOException e ) {
      LOG.warn( "Could not parse data for {}", name );
      return null;
    }
  }

  @Override
  public String getAdaptorPrefix() {
    return "tika";
  }

  /**
   * Creates an element out of the parsed metadata map.
   * 
   * @param metadata
   *          the map of property value pairs.
   * @return the element object.
   */
  private Element createElement( Map<String, String> metadata ) {

    String name = metadata.remove( "resourceName" );
    if ( name == null ) {
      return null;
    }

    Element element = new Element( name, name );
    ReadOnlyCache cache = this.getCache();
    List<MetadataRecord> records = new ArrayList<MetadataRecord>();

    for ( String key : metadata.keySet() ) {
      String value = metadata.get( key );
      key = key.replace( '.', '_' ).replace( ' ', '_' ).replace( ':', '_' ).toLowerCase();
      key = TIKAHelper.getPropertyKeyByTikaName( key );

      if ( key != null ) {
        Property prop = cache.getProperty( key );
        if ( prop.getType().equals( PropertyType.INTEGER.name() ) || prop.getType().equals( PropertyType.FLOAT.name() ) ) {
          value = value.split( " " )[0];
        }
        MetadataRecord record = new MetadataRecord( prop, value );
        Source source = cache.getSource( "Tika", this.version );
        record.setSources( Arrays.asList( source.getId() ) );
        records.add( record );
      }
    }

    element.setMetadata( records );
    return element;

  }

} public class CSVGenerator {

  /**
   * Default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( CSVGenerator.class );

  /**
   * The persistence layer.
   */
  private PersistenceLayer persistence;

  /**
   * Creates the generator with the given persistence layer.
   * 
   * @param p
   */
  public CSVGenerator(PersistenceLayer p) {
    this.persistence = p;
  }

  /**
   * Exports all the data from the given collection to a sparse matrix view
   * where each column is a property and each row is an element with the values
   * for the corresponding property.
   * 
   * @param collection
   *          the collection to export
   * @param output
   *          the output file
   */
  public void exportAll( String collection, String output ) {
    final Iterator<Element> matrix = this.buildMatrix( collection );
    final Iterator<Property> allprops = this.persistence.find( Property.class, null );
    final List<Property> props = this.getProperties( allprops );

    this.write( matrix, props, output );
  }

  /**
   * Exports all the given properties for the given mimetype to a sparse matrix
   * view where each column is a property and each row is an element with the
   * values for the corresponding property.
   * 
   * @param collection
   *          the collection to export
   * @param props
   *          the properties filter
   * @param output
   *          the output file.
   */
  public void export( final String collection, final List<Property> props, String output ) {
    final Iterator<Element> matrix = this.buildMatrix( collection, props );

    this.write( matrix, props, output );
  }

  /**
   * Exports all the data matching the given filter to the given output file.
   * 
   * @param filter
   *          the filter to match.
   * @param output
   *          the output file.
   */
  public void export( final Filter filter, String output ) {
    final Iterator<Property> allprops = this.persistence.find( Property.class, null );
    final List<Property> props = this.getProperties( allprops );

    this.export( filter, props, output );
  }

  /**
   * Exports all the given properties and data that matches the given filter to
   * the given output file.
   * 
   * @param filter
   *          the filter to apply.
   * @param props
   *          the properties to generate.
   * @param output
   *          the output file.
   */
  public void export( final Filter filter, final List<Property> props, String output ) {
    Iterator<Element> matrix = this.persistence.find( Element.class, filter );

    this.write( matrix, props, output );
  }

  /**
   * Writes a file containing the data of all elements in the matrix iterator
   * over all properties to the given output file.
   * 
   * @param matrix
   *          the elements that the csv file should contain
   * @param props
   *          the properties that the csv file should contain
   * @param output
   *          the output file where to write to.
   */
  private void write( Iterator<Element> matrix, List<Property> props, String output ) {
    try {

      final File file = new File( output );

      LOG.info( "Will export data in {}", file.getAbsolutePath() );

      if ( file.getParentFile() != null && !file.getParentFile().exists() ) {
        file.getParentFile().mkdirs();
      }

      file.createNewFile();

      final FileWriter writer = new FileWriter( file );

      // build header of csv
      writer.append( "uid, " );
      for ( Property p : props ) {
        writer.append( p.getKey() + ", " );
      }
      writer.append( "\n" );

      // for all elements append the values in the correct column
      while ( matrix.hasNext() ) {
        Element next = matrix.next();

        // first the uid
        writer.append( replace( next.getUid() ) + ", " );

        // then the properties
        for ( Property p : props ) {
          List<MetadataRecord> value = next.removeMetadata( p.getId() );

          assert value.size() == 0 || value.size() == 1;

          final String val = this.getValueFromMetaDataRecord( value );
          writer.append( val );
          writer.append( ", " );
        }
        writer.append( "\n" );
      }

      writer.flush();
      writer.close();

    } catch ( IOException e ) {
      e.printStackTrace();
    }
  }

  /**
   * Queries the db and obtains all elements for the given collection.
   * 
   * @param collection
   *          the name of the colleciton.
   * @return the matching elements.
   */
  private Iterator<Element> buildMatrix( final String collection ) {
    Filter filter = new Filter( new FilterCondition( "collection", collection ) );

    return this.persistence.find( Element.class, filter );
  }

  /**
   * Queries the db and obtains all elements for the given collection provided
   * they have a value for all properties in the list.
   * 
   * @param collection
   *          the name of the collection.
   * @param props
   *          the properties to look for.
   * @return the matching elements.
   */
  private Iterator<Element> buildMatrix( final String collection, List<Property> props ) {
    Filter filter = new Filter( new FilterCondition( "collection", collection ) );

    for ( Property p : props ) {
      filter.addFilterCondition( new FilterCondition( p.getId(), null ) );
    }

    return this.persistence.find( Element.class, filter );
  }

  /**
   * Gets the value for the given list of meta data rocords or 'CONFLICT' if
   * there are more values.
   * 
   * @param value
   *          the list of records.
   * @return the output string for the csv cell.
   */
  private String getValueFromMetaDataRecord( List<MetadataRecord> value ) {
    String result = "";
    if ( value.size() != 0 ) {
      final String v = value.get( 0 ).getValue();
      result = (v == null) ? "CONFLICT" : replace( v.toString() );
    }

    return result;
  }

  /**
   * Extracts {@link Property} objects from the given cursor and only sets the
   * id and the name field.
   * 
   * @param cursor
   *          the iterator over the properties
   * @return a list of properties or an empty list.
   */
  private List<Property> getProperties( final Iterator<Property> cursor ) {
    final List<Property> result = new ArrayList<Property>();

    while ( cursor.hasNext() ) {
      final Property next = cursor.next();
      result.add( next );
    }

    return result;
  }

  /**
   * replaces all comma ocurrences in the values with an empty string.
   * 
   * @param str
   *          the string to check
   * @return a new altered string or an empty string if the input was null.
   */
  private String replace( String str ) {
    return (str == null) ? "" : str.replaceAll( ",", "" ).replaceAll( "\n", " " );
  }

} public class ProfileGenerator {

  /**
   * Default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( ProfileGenerator.class );

  /**
   * The {@link com.petpet.c3po.api.model.Element} class.
   */
  private static final Class<com.petpet.c3po.api.model.Element> ELEMENT_CLASS = com.petpet.c3po.api.model.Element.class;

  /**
   * A set of properties that will be expanded with histograms.
   */
  private static final String[] PROPERTIES = { "format", "format_version", "puid", "mimetype", "charset", "linebreak",
      "compressionscheme", "creating_os", "byteorder", "compression_scheme", "colorspace", "icc_profile_name",
      "icc_profile_version", "created", "creating.application.name" };

  /**
   * The persistence layer.
   */
  private PersistenceLayer persistence;

  /**
   * The representative samples generator.
   */
  private final RepresentativeGenerator sampleSelector;

  /**
   * Creates this generator.
   * 
   * @param persistence
   *          the persistence to use.
   * @param generator
   *          the samples generator to use.
   */
  public ProfileGenerator(final PersistenceLayer persistence, RepresentativeGenerator generator) {
    this.persistence = persistence;
    this.sampleSelector = generator;
  }

  /**
   * writes the xml string to a local folder 'profiles' in a file called
   * output.xml.
   * 
   * @param xml
   *          the xml to write.
   */
  public void write( final String xml ) {
    try {
      final Document doc = DocumentHelper.parseText( xml );
      this.write( doc );

    } catch ( final DocumentException e ) {
      LOG.warn( "An error occurred: {}", e.getMessage() );
    }
  }

  /**
   * Writes the xml document to a local 'profiles' folder.
   * 
   * @param doc
   *          the document to write.
   */
  public void write( final Document doc ) {
    this.write( doc, "profiles/output.xml" );
  }

  /**
   * Writes the given document to the given path.
   * 
   * @param doc
   *          the document to write.
   * @param path
   *          the path to write to.
   */
  public void write( final Document doc, final String path ) {
    try {
      final OutputFormat format = OutputFormat.createPrettyPrint();
      final File file = new File( path );

      LOG.info( "Will create profile in {}", file.getAbsolutePath() );

      if ( file.getParentFile() != null && !file.getParentFile().exists() ) {
        file.getParentFile().mkdirs();
      }

      file.createNewFile();

      final XMLWriter writer = new XMLWriter( new FileWriter( path ), format );
      writer.write( doc );
      writer.close();

    } catch ( final IOException e ) {
      e.printStackTrace();
    }
  }

  /**
   * Generates a profile for the given filter with a default sample size set of
   * 5 and without including the element identifiers.
   * 
   * @param filter
   *          the filter to apply.
   * @return a xml document for the given profile.
   */
  public Document generateProfile( Filter filter ) {
    return this.generateProfile( filter, 5, false );
  }

  /**
   * Generates a profile matching the given filter with the given sample set
   * size and including the elements if set to true.
   * 
   * @param filter
   *          the filter to use.
   * @param sampleSize
   *          the size of the samples.
   * @param includeelements
   *          whether or not to include the element identifiers.
   * @return the xml document of the profile.
   */
  public Document generateProfile( Filter filter, int sampleSize, boolean includeelements ) {
    // TODO check if subFilter is changed
    // if not then it does not have a collection filter...
    // and the collection name should be different...
    Filter subFilter = filter.subFilter( "collection" );
    final long count = this.persistence.count( ELEMENT_CLASS, subFilter );

    final Document document = DocumentHelper.createDocument();
    final String name = this.getCollectionNameFromFilter( filter );
    final Element root = this.createRootElement( document, name, count );
    final Element partition = this.createPartition( root, filter );
    this.genereateFilterElement( partition, filter );
    final Element properties = this.createPropertiesElement( partition );
    this.generateProperties( filter, properties );
    this.createSamples( filter, partition, sampleSize );
    this.createElements( filter, partition, includeelements );

    return document;
  }

  /**
   * Returns the name of the collection from the filter. If the filter is null,
   * then 'all-data' is returned. If the filter is not null, but does not
   * contain a collection condition, then an empty string is returned.
   * 
   * @param filter
   *          the filter to check.
   * @return the name of the collection.
   */
  private String getCollectionNameFromFilter( Filter filter ) {

    if ( filter == null ) {
      return "all-data";
    }

    String result = "";
    List<FilterCondition> conditions = filter.getConditions();
    for ( FilterCondition fc : conditions ) {
      if ( fc.getField().equals( "collection" ) ) {
        result += fc.getValue().toString() + " ";
      }
    }

    return result;
  }

  private void genereateFilterElement( Element partition, Filter filter ) {
    Element elmntFilter = partition.addElement( "filter" );
    // TODO get rid of id
    elmntFilter.addAttribute( "id", UUID.randomUUID().toString() );
    Element parameters = elmntFilter.addElement( "parameters" );

    for ( FilterCondition fc : filter.getConditions() ) {
      Element parameter = parameters.addElement( "parameter" );
      parameter.addElement( "name" ).addText( fc.getField() );
      parameter.addElement( "value" ).addText( fc.getValue().toString() );
    }

  }

  private Element createPartition( Element root, Filter filter ) {

    long count = this.persistence.count( ELEMENT_CLASS, filter );

    final Element partition = root.addElement( "partition" ).addAttribute( "count", count + "" );
    return partition;
  }

  private void generateProperties( final Filter filter, final Element properties ) {
    Iterator<Property> allprops = this.persistence.find( Property.class, null );

    while ( allprops.hasNext() ) {
      Property p = allprops.next();

      Filter copy = new Filter( filter );

      if ( !copy.contains( p.getId() ) ) {
        copy.addFilterCondition( new FilterCondition( p.getId(), null ) );
      }

      long count = this.persistence.count( ELEMENT_CLASS, copy );

      if ( count != 0 ) {
        this.createPropertyElement( filter, properties, p, count );
      }
    }
  }

  private void createPropertyElement( final Filter filter, final Element properties, final Property p, long count ) {
    final Element prop = properties.addElement( "property" ).addAttribute( "id", p.getKey() ).addAttribute( "type",
        p.getType() ).addAttribute( "count", count + "" );

    final PropertyType type = PropertyType.valueOf( p.getType() );

    switch ( type ) {
      case STRING:
        this.processStringProperty( filter, prop, p );
        break;
      case BOOL:
        this.processBoolProperty( filter, prop, p );
        break;
      case INTEGER:
      case FLOAT:
        this.processNumericProperty( filter, prop, p );
        break;
      case DATE:
        this.processDateProperty( filter, prop, p );
        break;
    }
  }

  private Element createRootElement( final Document doc, final String collection, final long count ) {
    final Element profile = doc.addElement( "profile", "http://ifs.tuwien.ac.at/dp/c3po" ).addAttribute( "xmlns:xsi",
        "http://www.w3.org/2001/XMLSchema-instance" ).addAttribute( "collection", collection ).addAttribute( "date",
        new Date() + "" ).addAttribute( "count", count + "" );

    return profile;
  }

  private Element createPropertiesElement( final Element partition ) {
    return partition.addElement( "properties" );
  }

  private void createSamples( final Filter filter, final Element partition, int sampleSize ) {
    final Element samples = partition.addElement( "samples" );
    samples.addAttribute( "type", this.sampleSelector.getType() );
    this.sampleSelector.setFilter( filter );
    final List<String> output = this.sampleSelector.execute( sampleSize );

    LOG.debug( "Found {} representatives", output.size() );
    for ( String s : output ) {
      LOG.debug( "Processing sample {}", s );
      createSampleElement( samples, s );
    }
  }

  private void createSampleElement( final Element samples, final String uid ) {
    Iterator<com.petpet.c3po.api.model.Element> iter = this.persistence.find( ELEMENT_CLASS, new Filter(
        new FilterCondition( "uid", uid ) ) );

    assert iter.hasNext();

    com.petpet.c3po.api.model.Element element = iter.next();

    Element sample = samples.addElement( "sample" ).addAttribute( "uid", uid );
    for ( MetadataRecord mr : element.getMetadata() ) {
      LOG.debug( "Metadata record: {}", mr.getProperty().getKey() );
      if ( mr.getStatus().equals( Status.CONFLICT.toString() ) ) {
        for ( int i = 0; i < mr.getValues().size(); i++ ) {
          Iterator<Source> sources = this.persistence.find( Source.class, new Filter( new FilterCondition( "_id", mr
              .getSources().get( i ) ) ) );
          Source source = sources.next();
          sample.addElement( "record" ).addAttribute( "name", mr.getProperty().getKey() ).addAttribute( "value",
              mr.getValues().get( i ).toString() ).addAttribute( "tool", source.getName() + " " + source.getVersion() );
        }

      } else {
        Iterator<Source> sources = this.persistence.find( Source.class, new Filter( new FilterCondition( "_id", mr
            .getSources().get( 0 ) ) ) );
        Source source = sources.next();
        sample.addElement( "record" ).addAttribute( "name", mr.getProperty().getKey() ).addAttribute( "value",
            mr.getValue().toString() ).addAttribute( "tool", source.getName() + " " + source.getVersion() );
      }
    }
  }

  private void createElements( final Filter filter, final Element partition, boolean includeelements ) {
    final Element elements = partition.addElement( "elements" );

    if ( includeelements ) {

      Iterator<com.petpet.c3po.api.model.Element> iter = this.persistence.find( ELEMENT_CLASS, filter );

      while ( iter.hasNext() ) {
        com.petpet.c3po.api.model.Element element = iter.next();
        elements.addElement( "element" ).addAttribute( "uid", element.getUid() );
      }
    }
  }

  private void processStringProperty( final Filter filter, final Element prop, final Property p ) {
    for ( final String s : PROPERTIES ) {
      if ( p.getKey().equals( s ) ) {

        Map<String, Long> histogram = this.persistence.getValueHistogramFor( p, filter );

        for ( String key : histogram.keySet() ) {
          Long val = histogram.get( key );
          prop.addElement( "item" ).addAttribute( "id", key ).addAttribute( "value", val + "" );
        }

        break;
      }
    }
  }

  private void processBoolProperty( final Filter filter, final Element prop, final Property p ) {

    Map<String, Long> histogram = this.persistence.getValueHistogramFor( p, filter );
    for ( String key : histogram.keySet() ) {
      Long val = histogram.get( key );
      prop.addElement( "item" ).addAttribute( "value", key ).addAttribute( "count", val + "" );
    }

  }

  // if also a histogram is done, do not forget the bin_width...
  private void processNumericProperty( final Filter filter, final Element prop, final Property p ) {
    NumericStatistics numericStatistics = this.persistence.getNumericStatistics( p, filter );

    prop.addAttribute( "count", numericStatistics.getCount() + "" );
    prop.addAttribute( "sum", numericStatistics.getSum() + "" );
    prop.addAttribute( "min", numericStatistics.getMin() + "" );
    prop.addAttribute( "max", numericStatistics.getMax() + "" );
    prop.addAttribute( "avg", numericStatistics.getAverage() + "" );
    prop.addAttribute( "var", numericStatistics.getVariance() + "" );
    prop.addAttribute( "sd", numericStatistics.getStandardDeviation() + "" );
  }

  private void processDateProperty( Filter filter, Element prop, Property p ) {
    Map<String, Long> histogram = this.persistence.getValueHistogramFor( p, filter );

    for ( String key : histogram.keySet() ) {
      Long val = histogram.get( key );
      prop.addElement( "item" ).addAttribute( "id", key ).addAttribute( "value", val + "" );
    }
  }

} public class RepresentativeAlgorithmFactory {

  /**
   * The known algorithms.
   */
  private static final String[] ALGORITHMS = { "sizesampling", "syssampling", "distsampling" };

  /**
   * The factory method.
   * 
   * @param id
   *          the id of the algorithm implementation
   * @return the implementation of the algorithm.
   */
  public RepresentativeGenerator getAlgorithm( String id ) {
    RepresentativeGenerator gen = null;

    int pos = 0;
    for ( int i = 0; i < ALGORITHMS.length; i++ ) {
      if ( ALGORITHMS[i].equalsIgnoreCase( id ) ) {
        pos = i;
        break;
      }
    }

    switch ( pos ) {
      case 0:
        gen = new SizeRepresentativeGenerator();
        break;
      case 1:
        gen = new SystematicSamplingRepresentativeGenerator();
        break;
      case 2:
        gen = new DistributionRepresentativeGenerator();
        break;
      default:
        gen = new SizeRepresentativeGenerator();
    }

    return gen;
  }

  /**
   * Checks if the given algorithm is supported and returns true if yes, false
   * otherwise.
   * 
   * @param alg
   *          the algorithm name to check.
   * @return true if supported, false otherwise.
   */
  public static boolean isValidAlgorithm( String alg ) {
    if ( alg == null ) {
      return false;
    }

    for ( String a : ALGORITHMS ) {
      if ( a.equalsIgnoreCase( alg ) ) {
        return true;
      }
    }

    return false;
  }
} public abstract class RepresentativeGenerator {

  /**
   * A filter used to determine the set on which the representative algorithm is
   * going to be used. Note, that the filter can only specify the collection,
   * meaning that all data should be considered.
   */
  private Filter filter;

  /**
   * A simple map to hold the configuration of the representative generator if
   * any is needed at all.
   */
  private Map<String, Object> options;

  /**
   * Retrieves a list of identifiers for the representative sample objects. It
   * is the responsibility of the method to decide how many objects should be
   * returned. Usually a small number of up to 10 is good for planning
   * experiments.
   * 
   * @return the list of sample object identifiers.
   */
  public abstract List<String> execute();

  /**
   * Retrieves a list of identifiers for the representative sample objects with
   * a predefined number of objects. If the limit is bigger then the set on
   * which the class operates, then all objects should be returned. Note that
   * the method implementation might return less sample records than the
   * specified limit.
   * 
   * @param limit
   *          a max number of sample records to be returned by this method.
   * @return the list of sample object identifiers.
   */
  public abstract List<String> execute( int limit );

  /**
   * Retrieves the type/name of the algorithm used to identify the sample
   * objects.
   * 
   * @return the type.
   */
  public abstract String getType();

  public Filter getFilter() {
    return filter;
  }

  public void setFilter( Filter filter ) {
    this.filter = filter;
  }

  public Map<String, Object> getOptions() {
    return options;
  }

  public void setOptions( Map<String, Object> options ) {
    this.options = options;
  }

} public abstract class AbstractAdaptor implements Runnable {

  /**
   * A default logger for this class.
   */
  private static final Logger LOG = LoggerFactory.getLogger( AbstractAdaptor.class );

  /**
   * The gatherer that is used to obtain the next meta data stream.
   */
  private MetaDataGatherer gatherer;

  /**
   * The configuration that is passed to this adaptor before invoking it.
   */
  private Map<String, String> config;

  /**
   * The processing rules that are passed to this adaptor before invoking it.
   */
  private List<ProcessingRule> rules;

  /**
   * A read only instance to the applications cache, so that the adaptor can
   * obtain {@link Property} and {@link Source} objects.
   */
  private ReadOnlyCache cache;

  /**
   * An internally managed queue that will store the parsed Elements for further
   * processing by the application.
   */
  private Queue<Element> elementsQueue;

  /**
   * The lock on which the adaptors and the gatherer synchronize.
   */
  private Object gatherLock;

  /**
   * This method will be called once before the adaptor is started and gives the
   * implementing class a chance to configure itself. The implementor should
   * make use of the getConfig() methods to read the values for the expected
   * configuration keys of the adaptor. The internally managed config map will
   * contain all configuration keys starting with the prefix 'c3po.adaptor' and
   * all configuration keys starting with the prefix 'c3po.adaptor.[prefix]',
   * where [prefix] is the value retunerd by
   * {@link AbstractAdaptor#getAdaptorPrefix()} of this class in lower case.
   * 
   */
  public abstract void configure();

  /**
   * The prefix that this adaptor should be associated with. This method will be
   * used in order to determine which configurations should be passed to this
   * adaptor. The return value of this method will be transformed to a lower
   * case string.
   * 
   * @return the prefix of this adaptor.
   */
  public abstract String getAdaptorPrefix();

  /**
   * This element is responsible for adapting the data in the object to a
   * {@link Element}. The implementing class should make use of the
   * {@link PreProcessingRule}s provided by the
   * {@link AbstractAdaptor#getPreProcessingRules()} method in cases where the
   * data allows it. Note that the data string will never be null and will
   * contain the contents of each object that was gathered.
   * 
   * @param name
   *          the name of the file/object that is read.
   * @param data
   *          the data to adapt.
   * @return the parsed {@link Element} object.
   */
  public abstract Element parseElement( String name, String data );

  /**
   * Sets the cache to the passed {@link ReadOnlyCache} iff it is not null and
   * the current cache is not set yet.
   * 
   * @param cache
   *          the cache to set.
   */
  public final void setCache( ReadOnlyCache cache ) {
    if ( cache != null && this.cache == null ) {
      this.cache = cache;
    }
  }

  /**
   * Sets the gatherer to the passed {@link MetaDataGatherer} iff it is not
   * null.
   * 
   * @param gatherer
   *          the gatherer to set.
   */
  public final void setGatherer( MetaDataGatherer gatherer ) {
    if ( gatherer != null ) {
      this.gatherer = gatherer;
    }
  }

  /**
   * Sets the object on which to synchronize with the gatherer.
   * 
   * @param gatherLock
   *          the lock object.
   */
  public void setGatherLock( Object gatherLock ) {
    if ( gatherLock != null && this.gatherLock == null ) {
      this.gatherLock = gatherLock;
    }
  }

  /**
   * Sets the processing rules to the given rules iff they are not null and the
   * current list is not set yet.
   * 
   * @param rules
   *          the list of processing rules.
   */
  public final void setRules( List<ProcessingRule> rules ) {
    if ( rules != null && this.rules == null ) {
      this.rules = rules;
    }
  }

  /**
   * Sets the configuration of this adaptor iff the given config is not null and
   * the current config is not set yet..
   * 
   * @param cnf
   *          the config to set.
   */
  public final void setConfig( Map<String, String> cnf ) {
    if ( cnf != null && this.config == null ) {
      this.config = cnf;
    }
  }

  /**
   * Sets the elements queue to the given queue, iff it is not null and the
   * current queue is not set yet. Otherwise a call to this method will do
   * nothing.
   * 
   * @param q
   *          the queue to use.
   */
  public final void setQueue( Queue<Element> q ) {
    if ( q != null && this.elementsQueue == null ) {
      this.elementsQueue = q;
    }
  }

  /**
   * Starts a loop that runs this thread. The thread checks if the gatherer has
   * a next stream to process. If no, then it sleeps until it is notified by the
   * gatherer. If yes, then it gets the next stream and calls the
   * {@link AbstractAdaptor#parseElement(String, String)} method. Once an
   * element is parsed, then all the post processing rules submitted to this
   * adaptor are run on the element. Then the element is submitted for further
   * processing by the next in chain.
   * 
   */
  @Override
  public final void run() {
    try {
      while ( !gatherer.isReady() || gatherer.hasNext() ) {
        try {

          MetadataStream stream = null;
          String name = null;
          String data = null;

          synchronized ( gatherLock ) {
            while ( !gatherer.hasNext() ) {

              if ( gatherer.isReady() ) {
                break;
              }

              gatherLock.wait();
            }

          }
          stream = this.gatherer.getNext();
          if ( stream != null ) {
            name = stream.getName();
            data = stream.getData();
          }

          Element element = null;
          try {

            if ( name != null || data != null ) {
              element = parseElement( name, data );

            }

          } catch ( Exception e ) {
            LOG.warn( "An error occurred while parsing, skipping {}: ", stream.getName(), e.getMessage() );
          }

          postProcessElement( element );

          submitElement( element );

        } catch ( InterruptedException e ) {
          e.printStackTrace();
          break;
        }

      }

      synchronized ( elementsQueue ) {
        this.elementsQueue.notifyAll();
      }

    } catch ( Exception e ) {
      LOG.error( "Adaptor stopped unexpectedly {}", Thread.currentThread().getName() );
      e.printStackTrace();
    }
  }

  /**
   * Gets a read only instance of the application cache. Can be used to obtain
   * {@link Property} and {@link Source} objects.
   * 
   * @return the {@link ReadOnlyCache}
   */
  protected ReadOnlyCache getCache() {
    return cache;
  }

  /**
   * Gets a unmodifiable list of {@link PreProcessingRule} objects. They are
   * sorted by priority, where 0 is the lowest and 1000 is the highest.
   * 
   * @return the rules.
   */
  protected final List<PreProcessingRule> getPreProcessingRules() {
    List<ProcessingRule> all = this.getRules();
    List<PreProcessingRule> prerules = new ArrayList<PreProcessingRule>();

    for ( ProcessingRule rule : all ) {
      if ( rule instanceof PreProcessingRule ) {
        prerules.add( (PreProcessingRule) rule );
      }
    }

    return Collections.unmodifiableList( prerules );
  }

  /**
   * Gets a string config for the given key. If there is no value associated
   * with that key, then an empty string is returned.
   * 
   * @param key
   *          the key to look for.
   * @return the value for the given key, or an empty string if nothing was
   *         associated with it.
   */
  protected String getConfig( String key ) {
    return this.getStringConfig( key, "" );
  }

  /**
   * Gets a string config for the given key. If there is no value associated
   * with that key, then the given defaultValue string is returned.
   * 
   * @param key
   *          the key to look for.
   * @param defaultValue
   *          the value to return if nothing is associated with the given key.
   * @return the value for the given key, or the given defaultValue string if
   *         nothing was associated with it.
   */
  protected String getStringConfig( String key, String defaultValue ) {
    String result = null;
    if ( this.config != null ) {
      result = (String) this.config.get( key );
    }

    if ( result == null ) {
      result = defaultValue;
    }

    return result;
  }

  /**
   * Gets a boolean config for the given key. If there is no value associated
   * with that key, then the given defaultValue is returned.
   * 
   * @param key
   *          the key to look for.
   * @param defaultValue
   *          the value to return if nothing is associated with the given key.
   * @return the value for the given key, or the given defaultValue if nothing
   *         was associated with it.
   */
  protected Boolean getBooleanConfig( String key, boolean defaultValue ) {
    Boolean result = null;
    if ( this.config != null ) {
      String val = this.config.get( key );
      if ( val != null && !val.equals( "" ) ) {
        result = Boolean.parseBoolean( val );
      }
    }

    if ( result == null ) {
      result = defaultValue;
    }

    return result;
  }

  /**
   * Gets a integer config for the given key. If there is no value associated
   * with that key or if it is not a valid integer, then the given defaultValue
   * int is returned.
   * 
   * @param key
   *          the key to look for.
   * @param defaultValue
   *          the value to return if nothing is associated with the given key.
   * @return the value for the given key, or the given defaultValue if nothing
   *         was associated with it.
   */
  protected Integer getIntegerConfig( String key, int defaultValue ) {
    Integer result = null;
    if ( this.config != null ) {
      String val = this.config.get( key );
      if ( val != null && !val.equals( "" ) ) {
        try {
          result = Integer.parseInt( val );
        } catch ( NumberFormatException e ) {
          // do nothing
        }
      }
    }

    if ( result == null ) {
      result = defaultValue;
    }

    return result;
  }

  /**
   * Gets an unmodifiable list of {@link ProcessingRule} objects. All rules are
   * sorted by priority, where 0 is lowest and 1000 is highest.
   * 
   * @return the list of rules.
   */
  private final List<ProcessingRule> getRules() {
    if ( this.rules != null ) {

      Collections.sort( this.rules, new Comparator<ProcessingRule>() {

        // sorts from descending
        @Override
        public int compare( ProcessingRule r1, ProcessingRule r2 ) {
          int first = this.fixPriority( r2.getPriority() );
          int second = this.fixPriority( r1.getPriority() );
          return new Integer( first ).compareTo( new Integer( second ) );
        }

        private int fixPriority( int prio ) {
          if ( prio < 0 )
            return 0;

          if ( prio > 1000 )
            return 1000;

          return prio;
        }

      } );

    }

    return Collections.unmodifiableList( rules );
  }

  /**
   * Gets a unmodifiable list of {@link PostProcessingRule} objects. They are
   * sorted by descending priority, where 1000 is the highest and 0 is the
   * lowest.
   * 
   * @return the rules.
   */
  private final List<PostProcessingRule> getPostProcessingRules() {
    List<ProcessingRule> all = this.getRules();
    List<PostProcessingRule> postrules = new ArrayList<PostProcessingRule>();

    for ( ProcessingRule rule : all ) {
      if ( rule instanceof PostProcessingRule ) {
        postrules.add( (PostProcessingRule) rule );
      }
    }

    return Collections.unmodifiableList( postrules );
  }

  /**
   * Gets all post processing rules and applies them to the given element if it
   * is not null.
   * 
   * @param element
   *          the element to post process.
   */
  private void postProcessElement( Element element ) {
    if ( element != null ) {
      List<PostProcessingRule> postProcessingRules = getPostProcessingRules();

      for ( PostProcessingRule rule : postProcessingRules ) {
        element = rule.process( element );
      }
    }
  }

  /**
   * Submits the passed element to the processing queue iff it is not null. This
   * method is synchronized on the queue.
   * 
   * @param e
   *          the element to submit to the queue.
   */
  private final void submitElement( Element e ) {
    synchronized ( elementsQueue ) {
      if ( e != null ) {
        elementsQueue.add( e );
        elementsQueue.notify();
      }
    }
  }
} public interface ProcessingRule {

  /**
   * A priority between 0 and 1000 in the case of more rules, where 0 is the
   * least important and 1000 is the most important rule. If the set priority is
   * smaller than 0, then 0 will be used, if it is larger than 1000, then 1000
   * will be used.
   * 
   * @return the priority of this rule.
   */
  int getPriority();
} public interface PersistenceLayer {

  /**
   * Clears the current cache of the application. This includes the
   * {@link Cache}, but also any other cached information that the backend
   * provider might have stored (e.g. cached statistics and aggregations).
   */
  void clearCache();

  /**
   * This method will be called once before the applications ends running.
   */
  void close() throws C3POPersistenceException;

  /**
   * Counts the number of objects of the given type that match the filter. If
   * the filter is null, the count of all objects of that type is retrieved.
   * 
   * @param clazz
   *          the type of objects
   * @param filter
   *          the filter.
   * @return the number of objects corresponding to that type and filter.
   */
  <T extends Model> long count( Class<T> clazz, Filter filter );

  /**
   * This method will be called upon initialisation of the application. The map
   * will contain all configurations (loaded from the config file) starting with
   * the prefix 'db'. Example: db.host=192.168.0.1
   * 
   * @param config
   *          the configuration
   */
  void establishConnection( Map<String, String> config ) throws C3POPersistenceException;

  /**
   * Finds objects corresponding to the supplied filter and type. The type has
   * to implement the Model interface. Retrieves an iterator for the objects.
   * 
   * Note that depending on the underlying implementation this method might
   * fetch a lot of data from the database.
   * 
   * @param clazz
   *          the generic type of the iterator
   * @param filter
   *          the optional filter.
   * @return a typed iterator for the given type of objects.
   */
  <T extends Model> Iterator<T> find( Class<T> clazz, Filter filter );

  /**
   * Returns the distinct values for the given property (in string form). If the
   * passed filter is not null, then it is applied on the collection before
   * 
   * @param clazz
   *          the type of the objects to look at.
   * @param f
   *          the field for which the distinct values have to be applied
   * @param filter
   *          the filter that has to be applied on the data before the distinct
   *          operation is executed
   * @return the distinct values for a given property
   */
  <T extends Model> List<String> distinct( Class<T> clazz, String f, Filter filter );

  /**
   * Gets a cache object.
   * 
   * @return the cache of the application.
   */
  Cache getCache();

  /**
   * Returns statistics for the given numeric property for all elements
   * according to the given filter. The values of the property have to have a
   * numeric data type as the computation might/will involve mathematical
   * calculations.
   * 
   * As this method is a higher-level query it can only support numeric
   * properties and the Element.class
   * 
   * @param p
   *          the property for which the statistcs will be created.
   * @param filter
   *          the filter to apply.
   * @return {@link NumericStatistics} object containing statistics like: min,
   *         max, avg, sd, var, etc.
   * @throws UnsupportedOperationException
   *           if the operation is not supported by the implementing backend
   * @throws IllegalArgumentException
   *           if the arguments are not appropriate, e.g. if the datatype of the
   *           property is not of a numeric type.
   */
  NumericStatistics getNumericStatistics( Property p, Filter filter ) throws UnsupportedOperationException,
      IllegalArgumentException;

  /**
   * Returns a property value histogram for the given property and type and
   * respecting the given filter.
   * 
   * As this method is a higher-level query it should only support the Element
   * class.
   * 
   * @param p
   *          the property for which the value histogram will be created in the
   *          corresponding model object
   * @param filter
   *          the filter that has to be applied before the operation is
   *          conducted
   * @return a map with the distinct values as keys (strings) and their
   *         occurrences as map-values (long)
   * 
   * @throws UnsupportedOperationException
   *           if the current persistence layer cannot create such a histogram.
   */
  <T extends Model> Map<String, Long> getValueHistogramFor( Property p, Filter filter )
      throws UnsupportedOperationException;

  /**
   * Inserts the given object to the underlying data store.
   * 
   * @param object
   *          the object to store.
   */
  <T extends Model> void insert( T object );

  /**
   * Whether or not there is an established connection to the database.
   * 
   * @return true if there is a connection, false otherwise.
   */
  boolean isConnected();

  /**
   * Removes the objects of the given type that match the filter from the
   * underlying data store. If the filter is null, then all objects are removed.
   * 
   * @param clazz
   *          the type of object to remove
   * @param filter
   *          the filter to match. Can be null.
   */
  <T extends Model> void remove( Class<T> clazz, Filter filter );

  /**
   * Removes the given object from the underlying data store.
   * 
   * @param object
   *          the object to remove.
   */
  <T extends Model> void remove( T object );

  /**
   * Sets the cache to the passed cache.
   * 
   * @param c
   *          the cache to use.
   */
  void setCache( Cache c );

  /**
   * Inserts or updates the given object to the underlying data store. Note that
   * if the filter is null the backend provider might not be able to infer which
   * object should be updated. Make sure the filter uniquely identifies all
   * objects that have to be updated with the passed one.
   * 
   * @param object
   *          the object to update.
   * @param f
   *          the filter that uniquely identifies the object.
   */
  <T extends Model> void update( T object, Filter f );
} public interface MetaDataGatherer extends Runnable {

  /**
   * Some gatherer specific configuration.
   * 
   * @param config
   */
  void setConfig( Map<String, String> config );

  /**
   * Gets the next {@link MetadataStream} object.
   * 
   * @return
   */
  MetadataStream getNext();

  /**
   * Whether or not the gatherer has a next element at the moment of the call.
   * 
   * @return
   */
  boolean hasNext();

  /**
   * Whether or not the gatherer has finished its work.
   * 
   * @return
   */
  boolean isReady();

} public class Element implements Model {

  /**
   * A back-end related identifier of this object.
   */
  private String id;

  /**
   * The collection to which the current element belongs.
   */
  private String collection;

  /**
   * Some non-unique name of this element.
   */
  private String name;

  /**
   * Some unique identifier of this element that references the original file
   * back in the source.
   */
  private String uid;

  /**
   * A list of {@link MetadataRecord} info.
   */
  private List<MetadataRecord> metadata;

  /**
   * Creates an element with the given uid and name.
   * 
   * @param uid
   *          the unique identifier of this element.
   * @param name
   *          the name of this element.
   */
  public Element(String uid, String name) {
    this.uid = uid;
    this.name = name;
    this.metadata = new ArrayList<MetadataRecord>();
  }

  /**
   * Creates an element with the given uid, name and collection.
   * 
   * @param collection
   * @param uid
   * @param name
   */
  public Element(String collection, String uid, String name) {
    this( uid, name );
    this.collection = collection;
  }

  public String getCollection() {
    return collection;
  }

  public void setCollection( String collection ) {
    this.collection = collection;
  }

  public String getName() {
    return name;
  }

  public void setName( String name ) {
    this.name = name;
  }

  public String getUid() {
    return uid;
  }

  public void setUid( String uid ) {
    this.uid = uid;
  }

  public List<MetadataRecord> getMetadata() {
    return metadata;
  }

  public void setMetadata( List<MetadataRecord> metadata ) {
    this.metadata = metadata;
  }

  public String getId() {
    return id;
  }

  public void setId( String id ) {
    this.id = id;
  }

  /**
   * Removes all records for the given property id and returns a list of all
   * removed meta data records.
   * 
   * @param property
   *          the id of the property
   * @return returns the records of the element matching this property that were
   *         deleted.
   */
  public List<MetadataRecord> removeMetadata( String property ) {
    List<MetadataRecord> result = new ArrayList<MetadataRecord>();

    Iterator<MetadataRecord> iterator = this.metadata.iterator();
    while ( iterator.hasNext() ) {
      MetadataRecord next = iterator.next();
      if ( next.getProperty().getId().equals( property ) ) {
        result.add( next );
        iterator.remove();
      }
    }

    return result;
  }

} public class Filter {

  /**
   * A list of filter conditions, that has to be interpreted by the persistence
   * provider to a query or addition to a query.
   */
  private List<FilterCondition> conditions;

  /**
   * Create a new empty filter with an initialized empty list of
   * {@link FilterCondition}s.
   */
  public Filter() {
    conditions = new ArrayList<FilterCondition>();
  }

  /**
   * Creates a new filter and adds the given conditions to the filter.
   * 
   * @param fc
   *          the condition to add.
   */
  public Filter(FilterCondition fc) {
    this();
    this.conditions.add( fc );
  }

  /**
   * Creates a new filter and sets the {@link FilterCondition}s to the passed
   * list.
   * 
   * @param fcs
   *          the filter conditions to use.
   */
  public Filter(List<FilterCondition> fcs) {
    conditions = fcs;
  }

  /**
   * Copies the given filter to this filter.
   * 
   * @param another
   *          the filter to copy.
   */
  public Filter(Filter another) {
    this.conditions = new ArrayList<FilterCondition>( another.conditions );
  }

  public List<FilterCondition> getConditions() {
    return conditions;
  }

  public void setConditions( List<FilterCondition> conditions ) {
    this.conditions = conditions;
  }

  /**
   * Adds a new filter condition unless it is null or it already exists.
   * 
   * @param fc
   *          the filter condition to add.
   */
  public void addFilterCondition( FilterCondition fc ) {
    if ( conditions == null ) {
      conditions = new ArrayList<FilterCondition>();
    }

    if ( fc != null && !conditions.contains( fc ) )
      conditions.add( fc );
  }

  /**
   * Create a new filter that contains only the {@link FilterCondition} objects
   * that are part of the fields array. However, if the new filter has no filter
   * condition, then the current filter is returned.
   * 
   * @param fields
   *          the fields that should remain in the subfilter.
   * @return the new filter or the same if none of the conditions were matching.
   */
  public Filter subFilter( String... fields ) {
    Filter subFilter = new Filter();
    List<String> fieldList = Arrays.asList( fields );

    for ( FilterCondition fc : this.conditions ) {
      if ( fieldList.contains( fc.getField() ) ) {
        subFilter.addFilterCondition( fc );
      }
    }

    return (subFilter.getConditions().size() > 0) ? subFilter : this;
  }

  /**
   * Returns true if this filter has at least one {@link FilterCondition} with
   * the given field.
   * 
   * @param field
   *          the field to check for.
   * @return true if it contains such a filter condition, false otherwise.
   */
  public boolean contains( String field ) {
    for ( FilterCondition fc : this.conditions ) {
      if ( fc.getField().equals( field ) ) {
        return true;
      }
    }

    return false;
  }

  /**
   * Checks if it contains the given {@link FilterCondition}. Note that the
   * values are checked for equality as well.
   * 
   * @param fc
   *          the {@link FilterCondition} to check.
   * @return true if it contains this {@link FilterCondition}, false otherwise.
   */
  public boolean contains( FilterCondition fc ) {
    return this.conditions.contains( fc );
  }

  @Override
  public int hashCode() {
    final int prime = 31;
    int result = 1;
    result = prime * result + ((conditions == null) ? 0 : conditions.hashCode());
    return result;
  }

  @Override
  public boolean equals( Object obj ) {
    if ( this == obj )
      return true;
    if ( obj == null )
      return false;
    if ( getClass() != obj.getClass() )
      return false;
    Filter other = (Filter) obj;
    if ( conditions == null ) {
      if ( other.conditions != null )
        return false;
    } else if ( !conditions.equals( other.conditions ) )
      return false;
    return true;
  }

} public class FilterCondition {

  /**
   * The property of this condition.
   */
  private String field;

  /**
   * The value of this condition.
   */
  private Object value;

  /**
   * Default empty constructor.
   */
  public FilterCondition() {

  }

  /**
   * Sets the fields of this object.
   * 
   * @param f
   *          the field
   * @param v
   *          the value for the property condition.
   */
  public FilterCondition(String f, Object v) {
    field = f;
    value = v;
  }

  public String getField() {
    return field;
  }

  public void setProperty( String f ) {
    this.field = f;
  }

  public Object getValue() {
    return value;
  }

  public void setValue( Object value ) {
    this.value = value;
  }

  @Override
  public int hashCode() {
    final int prime = 31;
    int result = 1;
    result = prime * result + ((field == null) ? 0 : field.hashCode());
    result = prime * result + ((value == null) ? 0 : value.hashCode());
    return result;
  }

  @Override
  public boolean equals( Object obj ) {
    if ( this == obj )
      return true;
    if ( obj == null )
      return false;
    if ( getClass() != obj.getClass() )
      return false;
    FilterCondition other = (FilterCondition) obj;
    if ( field == null ) {
      if ( other.field != null )
        return false;
    } else if ( !field.equals( other.field ) )
      return false;
    if ( value == null ) {
      if ( other.value != null )
        return false;
    } else if ( !value.equals( other.value ) )
      return false;
    return true;
  }

} public final class Constants {

  /**
   * The url for the xml schema property used by the sax parser while validating
   * xml files against their schemata.
   */
  public static final String XML_SCHEMA_PROPERTY = "http://java.sun.com/xml/jaxp/properties/schemaLanguage";

  /**
   * The url for the xml schema language used by the sax parser while validating
   * xml files against their schemata.
   */
  public static final String XML_SCHEMA_LANGUAGE = "http://www.w3.org/2001/XMLSchema";

  /**
   * A c3po configuration for the collection on which to operate.
   */
  public static final String OPT_COLLECTION_NAME = "c3po.collection.name";

  /**
   * A c3po configuration for the location where the metadata is.
   */
  public static final String OPT_COLLECTION_LOCATION = "c3po.collection.location";

  /**
   * A c3po configuration for the type of the input files. Currently only FITS
   * and TIKA are supported. This config is required for the controller to
   * operate.
   */
  public static final String OPT_INPUT_TYPE = "c3po.input.type";

  /**
   * A configuartion for recursive processing.
   */
  public static final String OPT_RECURSIVE = "c3po.recursive";

  /**
   * A configuration option for the sampling algorithm to use.
   */
  public static final String OPT_SAMPLING_ALGORITHM = "c3po.sampling.alogrithm";

  /**
   * A configuration option for the size of the sample set.
   */
  public static final String OPT_SAMPLING_SIZE = "c3po.sampling.size";

  /**
   * A configuration option for the properties over which some sample algorithms
   * should work.
   */
  public static final String OPT_SAMPLING_PROPERTIES = "c3po.sampling.properties";

  /**
   * A configuration option for including the element identifiers.
   */
  public static final String OPT_INCLUDE_ELEMENTS = "c3po.profile.includeelements";

  /**
   * A configuration option for the output location of a CLI command.
   */
  public static final String OPT_OUTPUT_LOCATION = "c3po.cli.outputlocation";

  /**
   * A config identifier for the persistence layer class.
   */
  public static final String CNF_PERSISTENCE = "c3po.persistence";

  /**
   * The thread count configuration for the adaptor workers.
   */
  public static final String CNF_ADAPTORS_COUNT = "c3po.controller.adaptors.count";

  /**
   * The thread count configuration for the consolidator workers.
   */
  public static final String CNF_CONSOLIDATORS_COUNT = "c3po.controller.consolidators.count";

  /**
   * The key for the create element identifier processing rule.
   */
  public static final String CNF_ELEMENT_IDENTIFIER_RULE = "c3po.rule.create_element_identifier";

  /**
   * The key for the empty value processing rule.
   */
  public static final String CNF_EMPTY_VALUE_RULE = "c3po.rule.empty_value_processing";

  /**
   * The key for the format version resolution processing rule.
   */
  public static final String CNF_VERSION_RESOLUTION_RULE = "c3po.rule.format_version_resolution";

  /**
   * The key for the html info processing rule.
   */
  public static final String CNF_HTML_INFO_RULE = "c3po.rule.html_info_processing";

  /**
   * The key for the infer date processing rule.
   */
  public static final String CNF_INFER_DATE_RULE = "c3po.rule.infer_date_from_file_name";

  /**
   * An array of the configurable processing rule keys.
   */
  public static final String[] RULE_KEYS = { CNF_ELEMENT_IDENTIFIER_RULE, CNF_VERSION_RESOLUTION_RULE,
      CNF_EMPTY_VALUE_RULE, CNF_INFER_DATE_RULE, CNF_HTML_INFO_RULE };

  /**
   * The version of the core module.
   */
  public static final String CORE_VERSION = "0.4.0-SNAPSHOT";

  /**
   * The version of the api module.
   */
  public static final String API_VERSION = "0.4.0-SNAPSHOT";

  /**
   * The same as {@link Constants#AGGREGATE_MAP} but it aggregates the desired
   * property only for elements where the passed filter has a specific value.
   * {1} - the filter property id (e.g. 'mimetype') {2} - the value of the
   * filter (e.g. 'application/pdf') {3} - the property to aggregate (e.g.
   * 'size')
   */
  @Deprecated
  public static final String FILTER_AGGREGATE_MAP = "function map() {if (this.metadata['{1}'].value === '{2}') {emit(1,{sum: this.metadata['{3}'].value, min: this.metadata['{3}'].value,max: this.metadata['{3}'].value,count:1,diff: 0,});}}";

  @Deprecated
  public static final String PROPERTIES_IN_COLLECTION_MAP = "function map() {for (var key in this) {if (key == 'metadata') {for (var subkey in this[key]) {emit(subkey, null);}}}}";

  @Deprecated
  public static final String PROPERTIES_IN_COLLECTION_REDUCE = "function reduce(key, values) {return null;}";

  private Constants() {

  }
} public class LocalFileGatherer implements MetaDataGatherer {

  /**
   * A default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( LocalFileGatherer.class );

  /**
   * A static integer for tmp folders.
   */
  private static int folder = 0;

  /**
   * A list of supported archive extensions.
   */
  private static final String[] ARCHIVE_EXTENSIONS = {
      ".zip", ".tar", ".bzip2", ".tar.bz2",
      ".bz2", ".tb2", ".tbz", ".tar.gz",
      ".tgz", ".gz", ".tar.xz", ".txz", ".xz"
  };

  /**
   * The configuration of the gatherer.
   */
  private Map<String, String> config;

  /**
   * A queue where the {@link FileMetadataStream} objects are stored.
   */
  private final Queue<MetadataStream> queue;

  /**
   * The count of objects gathered.
   */
  private long count;

  /**
   * A flag denoting whether this gatherer is ready traversing the file system.
   */
  private boolean ready;

  /**
   * A lock for synchronization with other workers.
   */
  private Object lock;

  /**
   * Creates a new gatherer.
   */
  public LocalFileGatherer() {
    this.queue = new LinkedList<MetadataStream>();
    this.ready = false;
  }

  /**
   * Creates a new gatherer with the given config.
   * 
   * @param config
   */
  public LocalFileGatherer(Map<String, String> config) {
    this();
    this.config = config;
  }

  /**
   * Creates a new gatherer with the given object lock.
   * 
   * @param lock
   */
  public LocalFileGatherer(Object lock) {
    this();
    this.lock = lock;
  }

  /**
   * Runs this gatherer and traverses the file system (optionally in a recursive
   * fashion) . Once the files are gathered, all waiting threads on the locks
   * monitor are notified.
   */
  @Override
  public synchronized void run() {
    String path = this.config.get( Constants.OPT_COLLECTION_LOCATION );
    boolean recursive = Boolean.valueOf( this.config.get( Constants.OPT_RECURSIVE ) );

    this.ready = false;
    this.traverseFiles( new File( path ), recursive, true );
    System.out.println( this.count + " files were submitted for processing" );
    LOG.info( "{} files were submitted for processing successfully", this.count );
    this.ready = true;

    synchronized ( lock ) {
      this.lock.notifyAll();

    }
  }

  /**
   * {@inheritDoc}
   */
  public synchronized MetadataStream getNext() {
    return queue.poll();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void setConfig( Map<String, String> config ) {
    this.config = config;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean hasNext() {
    return !this.queue.isEmpty();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isReady() {
    return this.ready;
  }

  /**
   * Traverses the file system starting from the given file. If the recursive
   * flag is true, then the traversal is recursive. Processes files and
   * archives. If an archive is encountered, then the it is extracted in a tmp
   * direcotory, which is then traversed recursively. At the end of the process
   * the tmp directory is removed.
   * 
   * @param file
   *          the directory to traverse.
   * @param recursive
   *          whether or not to do it recursively.
   * @param firstLevel
   *          denotes whether this is the first level of traversal.
   */
  private void traverseFiles( File file, boolean recursive, boolean firstLevel ) {

    if ( file.isDirectory() && (recursive || firstLevel) ) {

      File[] files = file.listFiles();
      for ( File f : files ) {
        traverseFiles( f, recursive, false );
      }
    } else {
      String filePath = file.getAbsolutePath();

      if ( isArchive( filePath ) ) {

        extractArchive( filePath );

      } else {

        submitMetadataResult( filePath );

      }
    }

  }

  /**
   * Checks if the file denoted by the given name is an archive based on the
   * supported extensions.
   * 
   * @param name
   *          the name to check.
   * @return true if it is an archive, false otherwise.
   */
  private boolean isArchive( String name ) {
    for ( String ext : ARCHIVE_EXTENSIONS ) {
      if ( name.endsWith( ext ) ) {
        return true;
      }
    }

    return false;
  }

  /**
   * Processes an archive file denoted by the given file path.
   * 
   * @param filePath
   *          the file path to the archive.
   */
  private void extractArchive( String filePath ) {
    TFile archive = new TFile( filePath );
    String tmp = FileUtils.getTempDirectory().getPath() + File.separator + "c3poarchives" + File.separator + folder++;
    File tmpDir = new File( tmp );

    try {

      tmpDir.mkdirs();
      TFile directory = new TFile( tmpDir );
      TFile.cp_r( archive, directory, TArchiveDetector.NULL, TArchiveDetector.NULL );
      traverseFiles( tmpDir, true, true );

    } catch ( IOException e ) {
      LOG.warn( "An error occurred whule extracting file {}. Error: {}", filePath, e.getMessage() );
    }
  }

  private void submitMetadataResult( String filePath ) {
    FileMetadataStream ms = new FileMetadataStream( filePath );
    this.queue.add( ms );
    count++;

    if ( (this.count % 1000) == 0 ) {
      LOG.info( "{} files were submitted for processing", this.count );
      synchronized ( lock ) {
        this.lock.notify();

      }
    }

    if ( this.queue.size() > 10000 && this.count % 1000 == 0 ) {
      synchronized ( lock ) {
        this.lock.notifyAll();
      }
    }

    if ( this.count % 10000 == 0 ) {
      System.out.println( this.count + " files were submitted for processing" );
    }

  }
} public final class Configurator {

  /**
   * Default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( Configurator.class );

  /**
   * The user specified configuration file path. Can be found in ~/.c3poconfig
   */
  public static final String USER_PROPERTIES = System.getProperty( "user.home" ) + File.separator + ".c3poconfig";

  /**
   * The Default persistence layer implementation.
   */
  private PersistenceLayer persistence;

  /**
   * A properties object holding the loaded configuration of the application.
   */
  private Properties config;

  /**
   * Obtains the singleton instance of this class.
   * 
   * @return the configurator.
   */
  public static Configurator getDefaultConfigurator() {
    return ConfiguratorHolder.UNIQUE_INSTANCE;
  }

  /**
   * Hidden default constructor.
   */
  private Configurator() {

  }

  /**
   * Configures the application in the following order: <br>
   * 1. loads the application configuration file <br>
   * 2. initializes the persistence layer with respect to the config <br>
   * 3. inits the helper objects.
   */
  public void configure() {
    this.loadApplicationConfiguration();
    this.initPersistenceLayer();
    this.initializeHelpers();
  }

  /**
   * Retrieves the current persistence layer.
   * 
   * @return {@link PersistenceLayer}
   */
  public PersistenceLayer getPersistence() {
    return this.persistence;
  }

  /**
   * Gets a String representation for the property key or an empty string if no
   * property was found.
   * 
   * @param key
   *          the key of the property
   * @return a string with the value or an empty string if none found.
   */
  public String getStringProperty( final String key ) {
    return this.config.getProperty( key, "" );
  }

  /**
   * Returns the string property for the given key, or returns the default value
   * if nothing is defined under the given key.
   * 
   * @param key
   *          the key to look for
   * @param def
   *          the default value, in case the key does not exists.
   * @return the string value corresponding to that key.
   */
  public String getStringProperty( final String key, final String def ) {
    return this.config.getProperty( key, def );
  }

  /**
   * Returns an integer value for the specified property key or -1.
   * 
   * @param key
   *          the key of the property.
   * @return the value or -1.
   */
  public int getIntProperty( final String key ) {
    return getIntProperty( key, -1 );
  }

  /**
   * Returns an integer for the specified property or the default value if the
   * key is not defined in the configs.
   * 
   * @param key
   *          the key to look for.
   * @param def
   *          the default value if the key does not exist.
   * @return the value corresponding to the key or the default value.
   */
  public int getIntProperty( final String key, final int def ) {
    String property = this.config.getProperty( key );
    int result = def;
    if ( property != null && !property.equals( "" ) ) {
      try {
        result = Integer.parseInt( property );
      } catch ( NumberFormatException e ) {
        // nothing to do - return the default.
        LOG.trace( "The provided property {} is not a number, returning the default: {}", key, def );
      }
    }
    return result;
  }

  /**
   * Returns a boolean value for the specified property key or false if none was
   * found.
   * 
   * @param key
   *          the key of the property.
   * @return the value or false.
   */
  public boolean getBooleanProperty( final String key ) {
    return Boolean.valueOf( this.config.getProperty( key, "false" ) );
  }

  /**
   * Loads the default configuration file and then looks in the users home for a
   * specific configuration. All properties defined there will overwrite the
   * default behavior.
   */
  private void loadApplicationConfiguration() {
    LOG.info( "Loading default configuration file" );
    this.config = new Properties();

    try {
      this.config.load( Configurator.class.getClassLoader().getResourceAsStream( "default.properties" ) );
    } catch ( final IOException e ) {
      LOG.error( "Default config file not found! {}", e.getMessage() );
    }

    LOG.info( "Lookig for user defined config file: {}", USER_PROPERTIES );

    final File f = new File( USER_PROPERTIES );

    if ( f.exists() && f.isFile() ) {
      LOG.debug( "Found user defined properties, loading." );
      FileInputStream stream = null;
      try {
        stream = new FileInputStream( f );
        this.config.load( stream );
        LOG.info( "User defined config is successfully loaded" );
        LOG.debug( this.config.toString() );

      } catch ( final IOException e ) {
        LOG.warn( "Could not load user defined properties file '{}', cause: {}", USER_PROPERTIES, e.getMessage() );
      } finally {
        try {
          if ( stream != null ) {
            stream.close();
          }
        } catch ( final IOException e ) {
          LOG.warn( "Could not close the stream in a clean fashion: {}", e.getMessage() );
        }
      }

    } else {
      LOG.info( "User defined config file was not found." );
    }

  }

  /**
   * Initialises the persistence layer based on the property defined within the
   * config file. If the config file does not define a persistence class or has
   * "default" as value, then the default persistence layer class is
   * initialised. If another class is found, then it is instantiated and set as
   * the persistence layer. However, if the instantiation fails for some reason,
   * then the default persistence layer is initialised, which still might be
   * error prone.
   */
  private void initPersistenceLayer() {

    String persistence = this.getStringProperty( Constants.CNF_PERSISTENCE );

    if ( persistence.equals( "" ) || persistence.equals( "default" ) ) {

      initDefaultPersistence();

    } else {

      try {
        Class<PersistenceLayer> persistenceLayerClazz = (Class<PersistenceLayer>) Class.forName( persistence );

        this.persistence = persistenceLayerClazz.newInstance();

        this.initCache();

      } catch ( ClassNotFoundException e ) {
        LOG.error( "Could not find persistence class '{}' on the classpath! Error: {}", persistence, e.getMessage() );
        LOG.warn( "Trying to use the default persistence layer" );
        initDefaultPersistence();

      } catch ( InstantiationException e ) {
        LOG.error( "Could not instantiate persistence layer class '{}'! Error: {}", persistence, e.getMessage() );
        LOG.warn( "Trying to use the default persistence layer" );
        initDefaultPersistence();

      } catch ( IllegalAccessException e ) {
        LOG.error( "Could not access persistence layer class '{}'! Error: {}", persistence, e.getMessage() );
        LOG.warn( "Trying to use the default persistence layer" );
        initDefaultPersistence();

      } catch ( Exception e ) {
        LOG.error( "Could not start persistence layer class '{}'! Error: {}", persistence, e.getMessage() );
        LOG.warn( "Trying to use the default persistence layer" );
        initDefaultPersistence();

      }

    }

    try {

      Map<String, String> persistenceConfigs = this.getPersistenceConfigs();
      this.persistence.establishConnection( persistenceConfigs );

    } catch ( C3POPersistenceException e ) {
      LOG.error( "Could not establish connection to the data store: {}", e.getMessage() );
    }
  }

  /**
   * Inits the default persistence layer and the cache.
   */
  private void initDefaultPersistence() {
    this.persistence = new DefaultPersistenceLayer();

    this.initCache();
  }

  /**
   * Inits the cache.
   */
  private void initCache() {
    DBCache c = new DBCache();
    c.setPersistence( this.persistence );

    this.persistence.setCache( c );
  }

  /**
   * Takes all properties starting with 'db.' and puts them in a map.
   * 
   * @return a map of all properties related to the db.
   */
  private Map<String, String> getPersistenceConfigs() {
    Map<String, String> dbConfig = new HashMap<String, String>();

    Set<Object> keySet = this.config.keySet();
    for ( Object k : keySet ) {
      String key = (String) k;
      if ( key.startsWith( "db." ) ) {
        dbConfig.put( key, this.getStringProperty( key ) );
      }
    }

    return dbConfig;
  }

  /**
   * Initializes some helper objects.
   */
  private void initializeHelpers() {
    LOG.debug( "Initializing helpers." );
    XMLUtils.init();
    FITSHelper.init();
    TIKAHelper.init();
    DataHelper.init();
    // initialize any other helpers...
  }

  /**
   * Static instance holder.
   * 
   */
  private static final class ConfiguratorHolder {

    /**
     * Unique instance.
     */
    public static final Configurator UNIQUE_INSTANCE = new Configurator();

    /**
     * Hidden default constructor.
     */
    private ConfiguratorHolder() {

    }
  }
}
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.dom4j.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.petpet.c3po.adaptor.fits.FITSAdaptor;
import com.petpet.c3po.adaptor.rules.AssignCollectionToElementRule;
import com.petpet.c3po.adaptor.rules.CreateElementIdentifierRule;
import com.petpet.c3po.adaptor.rules.EmptyValueProcessingRule;
import com.petpet.c3po.adaptor.rules.FormatVersionResolutionRule;
import com.petpet.c3po.adaptor.rules.HtmlInfoProcessingRule;
import com.petpet.c3po.adaptor.rules.InferDateFromFileNameRule;
import com.petpet.c3po.adaptor.tika.TIKAAdaptor;
import com.petpet.c3po.analysis.CSVGenerator;
import com.petpet.c3po.analysis.ProfileGenerator;
import com.petpet.c3po.analysis.RepresentativeAlgorithmFactory;
import com.petpet.c3po.analysis.RepresentativeGenerator;
import com.petpet.c3po.api.adaptor.AbstractAdaptor;
import com.petpet.c3po.api.adaptor.ProcessingRule;
import com.petpet.c3po.api.dao.PersistenceLayer;
import com.petpet.c3po.api.gatherer.MetaDataGatherer;
import com.petpet.c3po.api.model.ActionLog;
import com.petpet.c3po.api.model.Element;
import com.petpet.c3po.api.model.helper.Filter;
import com.petpet.c3po.api.model.helper.FilterCondition;
import com.petpet.c3po.common.Constants;
import com.petpet.c3po.gatherer.LocalFileGatherer;
import com.petpet.c3po.utils.ActionLogHelper;
import com.petpet.c3po.utils.Configurator;
import com.petpet.c3po.utils.exceptions.C3POConfigurationException;
/*******************************************************************************
 * Copyright 2013 Petar Petrov <me@petarpetrov.org>
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 ******************************************************************************/
package com.petpet.c3po.controller;




/**
 * A controller that manages the operations coming as input from the client
 * applications. This class ties up the gathering, adaptation and consolidation
 * of data. It acts as a facade of the core to the client applications.
 * 
 * @author Petar Petrov <me@petarpetrov.org>
 * 
 */
public class Controller {

  /**
   * A default logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger( Controller.class );

  /**
   * The persistence layer that this class uses.
   */
  private PersistenceLayer persistence;

  /**
   * A thread pool for the adaptors.
   */
  private ExecutorService adaptorPool;

  /**
   * A thread pool for the consolidators.
   */
  private ExecutorService consolidatorPool;

  /**
   * A meta data gatherer that collects meta data objects.
   */
  private MetaDataGatherer gatherer;

  /**
   * A processing queue that is passed to each adaptor and is used for the
   * synchronisation between adaptors and consolidators.
   */
  private final Queue<Element> processingQueue;

  /**
   * A map of the known adaptors.
   */
  private final Map<String, Class<? extends AbstractAdaptor>> knownAdaptors;

  /**
   * A map of processing rules.
   */
  private final Map<String, Class<? extends ProcessingRule>> knownRules;

  /**
   * A {@link Configurator} that holds applications specific configurations.
   */
  private Configurator configurator;

  /**
   * A lock object for synchronization between the gatherer and the adaptors.
   */
  private Object gatherLock;

  /**
   * This constructors sets the persistence layer, initializes the processing
   * queue and the {@link LocalFileGatherer};
   * 
   * @param config
   *          a configurator that holds application specific configs and can
   *          initialize the application.
   * 
   */
  public Controller(Configurator config) {
    this.configurator = config;
    this.persistence = config.getPersistence();
    this.processingQueue = new LinkedList<Element>();
    this.gatherLock = new Object();
    this.gatherer = new LocalFileGatherer( this.gatherLock );
    this.knownAdaptors = new HashMap<String, Class<? extends AbstractAdaptor>>();
    this.knownRules = new HashMap<String, Class<? extends ProcessingRule>>();

    // TODO detect adaptors automatically from the class path
    // and add them to this map.
    this.knownAdaptors.put( "FITS", FITSAdaptor.class );
    this.knownAdaptors.put( "TIKA", TIKAAdaptor.class );

    // TODO detect these automatically from the class path
    // and add them to this map.
    // TODO the InferDateFromFileNameRule needs a setter for the cache.
    this.knownRules.put( Constants.CNF_ELEMENT_IDENTIFIER_RULE, CreateElementIdentifierRule.class );
    this.knownRules.put( Constants.CNF_EMPTY_VALUE_RULE, EmptyValueProcessingRule.class );
    this.knownRules.put( Constants.CNF_VERSION_RESOLUTION_RULE, FormatVersionResolutionRule.class );
    this.knownRules.put( Constants.CNF_HTML_INFO_RULE, HtmlInfoProcessingRule.class );
    this.knownRules.put( Constants.CNF_INFER_DATE_RULE, InferDateFromFileNameRule.class );