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