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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9 package org.opendaylight.controller.cluster.raft;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.mockito.Mockito.mock;
  13
  14 import akka.actor.ActorRef;
  15 import akka.actor.Props;
  16 import com.google.common.base.Function;
  17 import com.google.common.base.Optional;
  18 import com.google.common.io.ByteSource;
  19 import com.google.common.util.concurrent.Uninterruptibles;
  20 import java.io.IOException;
  21 import java.io.OutputStream;
  22 import java.util.ArrayList;
  23 import java.util.Collections;
  24 import java.util.List;
  25 import java.util.Map;
  26 import java.util.concurrent.CountDownLatch;
  27 import java.util.concurrent.TimeUnit;
  28 import javax.annotation.Nonnull;
  29 import org.apache.commons.lang3.SerializationUtils;
  30 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  31 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  32 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  33 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  34 import org.opendaylight.yangtools.concepts.Identifier;
  35
  36 public class MockRaftActor extends RaftActor implements RaftActorRecoveryCohort, RaftActorSnapshotCohort {
  37     public static final short PAYLOAD_VERSION = 5;
  38
  39     final RaftActor actorDelegate;
  40     final RaftActorRecoveryCohort recoveryCohortDelegate;
  41     volatile RaftActorSnapshotCohort snapshotCohortDelegate;
  42     private final CountDownLatch recoveryComplete = new CountDownLatch(1);
  43     private final List<Object> state;
  44     private final ActorRef roleChangeNotifier;
  45     protected final CountDownLatch initializeBehaviorComplete = new CountDownLatch(1);
  46     private RaftActorRecoverySupport raftActorRecoverySupport;
  47     private RaftActorSnapshotMessageSupport snapshotMessageSupport;
  48     private final Snapshot restoreFromSnapshot;
  49     final CountDownLatch snapshotCommitted = new CountDownLatch(1);
  50     private final Function<Runnable, Void> pauseLeaderFunction;
  51
  52     protected MockRaftActor(final AbstractBuilder<?, ?> builder) {
  53         super(builder.id, builder.peerAddresses != null ? builder.peerAddresses :
  54             Collections.<String, String>emptyMap(), Optional.fromNullable(builder.config), PAYLOAD_VERSION);
  55         state = new ArrayList<>();
  56         this.actorDelegate = mock(RaftActor.class);
  57         this.recoveryCohortDelegate = mock(RaftActorRecoveryCohort.class);
  58
  59         this.snapshotCohortDelegate = builder.snapshotCohort != null ? builder.snapshotCohort :
  60             mock(RaftActorSnapshotCohort.class);
  61
  62         if (builder.dataPersistenceProvider == null) {
  63             setPersistence(builder.persistent.isPresent() ? builder.persistent.get() : true);
  64         } else {
  65             setPersistence(builder.dataPersistenceProvider);
  66         }
  67
  68         roleChangeNotifier = builder.roleChangeNotifier;
  69         snapshotMessageSupport = builder.snapshotMessageSupport;
  70         restoreFromSnapshot = builder.restoreFromSnapshot;
  71         pauseLeaderFunction = builder.pauseLeaderFunction;
  72     }
  73
  74     public void setRaftActorRecoverySupport(final RaftActorRecoverySupport support) {
  75         raftActorRecoverySupport = support;
  76     }
  77
  78     @Override
  79     public RaftActorRecoverySupport newRaftActorRecoverySupport() {
  80         return raftActorRecoverySupport != null ? raftActorRecoverySupport : super.newRaftActorRecoverySupport();
  81     }
  82
  83     @Override
  84     protected RaftActorSnapshotMessageSupport newRaftActorSnapshotMessageSupport() {
  85         return snapshotMessageSupport != null ? snapshotMessageSupport :
  86             (snapshotMessageSupport = super.newRaftActorSnapshotMessageSupport());
  87     }
  88
  89     @Override
  90     public RaftActorContext getRaftActorContext() {
  91         return super.getRaftActorContext();
  92     }
  93
  94     public RaftActorSnapshotMessageSupport getSnapshotMessageSupport() {
  95         return snapshotMessageSupport;
  96     }
  97
  98     public void waitForRecoveryComplete() {
  99         try {
 100             assertEquals("Recovery complete", true, recoveryComplete.await(5,  TimeUnit.SECONDS));
 101         } catch (InterruptedException e) {
 102             throw new RuntimeException(e);
 103         }
 104     }
 105
 106     public void waitForInitializeBehaviorComplete() {
 107         try {
 108             assertEquals("Behavior initialized", true, initializeBehaviorComplete.await(5,  TimeUnit.SECONDS));
 109         } catch (InterruptedException e) {
 110             throw new RuntimeException(e);
 111         }
 112     }
 113
 114
 115     public void waitUntilLeader() {
 116         for (int i = 0; i < 10; i++) {
 117             if (isLeader()) {
 118                 break;
 119             }
 120             Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
 121         }
 122     }
 123
 124     public List<Object> getState() {
 125         return state;
 126     }
 127
 128     @Override
 129     protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) {
 130         actorDelegate.applyState(clientActor, identifier, data);
 131         LOG.info("{}: applyState called: {}", persistenceId(), data);
 132
 133         state.add(data);
 134     }
 135
 136     @Override
 137     @Nonnull
 138     protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() {
 139         return this;
 140     }
 141
 142     @Override
 143     protected RaftActorSnapshotCohort getRaftActorSnapshotCohort() {
 144         return this;
 145     }
 146
 147     @Override
 148     public void startLogRecoveryBatch(final int maxBatchSize) {
 149     }
 150
 151     @Override
 152     public void appendRecoveredLogEntry(final Payload data) {
 153         state.add(data);
 154     }
 155
 156     @Override
 157     public void applyCurrentLogRecoveryBatch() {
 158     }
 159
 160     @Override
 161     protected void onRecoveryComplete() {
 162         actorDelegate.onRecoveryComplete();
 163         recoveryComplete.countDown();
 164     }
 165
 166     @Override
 167     protected void initializeBehavior() {
 168         super.initializeBehavior();
 169         initializeBehaviorComplete.countDown();
 170     }
 171
 172     @Override
 173     public void applyRecoverySnapshot(final Snapshot.State newState) {
 174         recoveryCohortDelegate.applyRecoverySnapshot(newState);
 175         applySnapshotState(newState);
 176     }
 177
 178     private void applySnapshotState(final Snapshot.State newState) {
 179         if (newState instanceof MockSnapshotState) {
 180             state.clear();
 181             state.addAll(((MockSnapshotState)newState).getState());
 182         }
 183     }
 184
 185     @Override
 186     public void createSnapshot(final ActorRef actorRef, final java.util.Optional<OutputStream> installSnapshotStream) {
 187         LOG.info("{}: createSnapshot called", persistenceId());
 188         snapshotCohortDelegate.createSnapshot(actorRef, installSnapshotStream);
 189     }
 190
 191     @Override
 192     public void applySnapshot(final Snapshot.State newState) {
 193         LOG.info("{}: applySnapshot called", persistenceId());
 194         applySnapshotState(newState);
 195         snapshotCohortDelegate.applySnapshot(newState);
 196     }
 197
 198     @Override
 199     public Snapshot.State deserializeSnapshot(final ByteSource snapshotBytes) {
 200         try {
 201             return (Snapshot.State) SerializationUtils.deserialize(snapshotBytes.read());
 202         } catch (IOException e) {
 203             throw new RuntimeException("Error deserializing state", e);
 204         }
 205     }
 206
 207     @Override
 208     protected void onStateChanged() {
 209         actorDelegate.onStateChanged();
 210     }
 211
 212     @Override
 213     protected Optional<ActorRef> getRoleChangeNotifier() {
 214         return Optional.fromNullable(roleChangeNotifier);
 215     }
 216
 217     @Override public String persistenceId() {
 218         return this.getId();
 219     }
 220
 221     protected void newBehavior(final RaftActorBehavior newBehavior) {
 222         self().tell(newBehavior, ActorRef.noSender());
 223     }
 224
 225     @Override
 226     protected void handleCommand(final Object message) {
 227         if (message instanceof RaftActorBehavior) {
 228             super.changeCurrentBehavior((RaftActorBehavior)message);
 229         } else {
 230             super.handleCommand(message);
 231
 232             if (RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT.equals(message)) {
 233                 snapshotCommitted.countDown();
 234             }
 235         }
 236     }
 237
 238     @Override
 239     protected void pauseLeader(final Runnable operation) {
 240         if (pauseLeaderFunction != null) {
 241             pauseLeaderFunction.apply(operation);
 242         } else {
 243             super.pauseLeader(operation);
 244         }
 245     }
 246
 247     public static List<Object> fromState(final Snapshot.State from) {
 248         if (from instanceof MockSnapshotState) {
 249             return ((MockSnapshotState)from).getState();
 250         }
 251
 252         throw new IllegalStateException("Unexpected snapshot State: " + from);
 253     }
 254
 255     public ReplicatedLog getReplicatedLog() {
 256         return this.getRaftActorContext().getReplicatedLog();
 257     }
 258
 259     @Override
 260     public Snapshot getRestoreFromSnapshot() {
 261         return restoreFromSnapshot;
 262     }
 263
 264     public static Props props(final String id, final Map<String, String> peerAddresses, final ConfigParams config) {
 265         return builder().id(id).peerAddresses(peerAddresses).config(config).props();
 266     }
 267
 268     public static Props props(final String id, final Map<String, String> peerAddresses,
 269                               final ConfigParams config, final DataPersistenceProvider dataPersistenceProvider) {
 270         return builder().id(id).peerAddresses(peerAddresses).config(config)
 271                 .dataPersistenceProvider(dataPersistenceProvider).props();
 272     }
 273
 274     public static Builder builder() {
 275         return new Builder();
 276     }
 277
 278     public static class AbstractBuilder<T extends AbstractBuilder<T, A>, A extends MockRaftActor> {
 279         private Map<String, String> peerAddresses = Collections.emptyMap();
 280         private String id;
 281         private ConfigParams config;
 282         private DataPersistenceProvider dataPersistenceProvider;
 283         private ActorRef roleChangeNotifier;
 284         private RaftActorSnapshotMessageSupport snapshotMessageSupport;
 285         private Snapshot restoreFromSnapshot;
 286         private Optional<Boolean> persistent = Optional.absent();
 287         private final Class<A> actorClass;
 288         private Function<Runnable, Void> pauseLeaderFunction;
 289         private RaftActorSnapshotCohort snapshotCohort;
 290
 291         protected AbstractBuilder(final Class<A> actorClass) {
 292             this.actorClass = actorClass;
 293         }
 294
 295         @SuppressWarnings("unchecked")
 296         private T self() {
 297             return (T) this;
 298         }
 299
 300         public T id(final String newId) {
 301             this.id = newId;
 302             return self();
 303         }
 304
 305         public T peerAddresses(final Map<String, String> newPeerAddresses) {
 306             this.peerAddresses = newPeerAddresses;
 307             return self();
 308         }
 309
 310         public T config(final ConfigParams newConfig) {
 311             this.config = newConfig;
 312             return self();
 313         }
 314
 315         public T dataPersistenceProvider(final DataPersistenceProvider newDataPersistenceProvider) {
 316             this.dataPersistenceProvider = newDataPersistenceProvider;
 317             return self();
 318         }
 319
 320         public T roleChangeNotifier(final ActorRef newRoleChangeNotifier) {
 321             this.roleChangeNotifier = newRoleChangeNotifier;
 322             return self();
 323         }
 324
 325         public T snapshotMessageSupport(final RaftActorSnapshotMessageSupport newSnapshotMessageSupport) {
 326             this.snapshotMessageSupport = newSnapshotMessageSupport;
 327             return self();
 328         }
 329
 330         public T restoreFromSnapshot(final Snapshot newRestoreFromSnapshot) {
 331             this.restoreFromSnapshot = newRestoreFromSnapshot;
 332             return self();
 333         }
 334
 335         public T persistent(final Optional<Boolean> newPersistent) {
 336             this.persistent = newPersistent;
 337             return self();
 338         }
 339
 340         public T pauseLeaderFunction(final Function<Runnable, Void> newPauseLeaderFunction) {
 341             this.pauseLeaderFunction = newPauseLeaderFunction;
 342             return self();
 343         }
 344
 345         public T snapshotCohort(final RaftActorSnapshotCohort newSnapshotCohort) {
 346             this.snapshotCohort = newSnapshotCohort;
 347             return self();
 348         }
 349
 350         public Props props() {
 351             return Props.create(actorClass, this);
 352         }
 353     }
 354
 355     public static class Builder extends AbstractBuilder<Builder, MockRaftActor> {
 356         private Builder() {
 357             super(MockRaftActor.class);
 358         }
 359     }
 360
 361     public static class MockSnapshotState implements Snapshot.State {
 362         private static final long serialVersionUID = 1L;
 363
 364         private final List<Object> state;
 365
 366         public MockSnapshotState(final List<Object> state) {
 367             this.state = state;
 368         }
 369
 370         public List<Object> getState() {
 371             return state;
 372         }
 373
 374         @Override
 375         public int hashCode() {
 376             final int prime = 31;
 377             int result = 1;
 378             result = prime * result + (state == null ? 0 : state.hashCode());
 379             return result;
 380         }
 381
 382         @Override
 383         public boolean equals(final Object obj) {
 384             if (this == obj) {
 385                 return true;
 386             }
 387             if (obj == null) {
 388                 return false;
 389             }
 390             if (getClass() != obj.getClass()) {
 391                 return false;
 392             }
 393             MockSnapshotState other = (MockSnapshotState) obj;
 394             if (state == null) {
 395                 if (other.state != null) {
 396                     return false;
 397                 }
 398             } else if (!state.equals(other.state)) {
 399                 return false;
 400             }
 401             return true;
 402         }
 403
 404         @Override
 405         public String toString() {
 406             return "MockSnapshotState [state=" + state + "]";
 407         }
 408     }
 409 }