2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.raft.behaviors;
11 import com.google.common.base.Optional;
12 import com.google.common.base.Preconditions;
13 import com.google.common.io.ByteSource;
14 import com.google.common.io.CountingOutputStream;
15 import java.io.ByteArrayOutputStream;
16 import java.io.IOException;
17 import java.util.Arrays;
18 import org.slf4j.Logger;
21 * Helper class that maintains state for a snapshot that is being installed in chunks on a Follower.
23 class SnapshotTracker implements AutoCloseable {
24 private final Logger log;
25 private final int totalChunks;
26 private final String leaderId;
27 private final CountingOutputStream countingStream;
28 private final ByteArrayOutputStream backingStream;
29 private int lastChunkIndex = LeaderInstallSnapshotState.FIRST_CHUNK_INDEX - 1;
30 private boolean sealed = false;
31 private int lastChunkHashCode = LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE;
33 SnapshotTracker(Logger log, int totalChunks, String leaderId) {
35 this.totalChunks = totalChunks;
36 this.leaderId = Preconditions.checkNotNull(leaderId);
38 backingStream = new ByteArrayOutputStream();
39 countingStream = new CountingOutputStream(backingStream);
43 * Adds a chunk to the tracker.
45 * @param chunkIndex the index of the chunk
46 * @param chunk the chunk data
47 * @param lastChunkHashCode the optional hash code for the chunk
48 * @return true if this is the last chunk is received
49 * @throws InvalidChunkException if the chunk index is invalid or out of order
51 boolean addChunk(int chunkIndex, byte[] chunk, Optional<Integer> maybeLastChunkHashCode)
52 throws InvalidChunkException, IOException {
53 log.debug("addChunk: chunkIndex={}, lastChunkIndex={}, collectedChunks.size={}, lastChunkHashCode={}",
54 chunkIndex, lastChunkIndex, countingStream.getCount(), this.lastChunkHashCode);
57 throw new InvalidChunkException("Invalid chunk received with chunkIndex " + chunkIndex
58 + " all chunks already received");
61 if (lastChunkIndex + 1 != chunkIndex) {
62 throw new InvalidChunkException("Expected chunkIndex " + (lastChunkIndex + 1) + " got " + chunkIndex);
65 if (maybeLastChunkHashCode.isPresent() && maybeLastChunkHashCode.get() != this.lastChunkHashCode) {
66 throw new InvalidChunkException("The hash code of the recorded last chunk does not match "
67 + "the senders hash code, expected " + this.lastChunkHashCode + " was "
68 + maybeLastChunkHashCode.get());
71 countingStream.write(chunk);
73 sealed = chunkIndex == totalChunks;
74 lastChunkIndex = chunkIndex;
75 this.lastChunkHashCode = Arrays.hashCode(chunk);
79 ByteSource getSnapshotBytes() {
81 throw new IllegalStateException("lastChunk not received yet");
84 return ByteSource.wrap(backingStream.toByteArray());
87 String getLeaderId() {
94 countingStream.close();
95 } catch (IOException e) {
96 log.warn("Error closing snapshot stream");
100 public static class InvalidChunkException extends IOException {
101 private static final long serialVersionUID = 1L;
103 InvalidChunkException(String message) {