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.
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
9 package org.opendaylight.controller.cluster.raft;
11 import static org.junit.Assert.assertEquals;
12 import static org.mockito.Mockito.mock;
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;
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;
36 public class MockRaftActor extends RaftActor implements RaftActorRecoveryCohort, RaftActorSnapshotCohort {
37 public static final short PAYLOAD_VERSION = 5;
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;
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);
59 this.snapshotCohortDelegate = builder.snapshotCohort != null ? builder.snapshotCohort :
60 mock(RaftActorSnapshotCohort.class);
62 if (builder.dataPersistenceProvider == null) {
63 setPersistence(builder.persistent.isPresent() ? builder.persistent.get() : true);
65 setPersistence(builder.dataPersistenceProvider);
68 roleChangeNotifier = builder.roleChangeNotifier;
69 snapshotMessageSupport = builder.snapshotMessageSupport;
70 restoreFromSnapshot = builder.restoreFromSnapshot;
71 pauseLeaderFunction = builder.pauseLeaderFunction;
74 public void setRaftActorRecoverySupport(final RaftActorRecoverySupport support) {
75 raftActorRecoverySupport = support;
79 public RaftActorRecoverySupport newRaftActorRecoverySupport() {
80 return raftActorRecoverySupport != null ? raftActorRecoverySupport : super.newRaftActorRecoverySupport();
84 protected RaftActorSnapshotMessageSupport newRaftActorSnapshotMessageSupport() {
85 return snapshotMessageSupport != null ? snapshotMessageSupport :
86 (snapshotMessageSupport = super.newRaftActorSnapshotMessageSupport());
90 public RaftActorContext getRaftActorContext() {
91 return super.getRaftActorContext();
94 public RaftActorSnapshotMessageSupport getSnapshotMessageSupport() {
95 return snapshotMessageSupport;
98 public void waitForRecoveryComplete() {
100 assertEquals("Recovery complete", true, recoveryComplete.await(5, TimeUnit.SECONDS));
101 } catch (InterruptedException e) {
102 throw new RuntimeException(e);
106 public void waitForInitializeBehaviorComplete() {
108 assertEquals("Behavior initialized", true, initializeBehaviorComplete.await(5, TimeUnit.SECONDS));
109 } catch (InterruptedException e) {
110 throw new RuntimeException(e);
115 public void waitUntilLeader() {
116 for (int i = 0; i < 10; i++) {
120 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
124 public List<Object> getState() {
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);
138 protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() {
143 protected RaftActorSnapshotCohort getRaftActorSnapshotCohort() {
148 public void startLogRecoveryBatch(final int maxBatchSize) {
152 public void appendRecoveredLogEntry(final Payload data) {
157 public void applyCurrentLogRecoveryBatch() {
161 protected void onRecoveryComplete() {
162 actorDelegate.onRecoveryComplete();
163 recoveryComplete.countDown();
167 protected void initializeBehavior() {
168 super.initializeBehavior();
169 initializeBehaviorComplete.countDown();
173 public void applyRecoverySnapshot(final Snapshot.State newState) {
174 recoveryCohortDelegate.applyRecoverySnapshot(newState);
175 applySnapshotState(newState);
178 private void applySnapshotState(final Snapshot.State newState) {
179 if (newState instanceof MockSnapshotState) {
181 state.addAll(((MockSnapshotState)newState).getState());
186 public void createSnapshot(final ActorRef actorRef, final java.util.Optional<OutputStream> installSnapshotStream) {
187 LOG.info("{}: createSnapshot called", persistenceId());
188 snapshotCohortDelegate.createSnapshot(actorRef, installSnapshotStream);
192 public void applySnapshot(final Snapshot.State newState) {
193 LOG.info("{}: applySnapshot called", persistenceId());
194 applySnapshotState(newState);
195 snapshotCohortDelegate.applySnapshot(newState);
199 public Snapshot.State deserializeSnapshot(final ByteSource snapshotBytes) {
201 return (Snapshot.State) SerializationUtils.deserialize(snapshotBytes.read());
202 } catch (IOException e) {
203 throw new RuntimeException("Error deserializing state", e);
208 protected void onStateChanged() {
209 actorDelegate.onStateChanged();
213 protected Optional<ActorRef> getRoleChangeNotifier() {
214 return Optional.fromNullable(roleChangeNotifier);
217 @Override public String persistenceId() {
221 protected void newBehavior(final RaftActorBehavior newBehavior) {
222 self().tell(newBehavior, ActorRef.noSender());
226 protected void handleCommand(final Object message) {
227 if (message instanceof RaftActorBehavior) {
228 super.changeCurrentBehavior((RaftActorBehavior)message);
230 super.handleCommand(message);
232 if (RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT.equals(message)) {
233 snapshotCommitted.countDown();
239 protected void pauseLeader(final Runnable operation) {
240 if (pauseLeaderFunction != null) {
241 pauseLeaderFunction.apply(operation);
243 super.pauseLeader(operation);
247 public static List<Object> fromState(final Snapshot.State from) {
248 if (from instanceof MockSnapshotState) {
249 return ((MockSnapshotState)from).getState();
252 throw new IllegalStateException("Unexpected snapshot State: " + from);
255 public ReplicatedLog getReplicatedLog() {
256 return this.getRaftActorContext().getReplicatedLog();
260 public Snapshot getRestoreFromSnapshot() {
261 return restoreFromSnapshot;
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();
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();
274 public static Builder builder() {
275 return new Builder();
278 public static class AbstractBuilder<T extends AbstractBuilder<T, A>, A extends MockRaftActor> {
279 private Map<String, String> peerAddresses = Collections.emptyMap();
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;
291 protected AbstractBuilder(final Class<A> actorClass) {
292 this.actorClass = actorClass;
295 @SuppressWarnings("unchecked")
300 public T id(final String newId) {
305 public T peerAddresses(final Map<String, String> newPeerAddresses) {
306 this.peerAddresses = newPeerAddresses;
310 public T config(final ConfigParams newConfig) {
311 this.config = newConfig;
315 public T dataPersistenceProvider(final DataPersistenceProvider newDataPersistenceProvider) {
316 this.dataPersistenceProvider = newDataPersistenceProvider;
320 public T roleChangeNotifier(final ActorRef newRoleChangeNotifier) {
321 this.roleChangeNotifier = newRoleChangeNotifier;
325 public T snapshotMessageSupport(final RaftActorSnapshotMessageSupport newSnapshotMessageSupport) {
326 this.snapshotMessageSupport = newSnapshotMessageSupport;
330 public T restoreFromSnapshot(final Snapshot newRestoreFromSnapshot) {
331 this.restoreFromSnapshot = newRestoreFromSnapshot;
335 public T persistent(final Optional<Boolean> newPersistent) {
336 this.persistent = newPersistent;
340 public T pauseLeaderFunction(final Function<Runnable, Void> newPauseLeaderFunction) {
341 this.pauseLeaderFunction = newPauseLeaderFunction;
345 public T snapshotCohort(final RaftActorSnapshotCohort newSnapshotCohort) {
346 this.snapshotCohort = newSnapshotCohort;
350 public Props props() {
351 return Props.create(actorClass, this);
355 public static class Builder extends AbstractBuilder<Builder, MockRaftActor> {
357 super(MockRaftActor.class);
361 public static class MockSnapshotState implements Snapshot.State {
362 private static final long serialVersionUID = 1L;
364 private final List<Object> state;
366 public MockSnapshotState(final List<Object> state) {
370 public List<Object> getState() {
375 public int hashCode() {
376 final int prime = 31;
378 result = prime * result + (state == null ? 0 : state.hashCode());
383 public boolean equals(final Object obj) {
390 if (getClass() != obj.getClass()) {
393 MockSnapshotState other = (MockSnapshotState) obj;
395 if (other.state != null) {
398 } else if (!state.equals(other.state)) {
405 public String toString() {
406 return "MockSnapshotState [state=" + state + "]";