import akka.actor.Props;
import akka.cluster.Cluster;
import com.google.common.annotations.VisibleForTesting;
-import java.util.ArrayList;
+import com.google.common.collect.ImmutableList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.LongSupplier;
import org.eclipse.jdt.annotation.NonNull;
private final ActorContext context;
+ private final @NonNull Executor executor;
+
private final String id;
private final ElectionTerm termInformation;
final @NonNull ElectionTerm termInformation, final long commitIndex, final long lastApplied,
final @NonNull Map<String, String> peerAddresses,
final @NonNull ConfigParams configParams, final @NonNull DataPersistenceProvider persistenceProvider,
- final @NonNull Consumer<ApplyState> applyStateConsumer, final @NonNull Logger logger) {
+ final @NonNull Consumer<ApplyState> applyStateConsumer, final @NonNull Logger logger,
+ final @NonNull Executor executor) {
this.actor = actor;
this.context = context;
this.id = id;
this.termInformation = requireNonNull(termInformation);
+ this.executor = requireNonNull(executor);
this.commitIndex = commitIndex;
this.lastApplied = lastApplied;
this.configParams = requireNonNull(configParams);
this.persistenceProvider = requireNonNull(persistenceProvider);
- this.log = requireNonNull(logger);
+ log = requireNonNull(logger);
this.applyStateConsumer = requireNonNull(applyStateConsumer);
fileBackedOutputStreamFactory = new FileBackedOutputStreamFactory(
return actor;
}
+ @Override
+ public final Executor getExecutor() {
+ return executor;
+ }
+
@Override
@SuppressWarnings("checkstyle:IllegalCatch")
public Optional<Cluster> getCluster() {
@Override
public Logger getLogger() {
- return this.log;
+ return log;
}
@Override
public void updatePeerIds(final ServerConfigurationPayload serverConfig) {
votingMember = true;
boolean foundSelf = false;
- Set<String> currentPeers = new HashSet<>(this.getPeerIds());
+ Set<String> currentPeers = new HashSet<>(getPeerIds());
for (ServerInfo server : serverConfig.getServerConfig()) {
if (getId().equals(server.getId())) {
foundSelf = true;
} else {
VotingState votingState = server.isVoting() ? VotingState.VOTING : VotingState.NON_VOTING;
if (!currentPeers.contains(server.getId())) {
- this.addToPeers(server.getId(), null, votingState);
+ addToPeers(server.getId(), null, votingState);
} else {
- this.getPeerInfo(server.getId()).setVotingState(votingState);
+ getPeerInfo(server.getId()).setVotingState(votingState);
currentPeers.remove(server.getId());
}
}
}
for (String peerIdToRemove : currentPeers) {
- this.removePeer(peerIdToRemove);
+ removePeer(peerIdToRemove);
}
if (!foundSelf) {
@Override
public void setDynamicServerConfigurationInUse() {
- this.dynamicServerConfiguration = true;
+ dynamicServerConfiguration = true;
}
@Override
if (!isDynamicServerConfigurationInUse()) {
return null;
}
- Collection<PeerInfo> peers = getPeers();
- List<ServerInfo> newConfig = new ArrayList<>(peers.size() + 1);
- for (PeerInfo peer: peers) {
+ final var peers = getPeers();
+ final var newConfig = ImmutableList.<ServerInfo>builderWithExpectedSize(peers.size() + (includeSelf ? 1 : 0));
+ for (PeerInfo peer : peers) {
newConfig.add(new ServerInfo(peer.getId(), peer.isVoting()));
}
newConfig.add(new ServerInfo(getId(), votingMember));
}
- return new ServerConfigurationPayload(newConfig);
+ return new ServerConfigurationPayload(newConfig.build());
}
@Override
}
void setCurrentBehavior(final RaftActorBehavior behavior) {
- this.currentBehavior = requireNonNull(behavior);
+ currentBehavior = requireNonNull(behavior);
}
@Override