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