private void emitTriggerIfAllBlocksReceived(FileMetadata fileMetadata,Map<Long,BlockMetadata.FileBlockMetadata> receivedBlocksMetadata){
  String filePath=fileMetadata.getFilePath();
  if (receivedBlocksMetadata.size() != fileMetadata.getNumberOfBlocks()) {
    fileMetadataMap.put(filePath,fileMetadata);
  }
 else {
    Set<Long> receivedBlocks=receivedBlocksMetadata.keySet();
    boolean blockMissing=false;
    if (!fileMetadata.isDirectory()) {
      for (      long blockId : fileMetadata.getBlockIds()) {
        if (!receivedBlocks.contains(blockId)) {
          blockMissing=true;
        }
      }
    }
    if (!blockMissing) {
      long fileProcessingTime=System.currentTimeMillis() - fileMetadata.getDiscoverTime();
      List<StitchBlock> outputBlocks=constructOutputBlockMetadataList(fileMetadata);
      OutputFileMetadata outputFileMetadata=new OutputFileMetadata(fileMetadata,outputBlocks);
      trigger.emit(outputFileMetadata);
      LOG.debug("Total time taken to process the file {} is {} ms",fileMetadata.getFilePath(),fileProcessingTime);
      fileMetadataMap.remove(filePath);
    }
  }
}

private void emitTriggerIfAllBlocksReceived ( FileMetadata fileMetadata , Map < Long , BlockMetadata . FileBlockMetadata > receivedBlocksMetadata ) { String filePath = fileMetadata . getFilePath ( ) ; if ( receivedBlocksMetadata . size ( ) != fileMetadata . getNumberOfBlocks ( ) ) { fileMetadataMap . put ( filePath , fileMetadata ) ; } else { Set < Long > receivedBlocks = receivedBlocksMetadata . keySet ( ) ; boolean blockMissing = false ; if ( ! fileMetadata . isDirectory ( ) ) { for ( long blockId : fileMetadata . getBlockIds ( ) ) { if ( ! receivedBlocks . contains ( blockId ) ) { blockMissing = true ; } } } if ( ! blockMissing ) { long fileProcessingTime = System . currentTimeMillis ( ) - fileMetadata . getDiscoverTime ( ) ; List < StitchBlock > outputBlocks = constructOutputBlockMetadataList ( fileMetadata ) ; OutputFileMetadata outputFileMetadata = new OutputFileMetadata ( fileMetadata , outputBlocks ) ; trigger . emit ( outputFileMetadata ) ; LOG . debug ( "Total time taken to process the file {} is {} ms" , fileMetadata . getFilePath ( ) , fileProcessingTime ) ; fileMetadataMap . remove ( filePath ) ; } } }
checks if all blocks for given file are received . sends triggger when all blocks are received .
