2 * Copyright (c) 2015 Brocade Communications 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.utils;
11 import akka.dispatch.Futures;
12 import akka.japi.Option;
13 import akka.persistence.SelectedSnapshot;
14 import akka.persistence.SnapshotMetadata;
15 import akka.persistence.SnapshotSelectionCriteria;
16 import akka.persistence.snapshot.japi.SnapshotStore;
17 import com.google.common.collect.Iterables;
18 import com.google.common.collect.Lists;
19 import com.google.common.util.concurrent.Uninterruptibles;
20 import java.util.ArrayList;
21 import java.util.Collections;
22 import java.util.Iterator;
23 import java.util.List;
25 import java.util.concurrent.ConcurrentHashMap;
26 import java.util.concurrent.CountDownLatch;
27 import java.util.concurrent.TimeUnit;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30 import scala.concurrent.Future;
33 * An akka SnapshotStore implementation that stores data in memory. This is intended for testing.
35 * @author Thomas Pantelis
37 public class InMemorySnapshotStore extends SnapshotStore {
39 static final Logger LOG = LoggerFactory.getLogger(InMemorySnapshotStore.class);
41 private static Map<String, List<StoredSnapshot>> snapshots = new ConcurrentHashMap<>();
42 private static final Map<String, CountDownLatch> snapshotSavedLatches = new ConcurrentHashMap<>();
44 public static void addSnapshot(String persistentId, Object snapshot) {
45 List<StoredSnapshot> snapshotList = snapshots.get(persistentId);
47 if(snapshotList == null) {
48 snapshotList = new ArrayList<>();
49 snapshots.put(persistentId, snapshotList);
52 synchronized (snapshotList) {
53 snapshotList.add(new StoredSnapshot(new SnapshotMetadata(persistentId, snapshotList.size(),
54 System.currentTimeMillis()), snapshot));
58 @SuppressWarnings("unchecked")
59 public static <T> List<T> getSnapshots(String persistentId, Class<T> type) {
60 List<StoredSnapshot> stored = snapshots.get(persistentId);
62 return Collections.emptyList();
66 synchronized (stored) {
67 retList = Lists.newArrayListWithCapacity(stored.size());
68 for(StoredSnapshot s: stored) {
69 if(type.isInstance(s.getData())) {
70 retList.add((T) s.getData());
78 public static void clear() {
82 public static void addSnapshotSavedLatch(String persistenceId) {
83 snapshotSavedLatches.put(persistenceId, new CountDownLatch(1));
86 public static <T> T waitForSavedSnapshot(String persistenceId, Class<T> type) {
87 if(!Uninterruptibles.awaitUninterruptibly(snapshotSavedLatches.get(persistenceId), 5, TimeUnit.SECONDS)) {
88 throw new AssertionError("Snapshot was not saved");
91 return getSnapshots(persistenceId, type).get(0);
95 public Future<Option<SelectedSnapshot>> doLoadAsync(String s,
96 SnapshotSelectionCriteria snapshotSelectionCriteria) {
97 List<StoredSnapshot> snapshotList = snapshots.get(s);
98 if(snapshotList == null){
99 return Futures.successful(Option.<SelectedSnapshot>none());
102 StoredSnapshot snapshot = Iterables.getLast(snapshotList);
103 SelectedSnapshot selectedSnapshot =
104 new SelectedSnapshot(snapshot.getMetadata(), snapshot.getData());
105 return Futures.successful(Option.some(selectedSnapshot));
109 public Future<Void> doSaveAsync(SnapshotMetadata snapshotMetadata, Object o) {
110 List<StoredSnapshot> snapshotList = snapshots.get(snapshotMetadata.persistenceId());
112 if(snapshotList == null){
113 snapshotList = new ArrayList<>();
114 snapshots.put(snapshotMetadata.persistenceId(), snapshotList);
116 synchronized (snapshotList) {
117 snapshotList.add(new StoredSnapshot(snapshotMetadata, o));
120 CountDownLatch latch = snapshotSavedLatches.get(snapshotMetadata.persistenceId());
125 return Futures.successful(null);
129 public void onSaved(SnapshotMetadata snapshotMetadata) throws Exception {
133 public void doDelete(SnapshotMetadata snapshotMetadata) throws Exception {
134 List<StoredSnapshot> snapshotList = snapshots.get(snapshotMetadata.persistenceId());
136 if(snapshotList == null){
140 int deleteIndex = -1;
142 synchronized (snapshotList) {
143 for(int i=0;i<snapshotList.size(); i++){
144 StoredSnapshot snapshot = snapshotList.get(i);
145 if(snapshotMetadata.equals(snapshot.getMetadata())){
151 if(deleteIndex != -1){
152 snapshotList.remove(deleteIndex);
158 public void doDelete(String persistentId, SnapshotSelectionCriteria snapshotSelectionCriteria)
160 List<StoredSnapshot> snapshotList = snapshots.get(persistentId);
162 if(snapshotList == null){
166 synchronized (snapshotList) {
167 Iterator<StoredSnapshot> iter = snapshotList.iterator();
168 while(iter.hasNext()) {
169 StoredSnapshot s = iter.next();
170 LOG.trace("doDelete: sequenceNr: {}, maxSequenceNr: {}", s.getMetadata().sequenceNr(),
171 snapshotSelectionCriteria.maxSequenceNr());
173 if(s.getMetadata().sequenceNr() <= snapshotSelectionCriteria.maxSequenceNr()) {
180 private static class StoredSnapshot {
181 private final SnapshotMetadata metadata;
182 private final Object data;
184 private StoredSnapshot(SnapshotMetadata metadata, Object data) {
185 this.metadata = metadata;
189 public SnapshotMetadata getMetadata() {
193 public Object getData() {