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()

protected void processMessageReceived()
{
  try {
    try {
      ExampleJavaExtractorService extrServ = new ExampleJavaExtractorService(this);
      jobReceived = getRepresentation(messageReceived, ExtractionJob.class);System.out.println("STARTED : ");
        
    }

         File textFile = extrServ.processJob(jobReceived);
 @Override
        jobReceived.setFlag("wasText");    public void onFailed(String msg, Throwable e) {
      log.info("Word count extraction complete. Returning word count file as intermediate resultSystem.out.println("FAILED  : " + msg);
      sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "Word count extraction complete. Returning word count file as intermediate result.", log          e.printStackTrace();
      uploadIntermediate(textFile, "text/plain", log);
          textFileSystem.deleteexit(0);
        
    }

  sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "DONE.", log);
    } catch (Exception ioe   @Override
            public void onDone(double value) {
              log.error  System.out.println("CouldDONE  not finish extraction: job.", + ioevalue);
       sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "Could not finish extraction job.", log         System.exit(0);
      sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "DONE.", log);
      }

       }
  } catch(Exception e) {@Override
      e.printStackTrace();
      public void System.exit(1);onAborted(String msg) {
  } 
}

public File processJob(ExtractionJob receivedMsg) throws Exception 
{    
  log.info("Downloading text file with ID "+ receivedMsg.getIntermediateId() +" from " + receivedMsg.getHost());
  callingExtractor.sendStatus(receivedMsg.getId(), callingExtractor.getClass().getSimpleName(), "Downloading text file.", log);
      
  DefaultHttpClient httpclient = new DefaultHttpClient( System.out.println("ABORTED : " + msg);
                System.exit(0);
            }
        });
  HttpGet httpGet = new HttpGet(receivedMsg.getHost() +"api/files/"+ receivedMsg.getIntermediateId()+"?key="+playserverKey);    
  HttpResponse fileResponse = httpclient.execute(httpGet);
  log.info(fileResponse.getStatusLine());

  if(fileResponse.getStatusLine().toString().indexOf("200") == -1){
    throw new IOException("File not found.");
  }

  HttpEntity fileEntity = fileResponse.getEntity();
  InputStream fileIs = fileEntity.getContent();
  
  Header[] hdrs = fileResponse.getHeaders("content-disposition");
  String contentDisp = hdrs[0].toString();
  
  String fileName = 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 }

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;
    }

    @Override
    public String getName() {
        return "Text Document";
    }

    @Override
    public List<String> getSupportedMediaTypes() {
        List<String> mediaTypes = new ArrayList<String>();
      
  EntityUtilsmediaTypes.consume(fileEntityadd("text/*");
 
      
  log.info("Download complete. Initiating word count generation");
  return mediaTypes;
  File textFile = processFile(tempFile, receivedMsg.getId());  
  return textFile;       
}

private File processFile(File tempFile, String originalFileId) throws Exception
{
    Runtime r = Runtime.getRuntime();// ----------------------------------------------------------------------
    Process p;// HasText
     // Process tracks one external native process
      
    String tempDir = System.getProperty("java.io.tmpdir");

	if (new Character(tempDir.charAt(tempDir.length()-1)).toString().equals(System.getProperty("file.separator")) == false){
// ----------------------------------------------------------------------
    @Override
    public List<String> getWords() {
        if (words == null) {
            tempDirwords = tempDir + System.getProperty("file.separator"new ArrayList<String>();
      }
      try {
     String  processCmd = "";
    String operSystem = System.getProperty("os.name").toLowerCase();

	// TODO: windows impl
    if(operSystem.indexOf("nix") >= 0 || operSystem.indexOf("nux") >= 0 || operSystem.indexOf("aix") > 0 ){ BufferedReader br = new BufferedReader(new FileReader(file));
                String line;
      "wc -w " + tempDir +  tempFile.getName();
    }  while((line = br.readLine()) != null) {
      
     p = r.exec(processCmd, null, new File(tempDir));
    StreamGobblerString[] outputGobblerw = new StreamGobbler(p.getInputStream(), "INFO", log);
line.split(" ");
                StreamGobbler errorGobbler = new StreamGobblerwords.addAll(pArrays.getErrorStreamasList(w),"ERROR", log);
    outputGobbler.start();
    errorGobbler.start();
    p.waitFor();
      }
    File outFile = new File(tempDir + tempFile.getName().substring(0, tempFile.getName().lastIndexOf(".")) + ".txt");
    tempFilebr.deleteclose();

	if(!Files.exists(outFile.toPath()))
      throw new Exception("File not processed correctly. File} iscatch possibly corrupt.");
(IOException e) {
           
  	return outFile;  
}

public class TextHistogramExtractor implements Extractor
{
    @OverrideMethod that is called by the AMQP library every time it has data
    public Adapter *newAdapter() {
 available that should be sent to RabbitMQ. 
throw (new RuntimeException("Not supported."));
   * }

 @param  connection @Override
 pointer to the mainpublic connectionString objectgetName()  {
      *  @paramreturn "Text dataHistogram Extractor";
    }

   memory buffer@Override
 with the data thatpublic shouldSet<Class<? beextends sent to RabbitMQAdapter>> supportedAdapters() {
      *  @paramSet<Class<? extends sizeAdapter>> adapters = new HashSet<Class<? extends Adapter>>();
  size of the buffer
      */adapters.add(HasText.class);
     virtual void onData(AMQP::Connection *connection, const char *data, size_t size) return adapters;
    }

     {@Override
    public Class<? extends Descriptor>  // @todo 
getFeatureType() {
        return LabelHistogramDescriptor.class;
  //  }

 Add your own implementation,@Override
 for example by doingpublic aDescriptor call to the
  extract(Adapter adapter) throws Exception {
       // if send()adapter systeminstanceof call. But be aware that the send() call may not
HasText) {
            LabelHistogramDescriptor desc = new  //LabelHistogramDescriptor();

   send all data at once, so you also need for to(String takeword care of buffering
: ((HasText) adapter).getWords()) {
           //  the bytes that could not immediately be sent, and try to send 
    desc.increaseBin(word);
            }

      //  them again when the return socketdesc;
 becomes writable again
     } else {

      /**
      *throw new MethodUnsupportedTypeException();
 that is called by the AMQP library when}
 the login attempt }
    
  *  succeeded.@Override
 After this method haspublic been called, the connection is ready boolean hasPreview(){
      *  toreturn use.false;
      *}
  @param  connection      The connection that can now be used
      */@Override
    public String virtual void onConnected(Connection *connection)
      previewName(){
         // @todoreturn null;
    }
}

public class LabelHistogramEuclidianDistanceMeasure implements Measure 
{
    @Override
    public SimilarityPercentage normalize(Similarity similarity) {
     //  add your own implementation, for example by creating a channel 
         //  instance, andreturn startnew publishingSimilarityPercentage(1 or consuming- similarity.getValue());
      }

      /**@Override
    public String *getFeatureType() {
 Method that is called by the AMQP library when a fatal error occursreturn LabelHistogramDescriptor.class.getName();
    }

    @Override
  *  onpublic the connection, for example because data received from RabbitMQ
      *  could not be recognized.String getName() {
        return "Histogram Distance";
    }

    @Override
    public Class<LabelHistogramEuclidianDistanceMeasure> *getType() {
 @param  connection     return TheLabelHistogramEuclidianDistanceMeasure.class;
 connection on which the error occured }

    // correlation

 *  @param @Override
 message   public Similarity compare(Descriptor desc1, Descriptor desc2) Athrows human readable error messageException {
      */
  if ((desc1 instanceof LabelHistogramDescriptor) virtual&& void onError(Connectiondesc2 *connection, const std::string &message)
instanceof LabelHistogramDescriptor)) {
       {
     LabelHistogramDescriptor lhd1 = // @todo(LabelHistogramDescriptor) desc1;
        //   add yourLabelHistogramDescriptor ownlhd2 implementation, for example by reporting the error
= (LabelHistogramDescriptor) desc2;

            // get toall thepossible userlabels
 of your program, log the error, and destruct the 
  Set<String> labels = new HashSet<String>();
  //  connection object because it is no longer in a usable state
 labels.addAll(lhd1.getLabels());
        }
  };

}

namespace CPPExample 
{ labels.addAll(lhd2.getLabels());

  /**
   *  Parse data that was recevied from// RabbitMQnormalize
   *  
   *  Every time that data comes in from RabbitMQ, you should call this method to parse lhd1.normalize();
            lhd2.normalize();
   *  the incoming data, and let it handle by
 the AMQP-CPP library. This method returns the number
   * // ofcompute bytesdistance
 that were processed.
   *
   *  If notdouble allsum bytes could be processed because it only contained a partial frame, you should
   *  call this same method later 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= 0;

            for (String s : labels) {
                Double b1 = lhd1.getBin(s);
                Double b2 = lhd2.getBin(s);
             
         buffer to decode
   *  if @param(b1 == sizenull) {
       size of the buffer to decode
   *  @return   sum += b2 * b2;
        number of bytes that were processed
   */
} else size_tif parse(char *buffer, size_t size)
  {
b2 == null) {
                    sum return _implementation.parse(buffer, size);
  }
}