package tachyon.worker.block;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import tachyon.worker.BlockStoreLocation;

/**
 * Represents the data the BlockWorker will send to the master in its periodic heartbeat.
 */
public class BlockHeartbeatReporter implements BlockMetaEventListener {

  private List<Long> mRemovedBlocks;
  private Map<Long, List<Long>> mAddedBlocks;

  public BlockHeartbeatReporter() {
    mRemovedBlocks = new ArrayList<Long>(100);
    mAddedBlocks = new HashMap<Long, List<Long>>();
  }

  public void preCommitBlock(long userId, long blockId) {
    // Do nothing
  }

  public void postCommitBlock(long userId, long blockId) {

  }

  public void preMoveBlock(long userId, long blockId, BlockStoreLocation newLocation) {
    // Do nothing
  }

  public void postMoveBlock(long userId, long blockId, BlockStoreLocation newLocation) {

  }

  public void preRemoveBlock(long userId, long blockId) {
    // Do nothing
  }

  public void postRemoveBlock(long userId, long blockId) {

  }
}

@@ -124,16 +124,57 @@
 l.Map;%0A%0A
+import com.google.common.collect.Lists;%0A%0A
 import t
@@ -301,16 +301,46 @@
 artbeat.
+ This%0A * class is thread safe.
 %0A */%0Apub
@@ -407,16 +407,46 @@
 stener %7B
+%0A  private final Object mLock;
 %0A%0A  priv
@@ -555,24 +555,50 @@
 eporter() %7B%0A
+    mLock = new Object();%0A
     mRemoved
@@ -685,24 +685,36 @@
 g%3E%3E();%0A  %7D%0A%0A
+  @Override%0A
   public voi
@@ -747,32 +747,61 @@
 Id, long blockId
+, BlockStoreLocation location
 ) %7B%0A    // Do no
@@ -803,32 +803,44 @@
 Do nothing%0A  %7D%0A%0A
+  @Override%0A
   public void po
@@ -878,25 +878,202 @@
  blockId
-) %7B%0A%0A  %7D%0A
+, BlockStoreLocation location) %7B%0A    Long storageDirId = location.getStorageDirId();%0A    synchronized (mLock) %7B%0A      addBlockToAddedBlocks(blockId, storageDirId);%0A    %7D%0A  %7D%0A%0A  @Override
 %0A  publi
@@ -1168,32 +1168,44 @@
 Do nothing%0A  %7D%0A%0A
+  @Override%0A
   public void po
@@ -1277,21 +1277,403 @@
 tion) %7B%0A
-%0A  %7D%0A
+    Long storageDirId = newLocation.getStorageDirId();%0A    synchronized (mLock) %7B%0A      // Remove the block from our list of added blocks in this heartbeat, if it was added to%0A      // prevent adding the block twice.%0A      removeBlockFromAddedBlocks(blockId);%0A      // Add the block back with the new storagedir.%0A      addBlockToAddedBlocks(blockId, storageDirId);%0A    %7D%0A  %7D%0A%0A  @Override
 %0A  publi
@@ -1746,16 +1746,28 @@
 ng%0A  %7D%0A%0A
+  @Override%0A
   public
@@ -1817,15 +1817,1086 @@
 ckId) %7B%0A
+    synchronized (mLock) %7B%0A      // Remove the block from list of added blocks, in case it was added in this heartbeat period.%0A      removeBlockFromAddedBlocks(blockId);%0A      // Add to the list of removed blocks in this heartbeat period.%0A      mRemovedBlocks.add(blockId);%0A    %7D%0A  %7D%0A%0A  /**%0A   * Adds a block to the list of added blocks in this heartbeat period.%0A   * @param blockId The id of the block to add%0A   * @param storageDirId The storage directory id containing the block%0A   */%0A  private void addBlockToAddedBlocks(long blockId, long storageDirId) %7B%0A    if (mAddedBlocks.containsKey(storageDirId)) %7B%0A      mAddedBlocks.get(storageDirId).add(blockId);%0A    %7D else %7B%0A      mAddedBlocks.put(storageDirId, Lists.newArrayList(blockId));%0A    %7D%0A  %7D%0A%0A  /**%0A   * Removes the block from the added blocks map, if it exists.%0A   * @param blockId The block to remove%0A   */%0A  private void removeBlockFromAddedBlocks(long blockId) %7B%0A    for (List%3CLong%3E blockList : mAddedBlocks.values()) %7B%0A      if (blockList.contains(blockId)) %7B%0A        blockList.remove(blockId);%0A      %7D%0A    %7D
 %0A  %7D%0A%7D%0A

Implement meta listener callbacks in block heartbeat reporter.