opendaylight/md-sal/cds-access-client/src/test/java/org/opendaylight/controller/cluster/access/client/MockedSnapshotStore.java

   1 /*
   2  * Copyright (c) 2017 Pantheon Technologies s.r.o. 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.access.client;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.actor.ActorRef;
  13 import akka.dispatch.OnComplete;
  14 import akka.pattern.Patterns;
  15 import akka.persistence.SelectedSnapshot;
  16 import akka.persistence.SnapshotMetadata;
  17 import akka.persistence.SnapshotSelectionCriteria;
  18 import akka.persistence.snapshot.japi.SnapshotStore;
  19 import java.util.Optional;
  20 import scala.concurrent.Future;
  21 import scala.concurrent.Promise;
  22
  23 /**
  24  * Instantiated by akka. MockedSnapshotStore forwards method calls as
  25  * {@link MockedSnapshotStoreMessage} messages to delegate actor. Delegate reference
  26  * must be sent as a message to this snapshot store.
  27  */
  28 class MockedSnapshotStore extends SnapshotStore {
  29
  30     private static final long TIMEOUT = 1000;
  31
  32     private ActorRef delegate;
  33
  34     /**
  35      * Marker interface for messages produced by MockedSnapshotStore.
  36      */
  37     interface MockedSnapshotStoreMessage {
  38     }
  39
  40     @Override
  41     public Future<Optional<SelectedSnapshot>> doLoadAsync(final String persistenceId,
  42                                                           final SnapshotSelectionCriteria criteria) {
  43         return askDelegate(new LoadRequest(persistenceId, criteria));
  44     }
  45
  46     @Override
  47     public Future<Void> doSaveAsync(final SnapshotMetadata metadata, final Object snapshot) {
  48         return askDelegate(new SaveRequest(metadata, snapshot));
  49     }
  50
  51     @Override
  52     public Future<Void> doDeleteAsync(final SnapshotMetadata metadata) {
  53         return askDelegate(new DeleteByMetadataRequest(metadata));
  54     }
  55
  56     @Override
  57     public Future<Void> doDeleteAsync(final String persistenceId, final SnapshotSelectionCriteria criteria) {
  58         return askDelegate(new DeleteByCriteriaRequest(persistenceId, criteria));
  59     }
  60
  61     @Override
  62     public void unhandled(final Object message) {
  63         if (message instanceof ActorRef) {
  64             delegate = (ActorRef) message;
  65             return;
  66         }
  67         super.unhandled(message);
  68     }
  69
  70     private <T> Future<T> askDelegate(final MockedSnapshotStoreMessage message) {
  71         return transform(Patterns.ask(requireNonNull(delegate, "Delegate ref was not sent"), message, TIMEOUT));
  72     }
  73
  74     private <T> Future<T> transform(final Future<Object> future) {
  75         final Promise<T> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
  76         future.onComplete(new OnComplete<Object>() {
  77             @Override
  78             public void onComplete(final Throwable failure, final Object success) {
  79                 if (success instanceof Throwable) {
  80                     promise.failure((Throwable) success);
  81                     return;
  82                 }
  83                 if (success == Void.TYPE) {
  84                     promise.success(null);
  85                     return;
  86                 }
  87                 promise.success((T) success);
  88             }
  89         }, context().dispatcher());
  90         return promise.future();
  91     }
  92
  93     class LoadRequest implements MockedSnapshotStoreMessage {
  94         private final String persistenceId;
  95         private final SnapshotSelectionCriteria criteria;
  96
  97         LoadRequest(final String persistenceId, final SnapshotSelectionCriteria criteria) {
  98             this.persistenceId = persistenceId;
  99             this.criteria = criteria;
 100         }
 101
 102         public String getPersistenceId() {
 103             return persistenceId;
 104         }
 105
 106         public SnapshotSelectionCriteria getCriteria() {
 107             return criteria;
 108         }
 109     }
 110
 111     class DeleteByCriteriaRequest implements MockedSnapshotStoreMessage {
 112         private final String persistenceId;
 113         private final SnapshotSelectionCriteria criteria;
 114
 115         DeleteByCriteriaRequest(final String persistenceId, final SnapshotSelectionCriteria criteria) {
 116             this.persistenceId = persistenceId;
 117             this.criteria = criteria;
 118         }
 119
 120         public String getPersistenceId() {
 121             return persistenceId;
 122         }
 123
 124         public SnapshotSelectionCriteria getCriteria() {
 125             return criteria;
 126         }
 127     }
 128
 129     class DeleteByMetadataRequest implements MockedSnapshotStoreMessage {
 130         private final SnapshotMetadata metadata;
 131
 132         DeleteByMetadataRequest(final SnapshotMetadata metadata) {
 133             this.metadata = metadata;
 134         }
 135
 136         public SnapshotMetadata getMetadata() {
 137             return metadata;
 138         }
 139     }
 140
 141     class SaveRequest implements MockedSnapshotStoreMessage {
 142         private final SnapshotMetadata metadata;
 143         private final Object snapshot;
 144
 145         SaveRequest(final SnapshotMetadata metadata, final Object snapshot) {
 146             this.metadata = metadata;
 147             this.snapshot = snapshot;
 148         }
 149
 150         public SnapshotMetadata getMetadata() {
 151             return metadata;
 152         }
 153
 154         public Object getSnapshot() {
 155             return snapshot;
 156         }
 157     }
 158 }