opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/SnapshotTracker.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft.behaviors;
  10
  11 import akka.event.LoggingAdapter;
  12 import com.google.common.base.Optional;
  13 import com.google.protobuf.ByteString;
  14
  15 /**
  16  * SnapshotTracker does house keeping for a snapshot that is being installed in chunks on the Follower
  17  */
  18 public class SnapshotTracker {
  19     private final LoggingAdapter LOG;
  20     private final int totalChunks;
  21     private ByteString collectedChunks = ByteString.EMPTY;
  22     private int lastChunkIndex = AbstractLeader.FIRST_CHUNK_INDEX - 1;
  23     private boolean sealed = false;
  24     private int lastChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
  25
  26     SnapshotTracker(LoggingAdapter LOG, int totalChunks){
  27         this.LOG = LOG;
  28         this.totalChunks = totalChunks;
  29     }
  30
  31     /**
  32      * Adds a chunk to the tracker
  33      *
  34      * @param chunkIndex
  35      * @param chunk
  36      * @return true when the lastChunk is received
  37      * @throws InvalidChunkException
  38      */
  39     boolean addChunk(int chunkIndex, ByteString chunk, Optional<Integer> lastChunkHashCode) throws InvalidChunkException{
  40         if(sealed){
  41             throw new InvalidChunkException("Invalid chunk received with chunkIndex " + chunkIndex + " all chunks already received");
  42         }
  43
  44         if(lastChunkIndex + 1 != chunkIndex){
  45             throw new InvalidChunkException("Expected chunkIndex " + (lastChunkIndex + 1) + " got " + chunkIndex);
  46         }
  47
  48         if(lastChunkHashCode.isPresent()){
  49             if(lastChunkHashCode.get() != this.lastChunkHashCode){
  50                 throw new InvalidChunkException("The hash code of the recorded last chunk does not match " +
  51                         "the senders hash code expected " + lastChunkHashCode + " was " + lastChunkHashCode.get());
  52             }
  53         }
  54
  55         if(LOG.isDebugEnabled()) {
  56             LOG.debug("Chunk={},collectedChunks.size:{}",
  57                     chunkIndex, collectedChunks.size());
  58         }
  59
  60         sealed = (chunkIndex == totalChunks);
  61         lastChunkIndex = chunkIndex;
  62         collectedChunks = collectedChunks.concat(chunk);
  63         this.lastChunkHashCode = chunk.hashCode();
  64         return sealed;
  65     }
  66
  67     byte[] getSnapshot(){
  68         if(!sealed) {
  69             throw new IllegalStateException("lastChunk not received yet");
  70         }
  71
  72         return collectedChunks.toByteArray();
  73     }
  74
  75     ByteString getCollectedChunks(){
  76         return collectedChunks;
  77     }
  78
  79     public static class InvalidChunkException extends Exception {
  80         InvalidChunkException(String message){
  81             super(message);
  82         }
  83     }
  84
  85 }