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
        
    public void jobReceived.setFlag("wasText");

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.", loge.printStackTrace();
        
      uploadIntermediate(textFile, "text/plain", log System.exit(0);
        
    }

  textFile.delete();
          @Override
      sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "DONE.", log);

    }public catchvoid onDone(Exceptiondouble ioevalue) {
      log.error("Could not finish extraction job.", ioe);
      sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "Could not finish extraction job.", logSystem.out.println("DONE    : " + value);
      sendStatus(jobReceived.getId(), this.getClass().getSimpleName(), "DONE.", log);
    }
  } catch(Exception e) {System.exit(0);
      e.printStackTrace();
      System.exit(1);
  } 
}

public File processJob(ExtractionJob receivedMsg) throws Exception}

            @Override
            public void onAborted(String msg) {
      
    log.info("Downloading text file with ID "+ receivedMsgSystem.out.getIntermediateIdprintln() +"ABORTED from: " + receivedMsg.getHost(msg));
  callingExtractor.sendStatus(receivedMsg.getId(), callingExtractor.getClass().getSimpleName(), "Downloading text file.", log);
           
  DefaultHttpClient httpclient = new DefaultHttpClient( 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)  }

public class TextAdapter implements FileLoader, HasText {
    throw new IOException("private File not found.");
  }
  HttpEntity fileEntity = fileResponse.getEntity()file;
  InputStream  fileIsprivate = fileEntity.getContent()List<String> words;
  
  Header[] hdrs =public fileResponse.getHeaders("content-disposition");TextAdapter() {}

  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();
      
  EntityUtils.consume(fileEntity); // ----------------------------------------------------------------------
    // FileLoader
    // ----------------------------------------------------------------------
    @Override
    public void load(File file) {
        this.file = file;
    }

    @Override
    public String getName() {
      
  
return  log.info("Download complete. Initiating word count generation");
  
  File textFile = processFile(tempFile, receivedMsg.getId());  
  return textFile;"Text Document";
    }

    @Override
    public List<String> getSupportedMediaTypes() {
        List<String>  
}

private File processFile(File tempFile, String originalFileId) throws Exception  {
    
    Runtime r = Runtime.getRuntime(mediaTypes = new ArrayList<String>();
        mediaTypes.add("text/*");
    Process p;     // Process tracks one external native processreturn mediaTypes;
      
    String tempDir = System.getProperty("java.io.tmpdir");}

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

	// TODO: windows impl
        if(operSystem.indexOf("nix") >= 0 || operSystem.indexOf("nux") >= 0 || operSystem.indexOf("aix") > 0 ){
 try {
              "wc -w "BufferedReader +br tempDir= +new  tempFile.getName()BufferedReader(new FileReader(file));
    }       
     String line;
      p = r.exec(processCmd, null, new File(tempDir));
    StreamGobbler outputGobblerwhile((line = new StreamGobbler(p.getInputStreambr.readLine(), "INFO", log);) != null) {
    StreamGobbler   errorGobbler = new StreamGobbler(p.getErrorStream(),"ERROR", log);
    outputGobbler.start();
    errorGobbler.start();
    p.waitFor( String[] w = line.split(" ");
        
     File outFile = new File(tempDir + tempFile.getName().substring(0, tempFile.getName().lastIndexOf(".")) + ".txt") words.addAll(Arrays.asList(w));
    tempFile.delete            }
                br.close();
     if(!Files.exists(outFile.toPath()))
       } catch (IOException e) {
       throw new Exception("File not processed correctly. File is possibly corrupte."printStackTrace();
      
     return outFile;}
      
}

public class TextHistogramExtractor implements Extractor
{
    @Override     *  Method 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 sentAdapter>> tosupportedAdapters() RabbitMQ{
      *  @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, for example by@Override
 doing a call topublic the
Descriptor extract(Adapter adapter) throws Exception {
    //  send() system call.if But(adapter beinstanceof awareHasText) that{
 the send() call may not
       LabelHistogramDescriptor desc //=  send all data at once, so you also need to take care of buffering
    new LabelHistogramDescriptor();

            for (String word : ((HasText) adapter).getWords()) {
     //  the bytes that could not immediately be sent, and try to send 
  desc.increaseBin(word);
            }

   //  them again when the socket becomes writable return againdesc;
      }

  } else {
  /**
      *  Method that isthrow called by the AMQP library when the login attempt new UnsupportedTypeException();
        }
    }
  *  succeeded.
 After this method has@Override
 been called, the connectionpublic 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     //  add your own implementation, for example by creating a channel 
    public SimilarityPercentage normalize(Similarity similarity) {
 //   instance,  and start publishingreturn or consuming
  new SimilarityPercentage(1 - 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 LabelHistogramEuclidianDistanceMeasure.class;
   The connection}

 on which the error// occuredcorrelation

    @Override
    public *Similarity compare(Descriptor @paramdesc1, Descriptor messagedesc2) throws Exception {
      A human readableif error message
      */((desc1 instanceof LabelHistogramDescriptor) && (desc2 instanceof LabelHistogramDescriptor)) {
      virtual  void onError(Connection *connection, const std::string &message)
      {LabelHistogramDescriptor lhd1 = (LabelHistogramDescriptor) desc1;
        // @todo
   LabelHistogramDescriptor lhd2 = (LabelHistogramDescriptor)  //  add your own implementation, for example by reporting the error
        //  to the user of your program, log the error, and destruct the desc2;

            // get all possible labels
            Set<String> labels = new HashSet<String>();
        //  connection object because it is no longer in a usable state
      }
  };

}

namespace CPPExample {

  /**
   *  Parse data that was recevied from RabbitMQ
   *  
   *  Every time that data comes in from RabbitMQ, you should call this method to parse
   *  the incoming data, and let it handle by the AMQP-CPP library. This method returns the number
   *  of bytes that were processed.
   *
   *  If not all 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      buffer to decode
   *  @param  size 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) {
               size of the buffer to decode
sum += b2 * b2;
   @return             number} else ofif bytes(b2 that== werenull) processed{
        */
   size_t parse(char *buffer, size_t size)
  {
   sum += return _implementation.parse(buffer, size);
  }
}