private static NodeConnectorId inPortMatch(Match sourceMatch) {
MatchField inPort = sourceMatch.getField(MatchType.IN_PORT);
if(inPort != null && inPort.getValue() != null && (inPort.getValue() instanceof NodeConnector)) {
- return new NodeConnectorId(((NodeConnector) inPort.getValue()).getNodeConnectorIdAsString());
+ NodeConnector port = (NodeConnector)inPort.getValue();
+ return (NodeConnectorId)MDFlowMapping.toUri(port);
}
return null;
}
MatchField vlan = sourceMatch.getField(MatchType.DL_VLAN);
if (vlan != null && vlan.getValue() != null) {
VlanIdBuilder vlanIDBuilder = new VlanIdBuilder();
+ short vid = (short)vlan.getValue();
+ boolean present = (vid != MatchType.DL_VLAN_NONE);
vlanIDBuilder.setVlanId(new VlanId((NetUtils
- .getUnsignedShort((short) vlan.getValue()))));
- vlanIDBuilder.setVlanIdPresent(true);
+ .getUnsignedShort(vid))));
+ vlanIDBuilder.setVlanIdPresent(present);
vlanMatchBuild.setVlanId(vlanIDBuilder.build());
}
if (vlanMatch != null) {
VlanId vlanId = vlanMatch.getVlanId();
if (vlanId != null) {
- org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.VlanId vlanIdInner = vlanId
- .getVlanId();
- if (vlanIdInner != null) {
- Integer vlanValue = vlanIdInner.getValue();
- if (vlanValue != null) {
- target.setField(DL_VLAN, vlanValue.shortValue());
+ if (Boolean.TRUE.equals(vlanId.isVlanIdPresent())) {
+ org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.VlanId vlanIdInner = vlanId
+ .getVlanId();
+ if (vlanIdInner != null) {
+ Integer vlanValue = vlanIdInner.getValue();
+ if (vlanValue != null) {
+ target.setField(DL_VLAN, vlanValue.shortValue());
+ }
}
+ } else {
+ target.setField(DL_VLAN, MatchType.DL_VLAN_NONE);
}
}
VlanPcp vlanPcp = vlanMatch.getVlanPcp();
/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ * Copyright (c) 2013-2014 Cisco Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
import org.opendaylight.controller.sal.action.SwPath;
import org.opendaylight.controller.sal.compatibility.MDFlowMapping;
import org.opendaylight.controller.sal.compatibility.ToSalConversionsUtils;
+import org.opendaylight.controller.sal.core.ConstructionException;
+import org.opendaylight.controller.sal.core.Node;
+import org.opendaylight.controller.sal.core.Node.NodeIDType;
+import org.opendaylight.controller.sal.core.NodeConnector;
+import org.opendaylight.controller.sal.core.NodeConnector.NodeConnectorIDType;
import org.opendaylight.controller.sal.flowprogrammer.Flow;
import org.opendaylight.controller.sal.match.Match;
import org.opendaylight.controller.sal.match.MatchType;
public class TestFromSalConversionsUtils {
private enum MtchType {
- other, ipv4, ipv6, arp, sctp, tcp, udp
+ other, untagged, ipv4, ipv6, arp, sctp, tcp, udp
}
@Test
- public void testFromSalConversion() {
+ public void testFromSalConversion() throws ConstructionException {
Flow salFlow = prepareSalFlowCommon();
NodeFlow odNodeFlow = MDFlowMapping.flowAdded(salFlow);
odNodeFlow = MDFlowMapping.flowAdded(prepareSalMatch(salFlow, MtchType.other));
checkOdMatch(odNodeFlow.getMatch(), MtchType.other);
+ odNodeFlow = MDFlowMapping.flowAdded(prepareSalMatch(salFlow, MtchType.untagged));
+ checkOdMatch(odNodeFlow.getMatch(), MtchType.untagged);
+
odNodeFlow = MDFlowMapping.flowAdded(prepareSalMatch(salFlow, MtchType.arp));
checkOdMatch(odNodeFlow.getMatch(), MtchType.arp);
assertNotNull("Ipv6 wasn't found", ipv6Found);
break;
case other:
+ assertEquals("Incoming port is wrong.", "openflow:12345:10", match.getInPort().getValue());
assertEquals("Source MAC address is wrong.", "ff:ee:dd:cc:bb:aa", match.getEthernetMatch()
.getEthernetSource().getAddress().getValue());
assertEquals("Destinatio MAC address is wrong.", "ff:ee:dd:cc:bb:aa", match.getEthernetMatch()
.getEthernetDestination().getAddress().getValue());
+ assertEquals("Vlan ID is not present.", Boolean.TRUE, match.getVlanMatch().getVlanId().isVlanIdPresent());
assertEquals("Vlan ID is wrong.", (Integer) 0xfff, match.getVlanMatch().getVlanId().getVlanId().getValue());
assertEquals("Vlan ID priority is wrong.", (short) 0x7, (short) match.getVlanMatch().getVlanPcp()
.getValue());
assertEquals("DCSP is wrong.", (short) 0x3f, (short) match.getIpMatch().getIpDscp().getValue());
break;
+ case untagged:
+ assertEquals("Source MAC address is wrong.", "ff:ee:dd:cc:bb:aa", match.getEthernetMatch()
+ .getEthernetSource().getAddress().getValue());
+ assertEquals("Destinatio MAC address is wrong.", "ff:ee:dd:cc:bb:aa", match.getEthernetMatch()
+ .getEthernetDestination().getAddress().getValue());
+ assertEquals("Vlan ID is present.", Boolean.FALSE, match.getVlanMatch().getVlanId().isVlanIdPresent());
+ assertEquals("Vlan ID is wrong.", Integer.valueOf(0), match.getVlanMatch().getVlanId().getVlanId().getValue());
+ assertEquals("DCSP is wrong.", (short) 0x3f, (short) match.getIpMatch().getIpDscp().getValue());
+ break;
case sctp:
boolean sctpFound = false;
assertEquals("Wrong protocol", CRUDP, match.getIpMatch().getIpProtocol().byteValue());
return salFlow;
}
- private Flow prepareSalMatch(Flow salFlow, MtchType mt) {
+ private Flow prepareSalMatch(Flow salFlow, MtchType mt) throws ConstructionException {
Match salMatch = new Match();
switch (mt) {
case arp:
salMatch.setField(MatchType.NW_DST, InetAddresses.forString("2001:0db8:85a3:0000:0000:8a2e:0370:7336"));
break;
case other:
+ Node node = new Node(NodeIDType.OPENFLOW, 12345L);
+ NodeConnector port = new NodeConnector(NodeConnectorIDType.OPENFLOW, Short.valueOf((short)10), node);
+ salMatch.setField(MatchType.IN_PORT, port);
salMatch.setField(MatchType.DL_SRC, new byte[]{(byte )0xff,(byte )0xee,(byte )0xdd,(byte )0xcc,(byte )0xbb,(byte )0xaa});
salMatch.setField(MatchType.DL_DST, new byte[]{(byte )0xff,(byte )0xee,(byte )0xdd,(byte )0xcc,(byte )0xbb,(byte )0xaa});
salMatch.setField(MatchType.DL_VLAN, (short) 0xfff);
salMatch.setField(MatchType.DL_VLAN_PR, (byte) 0x7);
salMatch.setField(MatchType.NW_TOS, (byte) 0x3f);
break;
+ case untagged:
+ salMatch.setField(MatchType.DL_SRC, new byte[]{(byte )0xff,(byte )0xee,(byte )0xdd,(byte )0xcc,(byte )0xbb,(byte )0xaa});
+ salMatch.setField(MatchType.DL_DST, new byte[]{(byte )0xff,(byte )0xee,(byte )0xdd,(byte )0xcc,(byte )0xbb,(byte )0xaa});
+ salMatch.setField(MatchType.DL_VLAN, MatchType.DL_VLAN_NONE);
+ salMatch.setField(MatchType.NW_TOS, (byte) 0x3f);
+ break;
case sctp:
salMatch.setField(MatchType.NW_PROTO, CRUDP);
salMatch.setField(MatchType.TP_SRC, (short) 0xffff);
/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ * Copyright (c) 2013-2014 Cisco Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
import org.opendaylight.controller.sal.core.Node;
import org.opendaylight.controller.sal.core.Node.NodeIDType;
import org.opendaylight.controller.sal.core.NodeConnector;
+import org.opendaylight.controller.sal.core.NodeConnector.NodeConnectorIDType;
import org.opendaylight.controller.sal.flowprogrammer.Flow;
import org.opendaylight.controller.sal.match.MatchType;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Dscp;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.apply.actions._case.ApplyActionsBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.InstructionBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.EtherType;
import org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.VlanId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.VlanPcp;
// prefix:
// od|Od = Open Daylight
private enum MtchType {
- other, ipv4, ipv6, arp, sctp, tcp, udp
+ other, untagged, ipv4, ipv6, arp, sctp, tcp, udp
}
@Test
Flow salFlow = ToSalConversionsUtils.toFlow(prepareOdFlow(odNodeFlowBuilder, MtchType.other), node);
checkSalMatch(salFlow.getMatch(), MtchType.other);
+ salFlow = ToSalConversionsUtils.toFlow(prepareOdFlow(odNodeFlowBuilder, MtchType.untagged), node);
+ checkSalMatch(salFlow.getMatch(), MtchType.untagged);
+
salFlow = ToSalConversionsUtils.toFlow(prepareOdFlow(odNodeFlowBuilder, MtchType.ipv4), node);
checkSalMatch(salFlow.getMatch(), MtchType.ipv4);
Assert.assertEquals("OF|1@OF|00:00:00:00:00:00:00:2a", nodeConnector.toString());
}
- private void checkSalMatch(org.opendaylight.controller.sal.match.Match match, MtchType mt) {
+ private void checkSalMatch(org.opendaylight.controller.sal.match.Match match, MtchType mt) throws ConstructionException {
switch (mt) {
case other:
/*assertNotNull("DL_DST isn't equal.", "3C:A9:F4:00:E0:C8",
assertEquals("DL_SRC isn't equal.", "24:77:03:7C:C5:F1",
new String((byte[]) match.getField(MatchType.DL_SRC).getValue()));
*/
+ Node node = new Node(NodeIDType.OPENFLOW, 12L);
+ NodeConnector port = new NodeConnector(NodeConnectorIDType.OPENFLOW, Short.valueOf((short)345), node);
+ assertEquals("IN_PORT isn't equal.", port, match.getField(MatchType.IN_PORT).getValue());
assertEquals("DL_TYPE isn't equal.", (short) 0xffff, (short) match.getField(MatchType.DL_TYPE).getValue());
assertEquals("NW_TOS isn't equal.", (byte) 0x33, (byte) match.getField(MatchType.NW_TOS).getValue());
assertEquals("NW_PROTO isn't equal.", (byte) 0x3f, (byte) match.getField(MatchType.NW_PROTO).getValue());
assertEquals("DL_VLAN isn't equal.", (short) 0xfff, (short) match.getField(MatchType.DL_VLAN).getValue());
assertEquals("DL_VLAN_PR isn't equal.", (byte) 0x7, (byte) match.getField(MatchType.DL_VLAN_PR).getValue());
break;
+ case untagged:
+ assertEquals("DL_TYPE isn't equal.", (short) 0xffff, (short) match.getField(MatchType.DL_TYPE).getValue());
+ assertEquals("NW_TOS isn't equal.", (byte) 0x33, (byte) match.getField(MatchType.NW_TOS).getValue());
+ assertEquals("NW_PROTO isn't equal.", (byte) 0x3f, (byte) match.getField(MatchType.NW_PROTO).getValue());
+ assertEquals("DL_VLAN isn't equal.", MatchType.DL_VLAN_NONE, (short) match.getField(MatchType.DL_VLAN).getValue());
+ break;
case arp:
/*
assertEquals("DL_SRC isn't equal.", "22:44:66:88:AA:CC",
MatchBuilder odMatchBuilder = new MatchBuilder();
switch (mt) {
case other:
+ odMatchBuilder.setInPort(new NodeConnectorId("openflow:12:345"));
odMatchBuilder.setEthernetMatch(prepEthernetMatch());
odMatchBuilder.setIpMatch(prepIpMatch());
odMatchBuilder.setVlanMatch(prepVlanMatch());
break;
+ case untagged:
+ odMatchBuilder.setEthernetMatch(prepEthernetMatch());
+ odMatchBuilder.setIpMatch(prepIpMatch());
+ odMatchBuilder.setVlanMatch(prepVlanNoneMatch());
+ break;
case ipv4:
odMatchBuilder.setLayer3Match(prepLayer3MatchIpv4());
break;
VlanMatchBuilder vlanMatchBuilder = new VlanMatchBuilder();
VlanIdBuilder vlanIdBuilder = new VlanIdBuilder().setVlanId(new VlanId(0xfff));
- vlanMatchBuilder.setVlanId(vlanIdBuilder.build());
+ vlanMatchBuilder.setVlanId(vlanIdBuilder.setVlanIdPresent(true).build());
vlanMatchBuilder.setVlanPcp(new VlanPcp((short) 0x7));
return vlanMatchBuilder.build();
+ }
+
+ private VlanMatch prepVlanNoneMatch() {
+ VlanMatchBuilder vlanMatchBuilder = new VlanMatchBuilder();
+ VlanIdBuilder vlanIdBuilder = new VlanIdBuilder().
+ setVlanIdPresent(false);
+ vlanMatchBuilder.setVlanId(vlanIdBuilder.build());
+
+ return vlanMatchBuilder.build();
}
private IpMatch prepIpMatch() {
<artifactId>sal-binding-config</artifactId>
</dependency>
- <!--
- Adding a temporary dependency on the sal-broker-impl so that we can use InMemoryDOMDataStore
+ <dependency>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-data-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-data-impl</artifactId>
+ </dependency>
- InMemoryDOMDataStore needs to be moved into its own module and be wired up using config subsystem before
- this bundle can use it
- -->
<dependency>
<groupId>org.opendaylight.controller</groupId>
- <artifactId>sal-broker-impl</artifactId>
+ <artifactId>sal-inmemory-datastore</artifactId>
+ <version>1.1-SNAPSHOT</version>
</dependency>
<dependency>
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
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.RegisterChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
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;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
Logging.getLogger(getContext().system(), this);
private Shard(String name) {
+ log.info("Creating shard : {}", name );
+
store = new InMemoryDOMDataStore(name, storeExecutor);
}
@Override
public void onReceive(Object message) throws Exception {
+ log.debug("Received message {}", message);
+
if (message instanceof CreateTransactionChain) {
createTransactionChain();
} else if (message instanceof RegisterChangeListener) {
handleForwardedCommit((ForwardedCommitTransaction) message);
} else if (message instanceof Persistent) {
commit((Persistent) message);
+ } else if (message instanceof CreateTransaction) {
+ createTransaction();
}
}
+ private void createTransaction() {
+ DOMStoreReadWriteTransaction transaction =
+ store.newReadWriteTransaction();
+ ActorRef transactionActor = getContext().actorOf(
+ ShardTransaction.props(transaction, getSelf()));
+ getSender()
+ .tell(new CreateTransactionReply(transactionActor.path()),
+ getSelf());
+ }
+
private void commit(Persistent message) {
Modification modification = (Modification) message.payload();
DOMStoreThreePhaseCommitCohort cohort =
package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.event.Logging;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
/**
* The ShardTransaction Actor represents a remote transaction
- *<p>
* The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
- *</p>
- *<p>
+ * </p>
* Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
* the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
* time there are no known advantages for creating a read-only or write-only transaction which may change over time
* at which point we can optimize things in the distributed store as well.
- *</p>
- *<p>
+ * </p>
* Handles Messages <br/>
* ---------------- <br/>
* <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
*/
public class ShardTransaction extends UntypedActor {
- private final ActorRef shardActor;
-
- private final DOMStoreReadWriteTransaction transaction;
-
- private final MutableCompositeModification modification = new MutableCompositeModification();
-
- private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
- public ShardTransaction(DOMStoreReadWriteTransaction transaction, ActorRef shardActor) {
- this.transaction = transaction;
- this.shardActor = shardActor;
- }
-
-
- public static Props props(final DOMStoreReadWriteTransaction transaction, final ActorRef shardActor){
- return Props.create(new Creator<ShardTransaction>(){
-
- @Override
- public ShardTransaction create() throws Exception {
- return new ShardTransaction(transaction, shardActor);
- }
- });
- }
-
- @Override
- public void onReceive(Object message) throws Exception {
- if(message instanceof ReadData){
- readData((ReadData) message);
- } else if(message instanceof WriteData){
- writeData((WriteData) message);
- } else if(message instanceof MergeData){
- mergeData((MergeData) message);
- } else if(message instanceof DeleteData){
- deleteData((DeleteData) message);
- } else if(message instanceof ReadyTransaction){
- readyTransaction((ReadyTransaction) message);
- } else if(message instanceof CloseTransaction){
- closeTransaction((CloseTransaction) message);
- } else if(message instanceof GetCompositedModification){
- // This is here for testing only
- getSender().tell(new GetCompositeModificationReply(new ImmutableCompositeModification(modification)), getSelf());
+ private final ActorRef shardActor;
+
+ // FIXME : see below
+ // If transactionChain is not null then this transaction is part of a
+ // transactionChain. Not really clear as to what that buys us
+ private final DOMStoreTransactionChain transactionChain;
+
+ private final DOMStoreReadWriteTransaction transaction;
+
+ private final MutableCompositeModification modification =
+ new MutableCompositeModification();
+
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+
+ public ShardTransaction(DOMStoreReadWriteTransaction transaction,
+ ActorRef shardActor) {
+ this(null, transaction, shardActor);
+ }
+
+ public ShardTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction,
+ ActorRef shardActor) {
+ this.transactionChain = transactionChain;
+ this.transaction = transaction;
+ this.shardActor = shardActor;
+ }
+
+
+
+ public static Props props(final DOMStoreReadWriteTransaction transaction,
+ final ActorRef shardActor) {
+ return Props.create(new Creator<ShardTransaction>() {
+
+ @Override
+ public ShardTransaction create() throws Exception {
+ return new ShardTransaction(transaction, shardActor);
+ }
+ });
+ }
+
+ public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
+ final ActorRef shardActor) {
+ return Props.create(new Creator<ShardTransaction>() {
+
+ @Override
+ public ShardTransaction create() throws Exception {
+ return new ShardTransaction(transactionChain, transaction, shardActor);
+ }
+ });
}
- }
-
- private void readData(ReadData message) {
- final ActorRef sender = getSender();
- final ActorRef self = getSelf();
- final InstanceIdentifier path = message.getPath();
- final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(path);
-
- future.addListener(new Runnable() {
- @Override
- public void run() {
- try {
- Optional<NormalizedNode<?, ?>> optional = future.get();
- if(optional.isPresent()){
- sender.tell(new ReadDataReply(optional.get()), self);
- } else {
- //TODO : Need to decide what to do here
- }
- } catch (InterruptedException | ExecutionException e) {
- log.error(e, "An exception happened when reading data from path : " + path.toString());
- }
- }
- }, getContext().dispatcher());
- }
+ @Override
+ public void onReceive(Object message) throws Exception {
+ log.debug("Received message {}", message);
+
+ if (message instanceof ReadData) {
+ readData((ReadData) message);
+ } else if (message instanceof WriteData) {
+ writeData((WriteData) message);
+ } else if (message instanceof MergeData) {
+ mergeData((MergeData) message);
+ } else if (message instanceof DeleteData) {
+ deleteData((DeleteData) message);
+ } else if (message instanceof ReadyTransaction) {
+ readyTransaction((ReadyTransaction) message);
+ } else if (message instanceof CloseTransaction) {
+ closeTransaction((CloseTransaction) message);
+ } else if (message instanceof GetCompositedModification) {
+ // This is here for testing only
+ getSender().tell(new GetCompositeModificationReply(
+ new ImmutableCompositeModification(modification)), getSelf());
+ }
+ }
- private void writeData(WriteData message){
- modification.addModification(new WriteModification(message.getPath(), message.getData()));
- transaction.write(message.getPath(), message.getData());
- getSender().tell(new WriteDataReply(), getSelf());
- }
+ private void readData(ReadData message) {
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ final InstanceIdentifier path = message.getPath();
+ final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
+ transaction.read(path);
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Optional<NormalizedNode<?, ?>> optional = future.get();
+ if (optional.isPresent()) {
+ sender.tell(new ReadDataReply(optional.get()), self);
+ } else {
+ sender.tell(new ReadDataReply(null), self);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ log.error(e,
+ "An exception happened when reading data from path : "
+ + path.toString());
+ }
+
+ }
+ }, getContext().dispatcher());
+ }
- private void mergeData(MergeData message){
- modification.addModification(new MergeModification(message.getPath(), message.getData()));
- transaction.merge(message.getPath(), message.getData());
- getSender().tell(new MergeDataReply(), getSelf());
- }
- private void deleteData(DeleteData message){
- modification.addModification(new DeleteModification(message.getPath()));
- transaction.delete(message.getPath());
- getSender().tell(new DeleteDataReply(), getSelf());
- }
+ private void writeData(WriteData message) {
+ modification.addModification(
+ new WriteModification(message.getPath(), message.getData()));
+ transaction.write(message.getPath(), message.getData());
+ getSender().tell(new WriteDataReply(), getSelf());
+ }
- private void readyTransaction(ReadyTransaction message){
- DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
- ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort, shardActor, modification));
- getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
+ private void mergeData(MergeData message) {
+ modification.addModification(
+ new MergeModification(message.getPath(), message.getData()));
+ transaction.merge(message.getPath(), message.getData());
+ getSender().tell(new MergeDataReply(), getSelf());
+ }
- }
+ private void deleteData(DeleteData message) {
+ modification.addModification(new DeleteModification(message.getPath()));
+ transaction.delete(message.getPath());
+ getSender().tell(new DeleteDataReply(), getSelf());
+ }
- private void closeTransaction(CloseTransaction message){
- transaction.close();
- getSender().tell(new CloseTransactionReply(), getSelf());
- }
+ private void readyTransaction(ReadyTransaction message) {
+ DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
+ ActorRef cohortActor = getContext().actorOf(
+ ThreePhaseCommitCohort.props(cohort, shardActor, modification));
+ getSender()
+ .tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
+ }
- // These classes are in here for test purposes only
+ private void closeTransaction(CloseTransaction message) {
+ transaction.close();
+ getSender().tell(new CloseTransactionReply(), getSelf());
+ getSelf().tell(PoisonPill.getInstance(), getSelf());
+ }
- static class GetCompositedModification {
- }
+ // These classes are in here for test purposes only
- static class GetCompositeModificationReply {
- private final CompositeModification modification;
+ static class GetCompositedModification {
- GetCompositeModificationReply(CompositeModification modification) {
- this.modification = modification;
}
- public CompositeModification getModification() {
- return modification;
+ static class GetCompositeModificationReply {
+ private final CompositeModification modification;
+
+
+ GetCompositeModificationReply(CompositeModification modification) {
+ this.modification = modification;
+ }
+
+
+ public CompositeModification getModification() {
+ return modification;
+ }
}
- }
}
public void onReceive(Object message) throws Exception {
if(message instanceof CreateTransaction){
DOMStoreReadWriteTransaction transaction = chain.newReadWriteTransaction();
- ActorRef transactionActor = getContext().actorOf(ShardTransaction.props(transaction, getContext().parent()));
+ ActorRef transactionActor = getContext().actorOf(ShardTransaction.props(chain, transaction, getContext().parent()));
getSender().tell(new CreateTransactionReply(transactionActor.path()), getSelf());
} else if (message instanceof CloseTransactionChain){
chain.close();
package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.event.Logging;
import java.util.concurrent.ExecutionException;
-public class ThreePhaseCommitCohort extends UntypedActor{
- private final DOMStoreThreePhaseCommitCohort cohort;
- private final ActorRef shardActor;
- private final CompositeModification modification;
-
- public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, ActorRef shardActor, CompositeModification modification) {
- this.cohort = cohort;
- this.shardActor = shardActor;
- this.modification = modification;
- }
-
- private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
- public static Props props(final DOMStoreThreePhaseCommitCohort cohort, final ActorRef shardActor, final CompositeModification modification) {
- return Props.create(new Creator<ThreePhaseCommitCohort>(){
- @Override
- public ThreePhaseCommitCohort create() throws Exception {
- return new ThreePhaseCommitCohort(cohort, shardActor, modification);
- }
- });
- }
-
- @Override
- public void onReceive(Object message) throws Exception {
- if(message instanceof CanCommitTransaction){
- canCommit((CanCommitTransaction) message);
- } else if(message instanceof PreCommitTransaction) {
- preCommit((PreCommitTransaction) message);
- } else if(message instanceof CommitTransaction){
- commit((CommitTransaction) message);
- } else if (message instanceof AbortTransaction){
- abort((AbortTransaction) message);
+public class ThreePhaseCommitCohort extends UntypedActor {
+ private final DOMStoreThreePhaseCommitCohort cohort;
+ private final ActorRef shardActor;
+ private final CompositeModification modification;
+
+ public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort,
+ ActorRef shardActor, CompositeModification modification) {
+
+ this.cohort = cohort;
+ this.shardActor = shardActor;
+ this.modification = modification;
}
- }
-
- private void abort(AbortTransaction message) {
- final ListenableFuture<Void> future = cohort.abort();
- final ActorRef sender = getSender();
- final ActorRef self = getSelf();
-
- future.addListener(new Runnable() {
- @Override
- public void run() {
- try {
- future.get();
- sender.tell(new AbortTransactionReply(), self);
- } catch (InterruptedException | ExecutionException e) {
- log.error(e, "An exception happened when aborting");
- }
- }
- }, getContext().dispatcher());
- }
-
- private void commit(CommitTransaction message) {
- // Forward the commit to the shard
- log.info("Commit transaction now + " + shardActor);
- shardActor.forward(new ForwardedCommitTransaction(cohort, modification), getContext());
-
- }
-
- private void preCommit(PreCommitTransaction message) {
- final ListenableFuture<Void> future = cohort.preCommit();
- final ActorRef sender = getSender();
- final ActorRef self = getSelf();
-
- future.addListener(new Runnable() {
- @Override
- public void run() {
- try {
- future.get();
- sender.tell(new PreCommitTransactionReply(), self);
- } catch (InterruptedException | ExecutionException e) {
- log.error(e, "An exception happened when preCommitting");
- }
- }
- }, getContext().dispatcher());
-
- }
-
- private void canCommit(CanCommitTransaction message) {
- final ListenableFuture<Boolean> future = cohort.canCommit();
- final ActorRef sender = getSender();
- final ActorRef self = getSelf();
-
- future.addListener(new Runnable() {
- @Override
- public void run() {
- try {
- Boolean canCommit = future.get();
- sender.tell(new CanCommitTransactionReply(canCommit), self);
- } catch (InterruptedException | ExecutionException e) {
- log.error(e, "An exception happened when aborting");
+
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+
+ public static Props props(final DOMStoreThreePhaseCommitCohort cohort,
+ final ActorRef shardActor, final CompositeModification modification) {
+ return Props.create(new Creator<ThreePhaseCommitCohort>() {
+ @Override
+ public ThreePhaseCommitCohort create() throws Exception {
+ return new ThreePhaseCommitCohort(cohort, shardActor,
+ modification);
+ }
+ });
+ }
+
+ @Override
+ public void onReceive(Object message) throws Exception {
+ log.debug("Received message {}", message);
+
+ if (message instanceof CanCommitTransaction) {
+ canCommit((CanCommitTransaction) message);
+ } else if (message instanceof PreCommitTransaction) {
+ preCommit((PreCommitTransaction) message);
+ } else if (message instanceof CommitTransaction) {
+ commit((CommitTransaction) message);
+ } else if (message instanceof AbortTransaction) {
+ abort((AbortTransaction) message);
}
- }
- }, getContext().dispatcher());
+ }
+
+ private void abort(AbortTransaction message) {
+ final ListenableFuture<Void> future = cohort.abort();
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ future.get();
+ sender.tell(new AbortTransactionReply(), self);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error(e, "An exception happened when aborting");
+ }
+ }
+ }, getContext().dispatcher());
+ }
- }
+ private void commit(CommitTransaction message) {
+ // Forward the commit to the shard
+ log.debug("Forward commit transaction to Shard {} ", shardActor);
+ shardActor.forward(new ForwardedCommitTransaction(cohort, modification),
+ getContext());
+
+ getContext().parent().tell(PoisonPill.getInstance(), getSelf());
+
+ }
+
+ private void preCommit(PreCommitTransaction message) {
+ final ListenableFuture<Void> future = cohort.preCommit();
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ future.get();
+ sender.tell(new PreCommitTransactionReply(), self);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error(e, "An exception happened when preCommitting");
+ }
+ }
+ }, getContext().dispatcher());
+
+ }
+
+ private void canCommit(CanCommitTransaction message) {
+ final ListenableFuture<Boolean> future = cohort.canCommit();
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Boolean canCommit = future.get();
+ sender.tell(new CanCommitTransactionReply(canCommit), self);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error(e, "An exception happened when aborting");
+ }
+ }
+ }, getContext().dispatcher());
+
+ }
}
package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorPath;
+import akka.actor.ActorSelection;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListenableFutureTask;
+import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
/**
* ThreePhaseCommitCohortProxy represents a set of remote cohort proxies
public class ThreePhaseCommitCohortProxy implements
DOMStoreThreePhaseCommitCohort{
+ private static final Logger
+ LOG = LoggerFactory.getLogger(DistributedDataStore.class);
+
+ private final ActorContext actorContext;
private final List<ActorPath> cohortPaths;
+ //FIXME : Use a thread pool here
+ private final ExecutorService executorService = Executors.newSingleThreadExecutor();
- public ThreePhaseCommitCohortProxy(List<ActorPath> cohortPaths) {
+ public ThreePhaseCommitCohortProxy(ActorContext actorContext, List<ActorPath> cohortPaths) {
+ this.actorContext = actorContext;
this.cohortPaths = cohortPaths;
}
@Override public ListenableFuture<Boolean> canCommit() {
- throw new UnsupportedOperationException("canCommit");
+ Callable<Boolean> call = new Callable() {
+
+ @Override public Boolean call() throws Exception {
+ for(ActorPath actorPath : cohortPaths){
+ ActorSelection cohort = actorContext.actorSelection(actorPath);
+
+ try {
+ Object response =
+ actorContext.executeRemoteOperation(cohort,
+ new CanCommitTransaction(),
+ ActorContext.ASK_DURATION);
+
+ if (response instanceof CanCommitTransactionReply) {
+ CanCommitTransactionReply reply =
+ (CanCommitTransactionReply) response;
+ if (!reply.getCanCommit()) {
+ return false;
+ }
+ }
+ } catch(RuntimeException e){
+ LOG.error("Unexpected Exception", e);
+ return false;
+ }
+
+
+ }
+ return true;
+ }
+ };
+
+ ListenableFutureTask<Boolean>
+ future = ListenableFutureTask.create(call);
+
+ executorService.submit(future);
+
+ return future;
}
@Override public ListenableFuture<Void> preCommit() {
- throw new UnsupportedOperationException("preCommit");
+ return voidOperation(new PreCommitTransaction(), PreCommitTransactionReply.class);
}
@Override public ListenableFuture<Void> abort() {
- throw new UnsupportedOperationException("abort");
+ return voidOperation(new AbortTransaction(), AbortTransactionReply.class);
}
@Override public ListenableFuture<Void> commit() {
- throw new UnsupportedOperationException("commit");
+ return voidOperation(new CommitTransaction(), CommitTransactionReply.class);
+ }
+
+ private ListenableFuture<Void> voidOperation(final Object message, final Class expectedResponseClass){
+ Callable<Void> call = new Callable<Void>() {
+
+ @Override public Void call() throws Exception {
+ for(ActorPath actorPath : cohortPaths){
+ ActorSelection cohort = actorContext.actorSelection(actorPath);
+
+ try {
+ Object response =
+ actorContext.executeRemoteOperation(cohort,
+ message,
+ ActorContext.ASK_DURATION);
+
+ if (response != null && !response.getClass()
+ .equals(expectedResponseClass)) {
+ throw new RuntimeException(
+ String.format(
+ "did not get the expected response \n\t\t expected : %s \n\t\t actual : %s",
+ expectedResponseClass.toString(),
+ response.getClass().toString())
+ );
+ }
+ } catch(TimeoutException e){
+ LOG.error(String.format("A timeout occurred when processing operation : %s", message));
+ }
+ }
+ return null;
+ }
+ };
+
+ ListenableFutureTask<Void>
+ future = ListenableFutureTask.create(call);
+
+ executorService.submit(future);
+
+ return future;
+
}
public List<ActorPath> getCohortPaths() {
@Override
public void close() {
+ // FIXME : The problem here is don't know which shard the transaction chain is to be created on ???
throw new UnsupportedOperationException("close - not sure what to do here?");
}
}
private static final AtomicLong counter = new AtomicLong();
- private final TransactionType readOnly;
+ private final TransactionType transactionType;
private final ActorContext actorContext;
private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
private final String identifier;
public TransactionProxy(
ActorContext actorContext,
- TransactionType readOnly) {
+ TransactionType transactionType) {
this.identifier = "transaction-" + counter.getAndIncrement();
- this.readOnly = readOnly;
+ this.transactionType = transactionType;
this.actorContext = actorContext;
Object response = actorContext.executeShardOperation(Shard.DEFAULT_NAME, new CreateTransaction(), ActorContext.ASK_DURATION);
}
}
- return new ThreePhaseCommitCohortProxy(cohortPaths);
+ return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths);
}
@Override
--- /dev/null
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.exceptions;
+
+public class PrimaryNotFoundException extends RuntimeException {
+}
--- /dev/null
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.exceptions;
+
+public class TimeoutException extends RuntimeException {
+ public TimeoutException(Exception e){
+ super(e);
+ }
+}
import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.util.Timeout;
+import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
+import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
import org.slf4j.Logger;
return actorSystem.actorSelection(found.getPrimaryPath());
}
- throw new RuntimeException("primary was not found");
+ throw new PrimaryNotFoundException();
}
/**
try {
return Await.result(future, AWAIT_DURATION);
} catch (Exception e) {
- throw new RuntimeException(e);
+ throw new TimeoutException(e);
}
}
try {
return Await.result(future, AWAIT_DURATION);
} catch (Exception e) {
- throw new RuntimeException(e);
+ throw new TimeoutException(e);
}
}
* @param shardName
* @param message
* @param duration
- * @throws java.lang.RuntimeException when a primary is not found or if the message to the remote shard fails or times out
+ * @throws org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException if the message to the remote shard times out
+ * @throws org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException if the primary shard is not found
*
* @return
*/
--- /dev/null
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.ConfigFactory;
+import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
+
+public class DistributedConfigDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedConfigDataStoreProviderModule {
+ public DistributedConfigDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public DistributedConfigDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedConfigDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ // add custom validation form module attributes here.
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+ final ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster-system", ConfigFactory
+ .load().getConfig("ODLCluster"));
+
+
+ final DistributedDataStore configDatastore = new DistributedDataStore(actorSystem, "config");
+ getSchemaServiceDependency().registerSchemaServiceListener(configDatastore);
+
+ final class AutoCloseableDistributedDataStore implements AutoCloseable {
+
+ @Override
+ public void close() throws Exception {
+ actorSystem.shutdown();
+ }
+ }
+
+ return new AutoCloseableDistributedDataStore();
+ }
+
+}
/*
* Generated file
*
-* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-datastore-provider
+* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-config-datastore-provider
* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
-* Generated at: Thu Jun 12 15:23:43 PDT 2014
+* Generated at: Tue Jun 24 17:14:50 PDT 2014
*
* Do not modify this file unless it is present under src/main directory
*/
package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
-public class DistributedDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedDataStoreProviderModuleFactory {
+public class DistributedConfigDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedConfigDataStoreProviderModuleFactory {
}
+++ /dev/null
-package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
-
-import akka.actor.ActorSystem;
-import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
-
-public class DistributedDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedDataStoreProviderModule {
- public DistributedDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
- super(identifier, dependencyResolver);
- }
-
- public DistributedDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
- super(identifier, dependencyResolver, oldModule, oldInstance);
- }
-
- @Override
- public void customValidation() {
- // add custom validation form module attributes here.
- }
-
- @Override
- public java.lang.AutoCloseable createInstance() {
- ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster");
- final DistributedDataStore configurationStore = new DistributedDataStore(actorSystem, "config");
- final DistributedDataStore operationalStore = new DistributedDataStore(actorSystem, "operational");
-
- final class AutoCloseableDistributedDataStore implements AutoCloseable {
-
- @Override
- public void close() throws Exception {
- }
- }
-
- return new AutoCloseableDistributedDataStore();
- }
-
-}
--- /dev/null
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.ConfigFactory;
+import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
+
+public class DistributedOperationalDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedOperationalDataStoreProviderModule {
+ public DistributedOperationalDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public DistributedOperationalDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedOperationalDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ // add custom validation form module attributes here.
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+ final ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster", ConfigFactory
+ .load().getConfig("ODLCluster"));
+ final DistributedDataStore operationalStore = new DistributedDataStore(actorSystem, "operational");
+ getSchemaServiceDependency().registerSchemaServiceListener(operationalStore);
+
+ final class AutoCloseableDistributedDataStore implements AutoCloseable {
+
+ @Override
+ public void close() throws Exception {
+ actorSystem.shutdown();
+ }
+ }
+
+ return new AutoCloseableDistributedDataStore();
+ }
+
+}
--- /dev/null
+/*
+* Generated file
+*
+* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-operational-datastore-provider
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Tue Jun 24 17:14:50 PDT 2014
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+public class DistributedOperationalDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedOperationalDataStoreProviderModuleFactory {
+
+}
--- /dev/null
+ODLCluster{
+
+}
\ No newline at end of file
import config { prefix config; revision-date 2013-04-05; }
import rpc-context { prefix rpcx; revision-date 2013-06-17; }
+ import opendaylight-config-dom-datastore {prefix config-dom-store-spi;}
+ import opendaylight-operational-dom-datastore {prefix operational-dom-store-spi;}
+ import opendaylight-md-sal-dom {prefix sal;}
description
"This module contains the base YANG definitions for
}
// This is the definition of the service implementation as a module identity.
- identity distributed-datastore-provider {
+ identity distributed-config-datastore-provider {
base config:module-type;
-
+ config:provided-service config-dom-store-spi:config-dom-datastore;
// Specifies the prefix for generated java classes.
- config:java-name-prefix DistributedDataStoreProvider;
+ config:java-name-prefix DistributedConfigDataStoreProvider;
}
+ // This is the definition of the service implementation as a module identity.
+ identity distributed-operational-datastore-provider {
+ base config:module-type;
+ config:provided-service operational-dom-store-spi:operational-dom-datastore;
+ // Specifies the prefix for generated java classes.
+ config:java-name-prefix DistributedOperationalDataStoreProvider;
+ }
+
// Augments the 'configuration' choice node under modules/module.
augment "/config:modules/config:module/config:configuration" {
- case distributed-datastore-provider {
- when "/config:modules/config:module/config:type = 'distributed-datastore-provider'";
+ case distributed-config-datastore-provider {
+ when "/config:modules/config:module/config:type = 'distributed-config-datastore-provider'";
+ container schema-service {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity sal:schema-service;
+ }
+ }
+ }
}
}
+
+ // Augments the 'configuration' choice node under modules/module.
+ augment "/config:modules/config:module/config:configuration" {
+ case distributed-operational-datastore-provider {
+ when "/config:modules/config:module/config:type = 'distributed-operational-datastore-provider'";
+ container schema-service {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity sal:schema-service;
+ }
+ }
+ }
+ }
+ }
}
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Props;
+import akka.actor.Terminated;
import akka.testkit.JavaTestKit;
import junit.framework.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
public class BasicIntegrationTest extends AbstractActorTest {
@Test
- public void integrationTest() {
+ public void integrationTest() throws Exception{
// This test will
// - create a Shard
// - initiate a transaction
shard.tell(new CreateTransactionChain(), getRef());
final ActorSelection transactionChain =
- new ExpectMsg<ActorSelection>("match hint") {
+ new ExpectMsg<ActorSelection>("CreateTransactionChainReply") {
protected ActorSelection match(Object in) {
if (in instanceof CreateTransactionChainReply) {
ActorPath transactionChainPath =
transactionChain.tell(new CreateTransaction(), getRef());
final ActorSelection transaction =
- new ExpectMsg<ActorSelection>("match hint") {
+ new ExpectMsg<ActorSelection>("CreateTransactionReply") {
protected ActorSelection match(Object in) {
if (in instanceof CreateTransactionReply) {
ActorPath transactionPath =
Assert.assertNotNull(transaction);
+ // Add a watch on the transaction actor so that we are notified when it dies
+ final ActorRef transactionActorRef = watchActor(transaction);
+
transaction.tell(new WriteData(TestModel.TEST_PATH,
ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
getRef());
- Boolean writeDone = new ExpectMsg<Boolean>("match hint") {
+ Boolean writeDone = new ExpectMsg<Boolean>("WriteDataReply") {
protected Boolean match(Object in) {
if (in instanceof WriteDataReply) {
return true;
transaction.tell(new ReadyTransaction(), getRef());
final ActorSelection cohort =
- new ExpectMsg<ActorSelection>("match hint") {
+ new ExpectMsg<ActorSelection>("ReadyTransactionReply") {
protected ActorSelection match(Object in) {
if (in instanceof ReadyTransactionReply) {
ActorPath cohortPath =
Assert.assertNotNull(cohort);
+ // Add a watch on the transaction actor so that we are notified when it dies
+ final ActorRef cohorActorRef = watchActor(cohort);
+
cohort.tell(new PreCommitTransaction(), getRef());
Boolean preCommitDone =
- new ExpectMsg<Boolean>("match hint") {
+ new ExpectMsg<Boolean>("PreCommitTransactionReply") {
protected Boolean match(Object in) {
if (in instanceof PreCommitTransactionReply) {
return true;
cohort.tell(new CommitTransaction(), getRef());
+ final Boolean terminatedCohort =
+ new ExpectMsg<Boolean>("Terminated Cohort") {
+ protected Boolean match(Object in) {
+ if (in instanceof Terminated) {
+ return cohorActorRef.equals(((Terminated) in).actor());
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ Assert.assertTrue(terminatedCohort);
+
+
+ final Boolean terminatedTransaction =
+ new ExpectMsg<Boolean>("Terminated Transaction") {
+ protected Boolean match(Object in) {
+ if (in instanceof Terminated) {
+ return transactionActorRef.equals(((Terminated) in).actor());
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ Assert.assertTrue(terminatedTransaction);
+
final Boolean commitDone =
- new ExpectMsg<Boolean>("match hint") {
+ new ExpectMsg<Boolean>("CommitTransactionReply") {
protected Boolean match(Object in) {
if (in instanceof CommitTransactionReply) {
return true;
};
- }};
+ }
+
+ private ActorRef watchActor(ActorSelection actor) {
+ Future<ActorRef> future = actor
+ .resolveOne(FiniteDuration.apply(100, "milliseconds"));
+
+ try {
+ ActorRef actorRef = Await.result(future,
+ FiniteDuration.apply(100, "milliseconds"));
+
+ watch(actorRef);
+
+ return actorRef;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
}
--- /dev/null
+package org.opendaylight.controller.cluster.datastore;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.junit.Test;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertTrue;
+
+public class DistributedDataStoreIntegrationTest extends AbstractActorTest {
+
+ @Test
+ public void integrationTest() throws Exception {
+ DistributedDataStore distributedDataStore =
+ new DistributedDataStore(getSystem(), "config");
+
+ distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
+
+ DOMStoreReadWriteTransaction transaction =
+ distributedDataStore.newReadWriteTransaction();
+
+ transaction.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
+ transaction.read(TestModel.TEST_PATH);
+
+ Optional<NormalizedNode<?, ?>> optional = future.get();
+
+ NormalizedNode<?, ?> normalizedNode = optional.get();
+
+ assertEquals(TestModel.TEST_QNAME, normalizedNode.getNodeType());
+
+ DOMStoreThreePhaseCommitCohort ready = transaction.ready();
+
+ ListenableFuture<Boolean> canCommit = ready.canCommit();
+
+ assertTrue(canCommit.get());
+
+ ListenableFuture<Void> preCommit = ready.preCommit();
+
+ preCommit.get();
+
+ ListenableFuture<Void> commit = ready.commit();
+
+ commit.get();
+
+ }
+
+}
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import org.junit.Test;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
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 static org.junit.Assert.assertTrue;
-public class ShardTest extends AbstractActorTest{
- @Test
- public void testOnReceiveCreateTransactionChain() throws Exception {
- new JavaTestKit(getSystem()) {{
- final Props props = Shard.props("config");
- final ActorRef subject = getSystem().actorOf(props, "testCreateTransactionChain");
-
- new Within(duration("1 seconds")) {
- protected void run() {
-
- subject.tell(new CreateTransactionChain(), getRef());
-
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof CreateTransactionChainReply) {
- CreateTransactionChainReply reply = (CreateTransactionChainReply) in;
- return reply.getTransactionChainPath().toString();
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- assertTrue(out.matches("akka:\\/\\/test\\/user\\/testCreateTransactionChain\\/\\$.*"));
- // Will wait for the rest of the 3 seconds
- expectNoMsg();
- }
-
-
- };
- }};
- }
-
- @Test
- public void testOnReceiveRegisterListener() throws Exception {
- new JavaTestKit(getSystem()) {{
- final Props props = Shard.props("config");
- final ActorRef subject = getSystem().actorOf(props, "testRegisterChangeListener");
-
- new Within(duration("1 seconds")) {
- protected void run() {
-
- subject.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
-
- subject.tell(new RegisterChangeListener(TestModel.TEST_PATH, getRef().path() , AsyncDataBroker.DataChangeScope.BASE), getRef());
+public class ShardTest extends AbstractActorTest {
+ @Test
+ public void testOnReceiveCreateTransactionChain() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final Props props = Shard.props("config");
+ final ActorRef subject =
+ getSystem().actorOf(props, "testCreateTransactionChain");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(new CreateTransactionChain(), getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof CreateTransactionChainReply) {
+ CreateTransactionChainReply reply =
+ (CreateTransactionChainReply) in;
+ return reply.getTransactionChainPath()
+ .toString();
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertTrue(out.matches(
+ "akka:\\/\\/test\\/user\\/testCreateTransactionChain\\/\\$.*"));
+ // Will wait for the rest of the 3 seconds
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
+
+ @Test
+ public void testOnReceiveRegisterListener() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final Props props = Shard.props("config");
+ final ActorRef subject =
+ getSystem().actorOf(props, "testRegisterChangeListener");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(
+ new UpdateSchemaContext(TestModel.createTestContext()),
+ getRef());
+
+ subject.tell(new RegisterChangeListener(TestModel.TEST_PATH,
+ getRef().path(), AsyncDataBroker.DataChangeScope.BASE),
+ getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof RegisterChangeListenerReply) {
+ RegisterChangeListenerReply reply =
+ (RegisterChangeListenerReply) in;
+ return reply.getListenerRegistrationPath()
+ .toString();
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertTrue(out.matches(
+ "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
+ // Will wait for the rest of the 3 seconds
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
+
+ @Test
+ public void testCreateTransaction(){
+ new JavaTestKit(getSystem()) {{
+ final Props props = Shard.props("config");
+ final ActorRef subject =
+ getSystem().actorOf(props, "testCreateTransaction");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(
+ new UpdateSchemaContext(TestModel.createTestContext()),
+ getRef());
+
+ subject.tell(new CreateTransaction(),
+ getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof CreateTransactionReply) {
+ CreateTransactionReply reply =
+ (CreateTransactionReply) in;
+ return reply.getTransactionPath()
+ .toString();
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertTrue(out.matches(
+ "akka:\\/\\/test\\/user\\/testCreateTransaction\\/\\$.*"));
+ // Will wait for the rest of the 3 seconds
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
+
+
+
+ private AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>> noOpDataChangeListener() {
+ return new AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>() {
+ @Override
+ public void onDataChanged(
+ AsyncDataChangeEvent<InstanceIdentifier, NormalizedNode<?, ?>> change) {
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof RegisterChangeListenerReply) {
- RegisterChangeListenerReply reply = (RegisterChangeListenerReply) in;
- return reply.getListenerRegistrationPath().toString();
- } else {
- throw noMatch();
- }
}
- }.get(); // this extracts the received message
-
- assertTrue(out.matches("akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
- // Will wait for the rest of the 3 seconds
- expectNoMsg();
- }
-
-
- };
- }};
- }
-
-
-
- private AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>> noOpDataChangeListener(){
- return new AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>() {
- @Override
- public void onDataChanged(AsyncDataChangeEvent<InstanceIdentifier, NormalizedNode<?, ?>> change) {
-
- }
- };
- }
+ };
+ }
}
import akka.actor.ActorRef;
import akka.actor.Props;
+import akka.actor.Terminated;
import akka.testkit.JavaTestKit;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import static org.junit.Assert.assertTrue;
public class ShardTransactionTest extends AbstractActorTest {
- private static ListeningExecutorService storeExecutor = MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
+ private static ListeningExecutorService storeExecutor =
+ MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
+
+ private static final InMemoryDOMDataStore store =
+ new InMemoryDOMDataStore("OPER", storeExecutor);
+
+ static {
+ store.onGlobalContextUpdated(TestModel.createTestContext());
+ }
+
+ @Test
+ public void testOnReceiveReadData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject = getSystem().actorOf(props, "testReadData");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(
+ new ReadData(InstanceIdentifier.builder().build()),
+ getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof ReadDataReply) {
+ if (((ReadDataReply) in).getNormalizedNode()
+ != null) {
+ return "match";
+ }
+ return null;
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertEquals("match", out);
+
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
+
+ @Test
+ public void testOnReceiveReadDataWhenDataNotFound() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject = getSystem().actorOf(props, "testReadDataWhenDataNotFound");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(
+ new ReadData(TestModel.TEST_PATH),
+ getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof ReadDataReply) {
+ if (((ReadDataReply) in).getNormalizedNode()
+ == null) {
+ return "match";
+ }
+ return null;
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertEquals("match", out);
+
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
+
+ private void assertModification(final ActorRef subject,
+ final Class<? extends Modification> modificationType) {
+ new JavaTestKit(getSystem()) {{
+ new Within(duration("1 seconds")) {
+ protected void run() {
+ subject
+ .tell(new ShardTransaction.GetCompositedModification(),
+ getRef());
+
+ final CompositeModification compositeModification =
+ new ExpectMsg<CompositeModification>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected CompositeModification match(Object in) {
+ if (in instanceof ShardTransaction.GetCompositeModificationReply) {
+ return ((ShardTransaction.GetCompositeModificationReply) in)
+ .getModification();
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertTrue(
+ compositeModification.getModifications().size() == 1);
+ assertEquals(modificationType,
+ compositeModification.getModifications().get(0)
+ .getClass());
+
+ }
+ };
+ }};
+ }
+
+ @Test
+ public void testOnReceiveWriteData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testWriteData");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(new WriteData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+ getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof WriteDataReply) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertEquals("match", out);
+
+ assertModification(subject, WriteModification.class);
+ expectNoMsg();
+ }
+
+
+ };
+ }};
+ }
- private static final InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", storeExecutor);
+ @Test
+ public void testOnReceiveMergeData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testMergeData");
- static {
- store.onGlobalContextUpdated(TestModel.createTestContext());
- }
+ new Within(duration("1 seconds")) {
+ protected void run() {
- @Test
- public void testOnReceiveReadData() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testReadData");
+ subject.tell(new MergeData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+ getRef());
- new Within(duration("1 seconds")) {
- protected void run() {
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof MergeDataReply) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
- subject.tell(new ReadData(InstanceIdentifier.builder().build()), getRef());
+ assertEquals("match", out);
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof ReadDataReply) {
- if (((ReadDataReply) in).getNormalizedNode() != null) {
- return "match";
+ assertModification(subject, MergeModification.class);
+
+ expectNoMsg();
}
- return null;
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- assertEquals("match", out);
-
- expectNoMsg();
- }
-
-
- };
- }};
- }
-
- private void assertModification(final ActorRef subject, final Class<? extends Modification> modificationType){
- new JavaTestKit(getSystem()) {{
- new Within(duration("1 seconds")) {
- protected void run() {
- subject.tell(new ShardTransaction.GetCompositedModification(), getRef());
-
- final CompositeModification compositeModification = new ExpectMsg<CompositeModification>("match hint") {
- // do not put code outside this method, will run afterwards
- protected CompositeModification match(Object in) {
- if (in instanceof ShardTransaction.GetCompositeModificationReply) {
- return ((ShardTransaction.GetCompositeModificationReply) in).getModification();
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- assertTrue(compositeModification.getModifications().size() == 1);
- assertEquals(modificationType, compositeModification.getModifications().get(0).getClass());
-
- }
- };
- }};
- }
-
- @Test
- public void testOnReceiveWriteData() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testWriteData");
-
- new Within(duration("1 seconds")) {
- protected void run() {
-
- subject.tell(new WriteData(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)), getRef());
-
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof WriteDataReply) {
- return "match";
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- assertEquals("match", out);
-
- assertModification(subject, WriteModification.class);
- expectNoMsg();
- }
-
-
- };
- }};
- }
-
- @Test
- public void testOnReceiveMergeData() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testMergeData");
-
- new Within(duration("1 seconds")) {
- protected void run() {
-
- subject.tell(new MergeData(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)), getRef());
-
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof MergeDataReply) {
- return "match";
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- assertEquals("match", out);
-
- assertModification(subject, MergeModification.class);
-
- expectNoMsg();
- }
-
-
- };
- }};
- }
-
- @Test
- public void testOnReceiveDeleteData() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testDeleteData");
- new Within(duration("1 seconds")) {
- protected void run() {
- subject.tell(new DeleteData(TestModel.TEST_PATH), getRef());
+ };
+ }};
+ }
+
+ @Test
+ public void testOnReceiveDeleteData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testDeleteData");
+
+ new Within(duration("1 seconds")) {
+ protected void run() {
+
+ subject.tell(new DeleteData(TestModel.TEST_PATH), getRef());
+
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof DeleteDataReply) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof DeleteDataReply) {
- return "match";
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
+ assertEquals("match", out);
+
+ assertModification(subject, DeleteModification.class);
+ expectNoMsg();
+ }
- assertEquals("match", out);
- assertModification(subject, DeleteModification.class);
- expectNoMsg();
- }
+ };
+ }};
+ }
- };
- }};
- }
+ @Test
+ public void testOnReceiveReadyTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testReadyTransaction");
+ new Within(duration("1 seconds")) {
+ protected void run() {
- @Test
- public void testOnReceiveReadyTransaction() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testReadyTransaction");
+ subject.tell(new ReadyTransaction(), getRef());
- new Within(duration("1 seconds")) {
- protected void run() {
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof ReadyTransactionReply) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
- subject.tell(new ReadyTransaction(), getRef());
+ assertEquals("match", out);
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof ReadyTransactionReply) {
- return "match";
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
+ expectNoMsg();
+ }
- assertEquals("match", out);
- expectNoMsg();
- }
+ };
+ }};
+ }
- };
- }};
+ @Test
+ public void testOnReceiveCloseTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+ final Props props =
+ ShardTransaction.props(store.newReadWriteTransaction(), shard);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testCloseTransaction");
- }
+ watch(subject);
- @Test
- public void testOnReceiveCloseTransaction() throws Exception {
- new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props("config"));
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
- final ActorRef subject = getSystem().actorOf(props, "testCloseTransaction");
+ new Within(duration("2 seconds")) {
+ protected void run() {
- new Within(duration("1 seconds")) {
- protected void run() {
+ subject.tell(new CloseTransaction(), getRef());
- subject.tell(new CloseTransaction(), getRef());
+ final String out = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof CloseTransactionReply) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
- final String out = new ExpectMsg<String>("match hint") {
- // do not put code outside this method, will run afterwards
- protected String match(Object in) {
- if (in instanceof CloseTransactionReply) {
- return "match";
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
+ assertEquals("match", out);
- assertEquals("match", out);
+ final String termination = new ExpectMsg<String>("match hint") {
+ // do not put code outside this method, will run afterwards
+ protected String match(Object in) {
+ if (in instanceof Terminated) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
- expectNoMsg();
- }
+
+ expectNoMsg();
+ }
- };
- }};
+ };
+ }};
- }
+ }
-}
\ No newline at end of file
+}
--- /dev/null
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import com.google.common.util.concurrent.ListenableFuture;
+import junit.framework.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
+import org.opendaylight.controller.cluster.datastore.utils.MockActorContext;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertNotNull;
+
+public class ThreePhaseCommitCohortProxyTest extends AbstractActorTest {
+
+ private ThreePhaseCommitCohortProxy proxy;
+ private Props props;
+ private ActorRef actorRef;
+ private MockActorContext actorContext;
+
+ @Before
+ public void setUp(){
+ props = Props.create(MessageCollectorActor.class);
+ actorRef = getSystem().actorOf(props);
+ actorContext = new MockActorContext(this.getSystem());
+
+ proxy =
+ new ThreePhaseCommitCohortProxy(actorContext,
+ Arrays.asList(actorRef.path()));
+
+ }
+
+ @Test
+ public void testCanCommit() throws Exception {
+ actorContext.setExecuteRemoteOperationResponse(new CanCommitTransactionReply(true));
+
+ ListenableFuture<Boolean> future = proxy.canCommit();
+
+ Assert.assertTrue(future.get().booleanValue());
+
+ }
+
+ @Test
+ public void testPreCommit() throws Exception {
+ actorContext.setExecuteRemoteOperationResponse(new PreCommitTransactionReply());
+
+ ListenableFuture<Void> future = proxy.preCommit();
+
+ future.get();
+
+ }
+
+ @Test
+ public void testAbort() throws Exception {
+ actorContext.setExecuteRemoteOperationResponse(new AbortTransactionReply());
+
+ ListenableFuture<Void> future = proxy.abort();
+
+ future.get();
+
+ }
+
+ @Test
+ public void testCommit() throws Exception {
+ actorContext.setExecuteRemoteOperationResponse(new CommitTransactionReply());
+
+ ListenableFuture<Void> future = proxy.commit();
+
+ future.get();
+ }
+
+ @Test
+ public void testGetCohortPaths() throws Exception {
+ assertNotNull(proxy.getCohortPaths());
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.utils;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import junit.framework.Assert;
+
+import java.util.List;
+
+public class TestUtils {
+
+ public static void assertFirstSentMessage(ActorSystem actorSystem, ActorRef actorRef, Class clazz){
+ ActorContext testContext = new ActorContext(actorSystem, actorSystem.actorOf(
+ Props.create(DoNothingActor.class)));
+ Object messages = testContext
+ .executeLocalOperation(actorRef, "messages",
+ ActorContext.ASK_DURATION);
+
+ Assert.assertNotNull(messages);
+
+ Assert.assertTrue(messages instanceof List);
+
+ List<Object> listMessages = (List<Object>) messages;
+
+ Assert.assertEquals(1, listMessages.size());
+
+ Assert.assertTrue(listMessages.get(0).getClass().equals(clazz));
+ }
+}
Code Review / controller.git / commitdiff