opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/InMemorySnapshotStore.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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.utils;
  10
  11 import akka.dispatch.Futures;
  12 import akka.persistence.SelectedSnapshot;
  13 import akka.persistence.SnapshotMetadata;
  14 import akka.persistence.SnapshotSelectionCriteria;
  15 import akka.persistence.snapshot.japi.SnapshotStore;
  16 import com.google.common.collect.Lists;
  17 import com.google.common.util.concurrent.Uninterruptibles;
  18 import java.util.ArrayList;
  19 import java.util.Collections;
  20 import java.util.Iterator;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Optional;
  24 import java.util.concurrent.ConcurrentHashMap;
  25 import java.util.concurrent.CountDownLatch;
  26 import java.util.concurrent.TimeUnit;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29 import scala.concurrent.Future;
  30
  31 /**
  32  * An akka SnapshotStore implementation that stores data in memory. This is intended for testing.
  33  *
  34  * @author Thomas Pantelis
  35  */
  36 public class InMemorySnapshotStore extends SnapshotStore {
  37
  38     static final Logger LOG = LoggerFactory.getLogger(InMemorySnapshotStore.class);
  39
  40     private static final Map<String, CountDownLatch> SNAPSHOT_SAVED_LATCHES = new ConcurrentHashMap<>();
  41     private static final Map<String, CountDownLatch> SNAPSHOT_DELETED_LATCHES = new ConcurrentHashMap<>();
  42     private static Map<String, List<StoredSnapshot>> snapshots = new ConcurrentHashMap<>();
  43
  44     public static void addSnapshot(String persistentId, Object snapshot) {
  45         List<StoredSnapshot> snapshotList = snapshots.get(persistentId);
  46
  47         if (snapshotList == null) {
  48             snapshotList = new ArrayList<>();
  49             snapshots.put(persistentId, snapshotList);
  50         }
  51
  52         synchronized (snapshotList) {
  53             snapshotList.add(new StoredSnapshot(new SnapshotMetadata(persistentId, snapshotList.size(),
  54                     System.currentTimeMillis()), snapshot));
  55         }
  56     }
  57
  58     @SuppressWarnings("unchecked")
  59     public static <T> List<T> getSnapshots(String persistentId, Class<T> type) {
  60         List<StoredSnapshot> stored = snapshots.get(persistentId);
  61         if (stored == null) {
  62             return Collections.emptyList();
  63         }
  64
  65         List<T> retList;
  66         synchronized (stored) {
  67             retList = Lists.newArrayListWithCapacity(stored.size());
  68             for (StoredSnapshot s: stored) {
  69                 if (type.isInstance(s.data)) {
  70                     retList.add((T) s.data);
  71                 }
  72             }
  73         }
  74
  75         return retList;
  76     }
  77
  78     public static void clear() {
  79         snapshots.clear();
  80     }
  81
  82     public static void addSnapshotSavedLatch(String persistenceId) {
  83         SNAPSHOT_SAVED_LATCHES.put(persistenceId, new CountDownLatch(1));
  84     }
  85
  86     public static void addSnapshotDeletedLatch(String persistenceId) {
  87         SNAPSHOT_DELETED_LATCHES.put(persistenceId, new CountDownLatch(1));
  88     }
  89
  90     public static <T> T waitForSavedSnapshot(String persistenceId, Class<T> type) {
  91         if (!Uninterruptibles.awaitUninterruptibly(SNAPSHOT_SAVED_LATCHES.get(persistenceId), 5, TimeUnit.SECONDS)) {
  92             throw new AssertionError("Snapshot was not saved");
  93         }
  94
  95         return getSnapshots(persistenceId, type).get(0);
  96     }
  97
  98     public static void waitForDeletedSnapshot(String persistenceId) {
  99         if (!Uninterruptibles.awaitUninterruptibly(SNAPSHOT_DELETED_LATCHES.get(persistenceId), 5, TimeUnit.SECONDS)) {
 100             throw new AssertionError("Snapshot was not deleted");
 101         }
 102     }
 103
 104     @Override
 105     public Future<Optional<SelectedSnapshot>> doLoadAsync(String persistenceId,
 106             SnapshotSelectionCriteria snapshotSelectionCriteria) {
 107         List<StoredSnapshot> snapshotList = snapshots.get(persistenceId);
 108         if (snapshotList == null) {
 109             return Futures.successful(Optional.<SelectedSnapshot>empty());
 110         }
 111
 112         synchronized (snapshotList) {
 113             for (int i = snapshotList.size() - 1; i >= 0; i--) {
 114                 StoredSnapshot snapshot = snapshotList.get(i);
 115                 if (matches(snapshot, snapshotSelectionCriteria)) {
 116                     return Futures.successful(Optional.of(new SelectedSnapshot(snapshot.metadata,
 117                             snapshot.data)));
 118                 }
 119             }
 120         }
 121
 122         return Futures.successful(Optional.<SelectedSnapshot>empty());
 123     }
 124
 125     private static boolean matches(StoredSnapshot snapshot, SnapshotSelectionCriteria criteria) {
 126         return snapshot.metadata.sequenceNr() <= criteria.maxSequenceNr()
 127                 && snapshot.metadata.timestamp() <= criteria.maxTimestamp();
 128     }
 129
 130     @Override
 131     public Future<Void> doSaveAsync(SnapshotMetadata snapshotMetadata, Object obj) {
 132         List<StoredSnapshot> snapshotList = snapshots.get(snapshotMetadata.persistenceId());
 133
 134         LOG.trace("doSaveAsync: persistentId {}: sequenceNr: {}: timestamp {}: {}", snapshotMetadata.persistenceId(),
 135                 snapshotMetadata.sequenceNr(), snapshotMetadata.timestamp(), obj);
 136
 137         if (snapshotList == null) {
 138             snapshotList = new ArrayList<>();
 139             snapshots.put(snapshotMetadata.persistenceId(), snapshotList);
 140         }
 141         synchronized (snapshotList) {
 142             snapshotList.add(new StoredSnapshot(snapshotMetadata, obj));
 143         }
 144
 145         CountDownLatch latch = SNAPSHOT_SAVED_LATCHES.get(snapshotMetadata.persistenceId());
 146         if (latch != null) {
 147             latch.countDown();
 148         }
 149
 150         return Futures.successful(null);
 151     }
 152
 153     @Override
 154     public Future<Void> doDeleteAsync(SnapshotMetadata metadata) {
 155         List<StoredSnapshot> snapshotList = snapshots.get(metadata.persistenceId());
 156
 157         if (snapshotList != null) {
 158             synchronized (snapshotList) {
 159                 for (int i = 0; i < snapshotList.size(); i++) {
 160                     StoredSnapshot snapshot = snapshotList.get(i);
 161                     if (metadata.equals(snapshot.metadata)) {
 162                         snapshotList.remove(i);
 163                         break;
 164                     }
 165                 }
 166             }
 167         }
 168
 169         return Futures.successful(null);
 170     }
 171
 172     @Override
 173     public Future<Void> doDeleteAsync(String persistenceId, SnapshotSelectionCriteria criteria) {
 174         LOG.trace("doDelete: persistentId {}: maxSequenceNr: {}: maxTimestamp {}", persistenceId,
 175             criteria.maxSequenceNr(), criteria.maxTimestamp());
 176
 177         List<StoredSnapshot> snapshotList = snapshots.get(persistenceId);
 178         if (snapshotList != null) {
 179             synchronized (snapshotList) {
 180                 Iterator<StoredSnapshot> iter = snapshotList.iterator();
 181                 while (iter.hasNext()) {
 182                     StoredSnapshot stored = iter.next();
 183                     if (matches(stored, criteria)) {
 184                         LOG.trace("Deleting snapshot for sequenceNr: {}, timestamp: {}: {}",
 185                                 stored.metadata.sequenceNr(), stored.metadata.timestamp(), stored.data);
 186
 187                         iter.remove();
 188                     }
 189                 }
 190             }
 191         }
 192
 193         CountDownLatch latch = SNAPSHOT_DELETED_LATCHES.get(persistenceId);
 194         if (latch != null) {
 195             latch.countDown();
 196         }
 197
 198         return Futures.successful(null);
 199     }
 200
 201     private static class StoredSnapshot {
 202         private final SnapshotMetadata metadata;
 203         private final Object data;
 204
 205         private StoredSnapshot(SnapshotMetadata metadata, Object data) {
 206             this.metadata = metadata;
 207             this.data = data;
 208         }
 209     }
 210 }