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(final String persistentId, final Object snapshot) {
  45         List<StoredSnapshot> snapshotList = snapshots.computeIfAbsent(persistentId, k -> new ArrayList<>());
  46
  47         synchronized (snapshotList) {
  48             snapshotList.add(new StoredSnapshot(new SnapshotMetadata(persistentId, snapshotList.size(),
  49                     System.currentTimeMillis()), snapshot));
  50         }
  51     }
  52
  53     @SuppressWarnings("unchecked")
  54     public static <T> List<T> getSnapshots(final String persistentId, final Class<T> type) {
  55         List<StoredSnapshot> stored = snapshots.get(persistentId);
  56         if (stored == null) {
  57             return Collections.emptyList();
  58         }
  59
  60         List<T> retList;
  61         synchronized (stored) {
  62             retList = Lists.newArrayListWithCapacity(stored.size());
  63             for (StoredSnapshot s: stored) {
  64                 if (type.isInstance(s.data)) {
  65                     retList.add((T) s.data);
  66                 }
  67             }
  68         }
  69
  70         return retList;
  71     }
  72
  73     public static void clearSnapshotsFor(final String persistenceId) {
  74         snapshots.remove(persistenceId);
  75     }
  76
  77     public static void clear() {
  78         snapshots.clear();
  79     }
  80
  81     public static void addSnapshotSavedLatch(final String persistenceId) {
  82         SNAPSHOT_SAVED_LATCHES.put(persistenceId, new CountDownLatch(1));
  83     }
  84
  85     public static void addSnapshotDeletedLatch(final String persistenceId) {
  86         SNAPSHOT_DELETED_LATCHES.put(persistenceId, new CountDownLatch(1));
  87     }
  88
  89     public static <T> T waitForSavedSnapshot(final String persistenceId, final Class<T> type) {
  90         if (!Uninterruptibles.awaitUninterruptibly(SNAPSHOT_SAVED_LATCHES.get(persistenceId), 5, TimeUnit.SECONDS)) {
  91             throw new AssertionError("Snapshot was not saved");
  92         }
  93
  94         return getSnapshots(persistenceId, type).get(0);
  95     }
  96
  97     public static void waitForDeletedSnapshot(final String persistenceId) {
  98         if (!Uninterruptibles.awaitUninterruptibly(SNAPSHOT_DELETED_LATCHES.get(persistenceId), 5, TimeUnit.SECONDS)) {
  99             throw new AssertionError("Snapshot was not deleted");
 100         }
 101     }
 102
 103     @Override
 104     public Future<Optional<SelectedSnapshot>> doLoadAsync(final String persistenceId,
 105             final SnapshotSelectionCriteria snapshotSelectionCriteria) {
 106         List<StoredSnapshot> snapshotList = snapshots.get(persistenceId);
 107         if (snapshotList == null) {
 108             return Futures.successful(Optional.<SelectedSnapshot>empty());
 109         }
 110
 111         synchronized (snapshotList) {
 112             for (int i = snapshotList.size() - 1; i >= 0; i--) {
 113                 StoredSnapshot snapshot = snapshotList.get(i);
 114                 if (matches(snapshot, snapshotSelectionCriteria)) {
 115                     return Futures.successful(Optional.of(new SelectedSnapshot(snapshot.metadata,
 116                             snapshot.data)));
 117                 }
 118             }
 119         }
 120
 121         return Futures.successful(Optional.<SelectedSnapshot>empty());
 122     }
 123
 124     private static boolean matches(final StoredSnapshot snapshot, final SnapshotSelectionCriteria criteria) {
 125         return snapshot.metadata.sequenceNr() <= criteria.maxSequenceNr()
 126                 && snapshot.metadata.timestamp() <= criteria.maxTimestamp();
 127     }
 128
 129     @Override
 130     public Future<Void> doSaveAsync(final SnapshotMetadata snapshotMetadata, final Object obj) {
 131         List<StoredSnapshot> snapshotList = snapshots.get(snapshotMetadata.persistenceId());
 132
 133         LOG.trace("doSaveAsync: persistentId {}: sequenceNr: {}: timestamp {}: {}", snapshotMetadata.persistenceId(),
 134                 snapshotMetadata.sequenceNr(), snapshotMetadata.timestamp(), obj);
 135
 136         if (snapshotList == null) {
 137             snapshotList = new ArrayList<>();
 138             snapshots.put(snapshotMetadata.persistenceId(), snapshotList);
 139         }
 140         synchronized (snapshotList) {
 141             snapshotList.add(new StoredSnapshot(snapshotMetadata, obj));
 142         }
 143
 144         CountDownLatch latch = SNAPSHOT_SAVED_LATCHES.get(snapshotMetadata.persistenceId());
 145         if (latch != null) {
 146             latch.countDown();
 147         }
 148
 149         return Futures.successful(null);
 150     }
 151
 152     @Override
 153     public Future<Void> doDeleteAsync(final SnapshotMetadata metadata) {
 154         List<StoredSnapshot> snapshotList = snapshots.get(metadata.persistenceId());
 155
 156         if (snapshotList != null) {
 157             synchronized (snapshotList) {
 158                 for (int i = 0; i < snapshotList.size(); i++) {
 159                     StoredSnapshot snapshot = snapshotList.get(i);
 160                     if (metadata.equals(snapshot.metadata)) {
 161                         snapshotList.remove(i);
 162                         break;
 163                     }
 164                 }
 165             }
 166         }
 167
 168         return Futures.successful(null);
 169     }
 170
 171     @Override
 172     public Future<Void> doDeleteAsync(final String persistenceId, final SnapshotSelectionCriteria criteria) {
 173         LOG.trace("doDelete: persistentId {}: maxSequenceNr: {}: maxTimestamp {}", persistenceId,
 174             criteria.maxSequenceNr(), criteria.maxTimestamp());
 175
 176         List<StoredSnapshot> snapshotList = snapshots.get(persistenceId);
 177         if (snapshotList != null) {
 178             synchronized (snapshotList) {
 179                 Iterator<StoredSnapshot> iter = snapshotList.iterator();
 180                 while (iter.hasNext()) {
 181                     StoredSnapshot stored = iter.next();
 182                     if (matches(stored, criteria)) {
 183                         LOG.trace("Deleting snapshot for sequenceNr: {}, timestamp: {}: {}",
 184                                 stored.metadata.sequenceNr(), stored.metadata.timestamp(), stored.data);
 185
 186                         iter.remove();
 187                     }
 188                 }
 189             }
 190         }
 191
 192         CountDownLatch latch = SNAPSHOT_DELETED_LATCHES.get(persistenceId);
 193         if (latch != null) {
 194             latch.countDown();
 195         }
 196
 197         return Futures.successful(null);
 198     }
 199
 200     private static final class StoredSnapshot {
 201         private final SnapshotMetadata metadata;
 202         private final Object data;
 203
 204         StoredSnapshot(final SnapshotMetadata metadata, final Object data) {
 205             this.metadata = metadata;
 206             this.data = data;
 207         }
 208     }
 209 }