import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.Creator;
-import akka.persistence.Persistent;
-import akka.persistence.RecoveryCompleted;
-import akka.persistence.UntypedProcessor;
+import akka.serialization.Serialization;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
-import org.opendaylight.controller.cluster.datastore.messages.NonPersistent;
+import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
+import org.opendaylight.controller.cluster.raft.RaftActor;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
* Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
* </p>
*/
-public class Shard extends UntypedProcessor {
+public class Shard extends RaftActor {
public static final String DEFAULT_NAME = "default";
// property persistent
private final boolean persistent;
+ private final String name;
+
private SchemaContext schemaContext;
private final ShardStats shardMBean;
- private Shard(String name) {
+ private Shard(String name, Map<String, String> peerAddresses) {
+ super(name, peerAddresses);
+
+ this.name = name;
String setting = System.getProperty("shard.persistent");
}
- public static Props props(final String name) {
+ public static Props props(final String name, final Map<String, String> peerAddresses) {
return Props.create(new Creator<Shard>() {
@Override
public Shard create() throws Exception {
- return new Shard(name);
+ return new Shard(name, peerAddresses);
}
});
}
- @Override
- public void onReceive(Object message) throws Exception {
- LOG.debug("Received message " + message.getClass().toString());
-
- if(!recoveryFinished()){
- // FIXME : Properly handle recovery
- return;
- }
+ @Override public void onReceiveCommand(Object message){
+ LOG.debug("Received message {} from {}", message.getClass().toString(), getSender());
if (message.getClass().equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
- createTransactionChain();
+ if(isLeader()) {
+ createTransactionChain();
+ } else if(getLeader() != null){
+ getLeader().forward(message, getContext());
+ }
} else if (message.getClass().equals(RegisterChangeListener.SERIALIZABLE_CLASS)) {
registerChangeListener(RegisterChangeListener.fromSerializable(getContext().system(), message));
} else if (message instanceof UpdateSchemaContext) {
updateSchemaContext((UpdateSchemaContext) message);
} else if (message instanceof ForwardedCommitTransaction) {
handleForwardedCommit((ForwardedCommitTransaction) message);
- } else if (message instanceof Persistent) {
- commit(((Persistent)message).payload());
} else if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
- createTransaction(CreateTransaction.fromSerializable(message));
- } else if(message instanceof NonPersistent){
- commit(((NonPersistent)message).payload());
- }else if (message instanceof RecoveryCompleted) {
- //FIXME: PROPERLY HANDLE RECOVERY COMPLETED
-
- }else {
- throw new Exception("Not recognized message found message=" + message);
+ if(isLeader()) {
+ createTransaction(CreateTransaction.fromSerializable(message));
+ } else if(getLeader() != null){
+ getLeader().forward(message, getContext());
+ }
+ } else if (message instanceof PeerAddressResolved){
+ PeerAddressResolved resolved = (PeerAddressResolved) message;
+ setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
+ } else {
+ super.onReceiveCommand(message);
}
}
private void createTransaction(CreateTransaction createTransaction) {
DOMStoreReadWriteTransaction transaction =
store.newReadWriteTransaction();
+ String transactionId = "shard-" + createTransaction.getTransactionId();
+ LOG.info("Creating transaction : {} " , transactionId);
ActorRef transactionActor = getContext().actorOf(
- ShardTransaction.props(transaction, getSelf(), schemaContext), "shard-" + createTransaction.getTransactionId());
+ ShardTransaction.props(transaction, getSelf(), schemaContext), transactionId);
+
getSender()
- .tell(new CreateTransactionReply(transactionActor.path().toString(), createTransaction.getTransactionId()).toSerializable(),
+ .tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), createTransaction.getTransactionId()).toSerializable(),
getSelf());
}
- private void commit(Object serialized) {
+ private void commit(final ActorRef sender, Object serialized) {
Modification modification = MutableCompositeModification.fromSerializable(serialized, schemaContext);
DOMStoreThreePhaseCommitCohort cohort =
modificationToCohort.remove(serialized);
if (cohort == null) {
LOG.error(
"Could not find cohort for modification : " + modification);
+ LOG.info("Writing modification using a new transaction");
+ modification.apply(store.newReadWriteTransaction());
return;
}
+
final ListenableFuture<Void> future = cohort.commit();
shardMBean.incrementCommittedTransactionCount();
- final ActorRef sender = getSender();
final ActorRef self = getSelf();
future.addListener(new Runnable() {
@Override
public void run() {
try {
future.get();
- sender.tell(new CommitTransactionReply().toSerializable(), self);
+
+ if(sender != null) {
+ sender
+ .tell(new CommitTransactionReply().toSerializable(),
+ self);
+ } else {
+ LOG.error("sender is null ???");
+ }
} catch (InterruptedException | ExecutionException e) {
// FIXME : Handle this properly
LOG.error(e, "An exception happened when committing");
modificationToCohort
.put(serializedModification , message.getCohort());
+
if(persistent) {
- getSelf().forward(Persistent.create(serializedModification),
- getContext());
+ this.persistData(getSender(), "identifier", new CompositeModificationPayload(serializedModification));
} else {
- getSelf().forward(NonPersistent.create(serializedModification),
- getContext());
+ this.commit(getSender(), serializedModification);
}
}
ActorSelection dataChangeListenerPath = getContext()
- .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
+ .system().actorSelection(
+ registerChangeListener.getDataChangeListenerPath());
AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
listener = new DataChangeListenerProxy(schemaContext,dataChangeListenerPath);
private void createTransactionChain() {
DOMStoreTransactionChain chain = store.createTransactionChain();
ActorRef transactionChain =
- getContext().actorOf(ShardTransactionChain.props(chain, schemaContext));
+ getContext().actorOf(
+ ShardTransactionChain.props(chain, schemaContext));
getSender()
- .tell(new CreateTransactionChainReply(transactionChain.path()).toSerializable(),
+ .tell(new CreateTransactionChainReply(transactionChain.path())
+ .toSerializable(),
getSelf());
}
+
+ @Override protected void applyState(ActorRef clientActor, String identifier,
+ Object data) {
+
+ if(data instanceof CompositeModificationPayload){
+ Object modification =
+ ((CompositeModificationPayload) data).getModification();
+ commit(clientActor, modification);
+ } else {
+ LOG.error("Unknown state received {}", data);
+ }
+
+ }
+
+ @Override protected Object createSnapshot() {
+ throw new UnsupportedOperationException("createSnapshot");
+ }
+
+ @Override protected void applySnapshot(Object snapshot) {
+ throw new UnsupportedOperationException("applySnapshot");
+ }
+
+ @Override public String persistenceId() {
+ return this.name;
+ }
}
Code Review / controller.git / blobdiff