static public void main(String[] args) {         
        PairwiseComparison comparison = new PairwiseComparison();
        comparison.setId(UUID.randomUUID().toString());
        comparison.setFirstDataset(new File("data/test1.txt"));
        comparison.setSecondDataset(new File("data/test2.txt"));
        comparison.setAdapterId(TextAdapter.class.getName());
        comparison.setExtractorId(TextHistogramExtractor.class.getName());
        comparison.setMeasureId(LabelHistogramEuclidianDistanceMeasure.class.getName());

        ExecutionEngine ee = new ExecutionEngine(); 
        ee.submit(comparison, new ComparisonStatusHandler() {
            @Override
            public void onStarted() {
                System.out.println("STARTED : "

public class TextAdapter implements FileLoader, HasText {
    private File         file;
    private List<String> words;

    public TextAdapter() {}

    // ----------------------------------------------------------------------
    // FileLoader
    // ----------------------------------------------------------------------
    @Override
    public void load(File file) {
         //this.file = file;
  send all data}

 at once, so you@Override
 also need to takepublic careString ofgetName() buffering{
        return //  the bytes that could not immediately be sent, and try to send 
  "Text Document";
    }

    @Override
    public List<String> getSupportedMediaTypes() {
       // List<String> themmediaTypes again= when the socket becomes writable againnew ArrayList<String>();
     }

      mediaTypes.add("text/**");
      *  Methodreturn that is called by the AMQP library when the login attempt mediaTypes;
    }

      *  succeeded. After this method has been called, the connection is ready // ----------------------------------------------------------------------
      *  to use.// HasText
      *  @param  connection// ----------------------------------------------------------------------
    @Override
    public  The connection that can now be used
      */List<String> getWords() {
        if (words == null) {
      virtual void onConnected(Connection *connection)
   words = new {ArrayList<String>();
         // @todo
  try {
      //  add your own implementation, for example by creating aBufferedReader channelbr 
= new BufferedReader(new FileReader(file));
      //  instance, and start publishing or consuming
   String line;
  }

      /**
      *  Method that is called by the AMQP library when a fatal error occurs
while((line = br.readLine()) != null) {
             *  on the connection, for example becauseString[] dataw received from RabbitMQ
= line.split(" ");
       *  could not be recognized.
      *  @paramwords.addAll(Arrays.asList(w));
  connection      The connection on which the error occured
  }
    *  @param  message         A human readable error messagebr.close();
      */
      virtual} voidcatch onError(Connection *connection, const std::string &message)
IOException e) {
       {
        // @todo e.printStackTrace();
        //   add your}
 own implementation, for example by reporting the error}
        //return words;
 to the user of your program, log the error, and destruct the 
        //  connection object because it is no longer in a usable state
      }
  };

}

public class TextHistogramExtractor implements Extractor

public File processJob(ExtractionJob receivedMsg) throws Exception{
    @Override
  
  log.info("Downloading text file with ID "+ receivedMsg.getIntermediateId() +" from " + receivedMsg.getHost());
  callingExtractor.sendStatus(receivedMsg.getId(), callingExtractor.getClass().getSimpleName(), "Downloading text file.", logpublic Adapter newAdapter() {
        throw (new RuntimeException("Not supported."));
    }

    @Override
  DefaultHttpClient httpclient =public newString DefaultHttpClientgetName(); {
  HttpGet   httpGet = new HttpGet(receivedMsg.getHost() +"api/files/"+ receivedMsg.getIntermediateId()+"?key="+playserverKey); return "Text Histogram Extractor";
    }

  HttpResponse fileResponse = httpclient.execute(httpGet);
  log.info(fileResponse.getStatusLine());
  if(fileResponse.getStatusLine().toString().indexOf("200") == -1){
    throw new IOException("File not found."  @Override
    public Set<Class<? extends Adapter>> supportedAdapters() {
        Set<Class<? extends Adapter>> adapters = new HashSet<Class<? extends Adapter>>();
  }
   HttpEntity fileEntity = fileResponseadapters.getEntityadd(HasText.class);
     InputStream  fileIs = fileEntity.getContent()return adapters;
    }

  Header[] hdrs = fileResponse.getHeaders("content-disposition"); @Override
  String contentDisp = hdrs[0].toString();public Class<? extends Descriptor> getFeatureType() {
  
    String fileName =return contentDisp.substring(contentDisp.indexOf("filename=")+9);
  File tempFile = File.createTempFile(fileName.substring(0, fileName.lastIndexOf(".")),      fileName.substring(fileName.lastIndexOf(".")).toLowerCase());
  OutputStream fileOs = new FileOutputStream(tempFile);   
  IOUtils.copy(fileIs,fileOs);
  fileIs.close();
  fileOs.close();
      
  EntityUtils.consume(fileEntity); LabelHistogramDescriptor.class;
    }

    @Override
    public Descriptor extract(Adapter adapter) throws Exception {
        if (adapter instanceof HasText) {
            LabelHistogramDescriptor desc = new LabelHistogramDescriptor();

      
  
  log.info("Download complete. Initiating  for (String word count generation");
  
  File textFile = processFile(tempFile, receivedMsg.getId());  
  return textFile;: ((HasText) adapter).getWords()) {
                desc.increaseBin(word);
            
}

namespace CPPExample {

  /**
   *  Parse data that was recevied from RabbitMQ return desc;
   *  
   *} else Every{
 time that data comes in from RabbitMQ, you should call this methodthrow to parsenew UnsupportedTypeException();
   *  the incoming data, and}
 let it handle by}
 the AMQP-CPP library. This
 method returns the number@Override
   * public of bytes that were processed.
boolean hasPreview(){
      *
  return *false;
  If not all}
 bytes could be processed
 because it only contained@Override
  a partial frame,public you shouldString previewName(){
   *  call this same methodreturn laternull;
 on when more data is available. The AMQP-CPP library does not do
   *  any buffering, so it is up to the caller to ensure that the old data is also passed in that
   *  later call.
   *
   *  @param  buffer      buffer to decode
   *  @param  size        size of the buffer to decode
   *  @return             number of bytes that were processed
   */
  size_t parse(char *buffer, size_t size)
  {
     return _implementation.parse(buffer, size);
  }
}

public class LabelHistogramEuclidianDistanceMeasure implements Measure 
{
    @Override
    public SimilarityPercentage normalize(Similarity similarity) {
        return new SimilarityPercentage(1 - similarity.getValue());
    }

    @Override
    public String getFeatureType() {
        return LabelHistogramDescriptor.class.getName();
    }

    @Override
    public String getName() {
        return "Histogram Distance";
    }

    @Override
    public Class<LabelHistogramEuclidianDistanceMeasure> getType() {
        return LabelHistogramEuclidianDistanceMeasure.class;
    }

    // correlation

    @Override
    public Similarity compare(Descriptor desc1, Descriptor desc2) throws Exception {
        if ((desc1 instanceof LabelHistogramDescriptor) && (desc2 instanceof LabelHistogramDescriptor)) {
            LabelHistogramDescriptor lhd1 = (LabelHistogramDescriptor) desc1;
            LabelHistogramDescriptor lhd2 = (LabelHistogramDescriptor) desc2;

            // get all possible labels
            Set<String> labels = new HashSet<String>();
            labels.addAll(lhd1.getLabels());
            labels.addAll(lhd2.getLabels());

            // normalize
            lhd1.normalize();
            lhd2.normalize();
            
            // compute distance
            double sum = 0;

            for (String s : labels) {
                Double b1 = lhd1.getBin(s);
                Double b2 = lhd2.getBin(s);
             
                if (b1 == null) {
                    sum += b2 * b2;
                } else if (b2 == null) {
                    sum += b1 * b1;
                } else {
                    sum += (b1 - b2) * (b1 - b2);
                }
            }

            return new SimilarityNumber(Math.sqrt(sum), 0, 1, 0);
        } else {
            throw new UnsupportedTypeException();
        }
    }