Merge "BUG 1082 Migrate sal-rest-connector to Async Data Broker API"
authorTony Tkacik <ttkacik@cisco.com>
Sun, 3 Aug 2014 18:10:03 +0000 (18:10 +0000)
committerGerrit Code Review <gerrit@opendaylight.org>
Sun, 3 Aug 2014 18:10:03 +0000 (18:10 +0000)
179 files changed:
.gitignore
features/config-netty/src/main/resources/features.xml
features/nsf/src/main/resources/features.xml
itests/base-features-it/pom.xml [new file with mode: 0644]
itests/base-features-it/src/test/java/org/opendaylight/controller/base/BaseFeatureTest.java [new file with mode: 0644]
itests/pom.xml [new file with mode: 0644]
opendaylight/commons/opendaylight/pom.xml
opendaylight/commons/protocol-framework/pom.xml
opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/AbstractDispatcher.java
opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ProtocolSessionPromise.java
opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/ReconnectPromise.java
opendaylight/commons/protocol-framework/src/main/java/org/opendaylight/protocol/framework/SessionListener.java
opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/ServerTest.java
opendaylight/commons/protocol-framework/src/test/java/org/opendaylight/protocol/framework/SimpleDispatcher.java
opendaylight/config/config-manager/src/main/java/org/opendaylight/controller/config/manager/impl/osgi/mapping/CodecRegistryProvider.java
opendaylight/config/config-netty-config/pom.xml [new file with mode: 0644]
opendaylight/config/config-netty-config/src/main/resources/initial/00-netty.xml [moved from opendaylight/distribution/opendaylight/src/main/resources/configuration/initial/00-netty.xml with 100% similarity]
opendaylight/config/pom.xml
opendaylight/distribution/opendaylight/pom.xml
opendaylight/hosttracker/api/src/main/java/org/opendaylight/controller/hosttracker/IHostTrackerShell.java [new file with mode: 0644]
opendaylight/hosttracker/implementation/src/main/java/org/opendaylight/controller/hosttracker/internal/Activator.java
opendaylight/hosttracker/implementation/src/main/java/org/opendaylight/controller/hosttracker/internal/HostTracker.java
opendaylight/hosttracker/shell/pom.xml [new file with mode: 0644]
opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpFailedARPReqList.java [new file with mode: 0644]
opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpPendingARPReqList.java [new file with mode: 0644]
opendaylight/hosttracker/shell/src/main/resources/OSGI-INF/blueprint/blueprint.xml [new file with mode: 0644]
opendaylight/hosttracker/shell/src/test/java/org/opendaylight/controller/hosttracker/shell/HostTrackerShellTest.java [new file with mode: 0644]
opendaylight/md-sal/md-sal-config/pom.xml [new file with mode: 0644]
opendaylight/md-sal/md-sal-config/src/main/resources/initial/01-md-sal.xml [moved from opendaylight/distribution/opendaylight/src/main/resources/configuration/initial/01-md-sal.xml with 100% similarity]
opendaylight/md-sal/pom.xml
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleActor.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleConfigParamsImpl.java [new file with mode: 0644]
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/Main.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java [new file with mode: 0644]
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java [new file with mode: 0644]
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java [new file with mode: 0644]
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorContext.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorContextImpl.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java
opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MockRaftActorContext.java
opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/config/yang/md/sal/binding/impl/RuntimeMappingModule.java
opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/md/sal/binding/impl/AbstractForwardedDataBroker.java
opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/MountPointManagerImpl.java
opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/md/sal/binding/impl/test/Bug1333DataChangeListenerTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/md/sal/binding/test/DataBrokerTestCustomizer.java
opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/sal/binding/test/util/BindingTestContext.java
opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/sal/binding/test/util/MockSchemaService.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreFactory.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/md/cluster/datastore/model/CarsModel.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/md/cluster/datastore/model/PeopleModel.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/md/cluster/datastore/model/SampleModelsTest.java
opendaylight/md-sal/sal-dom-api/src/main/java/org/opendaylight/controller/sal/core/api/model/SchemaService.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/config/yang/md/sal/dom/impl/DomBrokerImplModule.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/config/yang/md/sal/dom/impl/DomInmemoryDataBrokerModule.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/config/yang/md/sal/dom/impl/SchemaServiceImplSingletonModule.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/compat/BackwardsCompatibleDataBroker.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BackwardsCompatibleMountPoint.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/GlobalBundleScanningSchemaServiceImpl.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/osgi/SchemaServiceProxy.java
opendaylight/md-sal/sal-dom-xsql/src/main/java/org/opendaylight/yang/gen/v1/http/netconfcentral/org/ns/xsql/rev140626/XSQLModule.java
opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/config/yang/inmemory_datastore_provider/InMemoryConfigDataStoreProviderModule.java
opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/config/yang/inmemory_datastore_provider/InMemoryOperationalDataStoreProviderModule.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/config/yang/md/sal/connector/netconf/NetconfConnectorModule.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/NetconfDevice.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicator.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilities.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/NetconfDeviceDataBroker.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/tx/NetconfDeviceReadOnlyTx.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/tx/NetconfDeviceWriteOnlyTx.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/util/NetconfMessageTransformUtil.java
opendaylight/md-sal/sal-netconf-connector/src/main/yang/odl-sal-netconf-connector-cfg.yang
opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicatorTest.java
opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilitiesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/tx/NetconfDeviceWriteOnlyTxTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/README-FIRST
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/NodeToNormalizedNodeBuilder.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/NodeValueCodec.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/NormalizedNodeToNodeCodec.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/NormalizedNodeToProtocolBufferNode.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/AugmentationIdentifierGenerator.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NodeIdentifierFactory.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NodeIdentifierGenerator.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NodeIdentifierWithPredicatesGenerator.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NodeIdentifierWithValueGenerator.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NormalizedNodeGetter.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NormalizedNodeNavigator.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NormalizedNodePrinter.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/NormalizedNodeVisitor.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/PathUtils.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/QNameFactory.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/util/EncoderDecoderUtil.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/cluster/datastore/util/InstanceIdentifierUtils.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/mdsal/CompositeModificationPayload.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/cohort3pc/ThreePhaseCommitCohortMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/common/NormalizedNodeMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/common/SimpleNormalizedNodeMessage.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/datachange/notification/DataChangeListenerMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/persistent/PersistentMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/registration/ListenerRegistrationMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/shard/ShardManagerMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/transaction/ShardTransactionChainMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/main/java/org/opendaylight/controller/protobuff/messages/transaction/ShardTransactionMessages.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/cluster/datastore/node/NormalizedNodeToNodeCodecTest.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/cluster/datastore/node/utils/NodeIdentifierFactoryTest.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/cluster/datastore/util/InstanceIdentifierUtilsTest.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/cluster/datastore/util/NormalizedNodeXmlConverterTest.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/cluster/datastore/util/TestModel.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/AbstractMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/ShardManagerMessagesTest.java [deleted file]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/cohort3pc/ThreePhaseCommitCohortMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/common/NormalizedNodeMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/datachange/notification/DataChangeListenerMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/persistent/PersistentMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/registration/ListenerRegistrationMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/shard/ShardManagerMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/transaction/ShardTransactionChainMessagesTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/java/org/opendaylight/controller/protobuff/messages/transaction/ShardTransactionMessagesTest.java
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/augment_choice.xml
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/simple_xml_with_attributes.xml
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/DataChangeListenerMessagesData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/ListenerRegistrationMessagesData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/NormalizedNodeMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/PersistentMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/ShardManagerMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/ShardTransactionChainMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/ShardTransactionMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/ThreePhaseCommitCohortMessagesTestData [new file with mode: 0644]
opendaylight/md-sal/sal-protocolbuffer-encoding/src/test/resources/version-compatibility-serialized-data/readme.txt [new file with mode: 0644]
opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/ActorConstants.java [new file with mode: 0644]
opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcProvider.java
opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RpcManager.java
opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistry.java
opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/rest/impl/RestconfProviderImpl.java
opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/restconf/impl/ControllerContext.java
opendaylight/md-sal/sal-rest-connector/src/main/java/org/opendaylight/controller/sal/streams/listeners/Notificator.java
opendaylight/md-sal/samples/pom.xml
opendaylight/md-sal/samples/toaster-config/pom.xml [new file with mode: 0644]
opendaylight/md-sal/samples/toaster-config/src/main/resources/initial/03-toaster-sample.xml [moved from opendaylight/distribution/opendaylight/src/main/resources/configuration/initial/03-toaster-sample.xml with 99% similarity]
opendaylight/netconf/netconf-config/pom.xml [new file with mode: 0644]
opendaylight/netconf/netconf-config/src/main/resources/initial/01-netconf.xml [moved from opendaylight/distribution/opendaylight/src/main/resources/configuration/initial/01-netconf.xml with 100% similarity]
opendaylight/netconf/netconf-connector-config/pom.xml [new file with mode: 0644]
opendaylight/netconf/netconf-connector-config/src/main/resources/initial/99-netconf-connector.xml [moved from opendaylight/distribution/opendaylight/src/main/resources/configuration/initial/99-netconf-connector.xml with 100% similarity]
opendaylight/netconf/pom.xml
opendaylight/networkconfiguration/neutron/implementation/src/main/java/org/opendaylight/controller/networkconfig/neutron/implementation/Activator.java
opendaylight/networkconfiguration/neutron/implementation/src/main/java/org/opendaylight/controller/networkconfig/neutron/implementation/NeutronFirewallInterface.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/implementation/src/main/java/org/opendaylight/controller/networkconfig/neutron/implementation/NeutronFirewallPolicyInterface.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/implementation/src/main/java/org/opendaylight/controller/networkconfig/neutron/implementation/NeutronFirewallRuleInterface.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallAware.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallCRUD.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallPolicyAware.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallPolicyCRUD.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallRuleAware.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/INeutronFirewallRuleCRUD.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/NeutronCRUDInterfaces.java
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/NeutronFirewall.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/NeutronFirewallPolicy.java [new file with mode: 0644]
opendaylight/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/NeutronFirewallRule.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallNorthbound.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallPolicyNorthbound.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallPolicyRequest.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallRequest.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallRuleRequest.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronFirewallRulesNorthbound.java [new file with mode: 0644]
opendaylight/northbound/networkconfiguration/neutron/src/main/java/org/opendaylight/controller/networkconfig/neutron/northbound/NeutronNorthboundRSApplication.java
opendaylight/topologymanager/implementation/pom.xml
opendaylight/topologymanager/implementation/src/main/java/org/opendaylight/controller/topologymanager/ITopologyManagerShell.java [new file with mode: 0644]
opendaylight/topologymanager/implementation/src/main/java/org/opendaylight/controller/topologymanager/internal/Activator.java
opendaylight/topologymanager/implementation/src/main/java/org/opendaylight/controller/topologymanager/internal/TopologyManagerImpl.java
opendaylight/topologymanager/shell/pom.xml [new file with mode: 0644]
opendaylight/topologymanager/shell/src/main/java/org/opendaylight/controller/topologymanager/shell/AddUserLink.java [new file with mode: 0644]
opendaylight/topologymanager/shell/src/main/java/org/opendaylight/controller/topologymanager/shell/DeleteUserLink.java [new file with mode: 0644]
opendaylight/topologymanager/shell/src/main/java/org/opendaylight/controller/topologymanager/shell/PrintNodeEdges.java [new file with mode: 0644]
opendaylight/topologymanager/shell/src/main/java/org/opendaylight/controller/topologymanager/shell/PrintUserLink.java [new file with mode: 0644]
opendaylight/topologymanager/shell/src/main/resources/OSGI-INF/blueprint/blueprint.xml [new file with mode: 0644]

index f8cf74f..9144cda 100644 (file)
@@ -27,3 +27,5 @@ out/
 maven-eclipse.xml
 .DS_STORE
 .metadata
+opendaylight/md-sal/sal-distributed-datastore/journal
+
index 3121ca0..f1b2d1f 100644 (file)
@@ -12,5 +12,6 @@
     <bundle>mvn:org.opendaylight.controller/threadpool-config-api/${project.version}</bundle>
     <bundle>mvn:org.opendaylight.controller/threadpool-config-impl/${project.version}</bundle>
     <feature version='${project.version}'>odl-config-startup</feature>
+    <configfile finalname="configuration/initial/00-netty.xml">mvn:org.opendaylight.controller/config-netty-config/${config.version}/xml/config</configfile>
   </feature>
 </features>
\ No newline at end of file
index fb61f33..130d72e 100644 (file)
         <bundle>mvn:org.opendaylight.controller/forwardingrulesmanager.implementation/${forwardingrulesmanager.implementation.version}</bundle>
 
         <bundle>mvn:org.opendaylight.controller/topologymanager/${topologymanager.version}</bundle>
+        <bundle>mvn:org.opendaylight.controller/topologymanager.shell/${topologymanager.shell.version}</bundle>
 
         <bundle>mvn:org.opendaylight.controller/networkconfig.neutron/${networkconfig.neutron.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/networkconfig.neutron.implementation/${networkconfig.neutron.implementation.version}</bundle>
 
         <bundle>mvn:org.opendaylight.controller/hosttracker/${hosttracker.api.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/hosttracker.implementation/${hosttracker.implementation.version}</bundle>
+        <bundle>mvn:org.opendaylight.controller/hosttracker.shell/${hosttracker.shell.version}</bundle>
 
         <bundle>mvn:org.opendaylight.controller/forwarding.staticrouting</bundle>
 
@@ -70,4 +72,4 @@
         <bundle>mvn:org.opendaylight.controller/topology.northbound/${topology.northbound.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/usermanager.northbound/${usermanager.northbound.version}</bundle>
     </feature>
-</features>
\ No newline at end of file
+</features>
diff --git a/itests/base-features-it/pom.xml b/itests/base-features-it/pom.xml
new file mode 100644 (file)
index 0000000..8f73779
--- /dev/null
@@ -0,0 +1,90 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>itests-controller</artifactId>
+        <version>1.4.2-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>base-features-it</artifactId>
+    <name>base-features-it</name>
+    <packaging>jar</packaging>
+    <dependencies>
+        <dependency>
+            <groupId>org.opendaylight.controller</groupId>
+            <artifactId>base-features</artifactId>
+            <version>${project.version}</version>
+            <classifier>features</classifier>
+            <type>xml</type>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <!-- Dependencies for pax exam karaf container -->
+        <dependency>
+            <groupId>org.ops4j.pax.exam</groupId>
+            <artifactId>pax-exam-container-karaf</artifactId>
+            <version>${pax.exam.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.ops4j.pax.exam</groupId>
+            <artifactId>pax-exam-junit4</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.ops4j.pax.exam</groupId>
+            <artifactId>pax-exam</artifactId>
+            <version>${pax.exam.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.ops4j.pax.url</groupId>
+            <artifactId>pax-url-aether</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>javax.inject</groupId>
+            <artifactId>javax.inject</artifactId>
+            <version>1</version>
+            <scope>test</scope>
+        </dependency>
+         <dependency>
+            <groupId>org.apache.karaf.features</groupId>
+            <artifactId>org.apache.karaf.features.core</artifactId>
+            <version>${karaf.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+           <groupId>org.osgi</groupId>
+           <artifactId>org.osgi.core</artifactId>
+           <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- Needed if you use versionAsInProject() -->
+            <!--    <plugin>
+                <groupId>org.apache.servicemix.tooling</groupId>
+                <artifactId>depends-maven-plugin</artifactId>
+                <version>1.2</version>
+                <executions>
+                    <execution>
+                        <id>generate-depends-file</id>
+                        <goals>
+                            <goal>generate-depends-file</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin> -->
+        </plugins>
+    </build>
+</project>
diff --git a/itests/base-features-it/src/test/java/org/opendaylight/controller/base/BaseFeatureTest.java b/itests/base-features-it/src/test/java/org/opendaylight/controller/base/BaseFeatureTest.java
new file mode 100644 (file)
index 0000000..0d38940
--- /dev/null
@@ -0,0 +1,74 @@
+package org.opendaylight.controller.base;
+
+import static org.ops4j.pax.exam.CoreOptions.maven;
+import static org.ops4j.pax.exam.karaf.options.KarafDistributionOption.configureConsole;
+import static org.ops4j.pax.exam.karaf.options.KarafDistributionOption.karafDistributionConfiguration;
+import static org.ops4j.pax.exam.karaf.options.KarafDistributionOption.keepRuntimeFolder;
+import static org.ops4j.pax.exam.karaf.options.KarafDistributionOption.logLevel;
+
+import java.io.File;
+import java.net.URI;
+import java.util.EnumSet;
+
+import javax.inject.Inject;
+
+import junit.framework.Assert;
+
+import org.apache.karaf.features.Feature;
+import org.apache.karaf.features.FeaturesService;
+import org.apache.karaf.features.Repository;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.LogLevelOption.LogLevel;
+
+
+@RunWith(PaxExam.class)
+public class BaseFeatureTest {
+   @Inject
+   private FeaturesService featuresService;
+    @Configuration
+    public Option[] config() {
+       return new Option[] {
+             // Provision and launch a container based on a distribution of Karaf (Apache ServiceMix).
+             karafDistributionConfiguration()
+                 .frameworkUrl(
+                          maven()
+                                  .groupId("org.opendaylight.controller")
+                                  .artifactId("distribution.opendaylight-karaf")
+                                  .type("tar.gz")
+                                  .version("1.4.2-SNAPSHOT"))
+                 .name("OpenDaylight")
+                 .unpackDirectory(new File("target/pax"))
+                 .useDeployFolder(false),
+             // It is really nice if the container sticks around after the test so you can check the contents
+             // of the data directory when things go wrong.
+             keepRuntimeFolder(),
+             // Don't bother with local console output as it just ends up cluttering the logs
+             configureConsole().ignoreLocalConsole(),
+             // Force the log level to INFO so we have more details during the test.  It defaults to WARN.
+             logLevel(LogLevel.WARN),
+             // Remember that the test executes in another process.  If you want to debug it, you need
+             // to tell Pax Exam to launch that process with debugging enabled.  Launching the test class itself with
+             // debugging enabled (for example in Eclipse) will not get you the desired results.
+             //debugConfiguration("5000", true),
+       };
+    }
+
+    @Test
+    public void testAllFeatures() throws Exception {
+       featuresService.addRepository(new URI("mvn:org.opendaylight.controller/base-features/1.4.2-SNAPSHOT/xml/features"));
+       Repository repoUnderTest = featuresService.getRepository("base-1.4.2-SNAPSHOT");
+       Assert.assertNotNull(repoUnderTest);
+       Feature[] featuresUnderTest = repoUnderTest.getFeatures();
+       for(int i=0; i< featuresUnderTest.length; i++)
+       {
+          Feature feature = featuresUnderTest[i];
+          featuresService.installFeature(feature,EnumSet.of(FeaturesService.Option.Verbose));
+          System.out.println("Testing Feature:"+feature.getName());
+          Assert.assertTrue(featuresService.isInstalled(feature));
+       }
+    }
+}
\ No newline at end of file
diff --git a/itests/pom.xml b/itests/pom.xml
new file mode 100644 (file)
index 0000000..19836a2
--- /dev/null
@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.opendaylight.controller</groupId>
+    <artifactId>commons.opendaylight</artifactId>
+    <version>1.4.2-SNAPSHOT</version>
+    <relativePath>../opendaylight/commons/opendaylight</relativePath>
+  </parent>
+  <artifactId>itests-controller</artifactId>
+  <packaging>pom</packaging>
+  <prerequisites>
+    <maven>3.0</maven>
+  </prerequisites>
+  <modules>
+    <module>base-features-it</module>
+  </modules>
+</project>
index 9468133..5ce5840 100644 (file)
@@ -95,6 +95,7 @@
     <hosttracker.api.version>0.5.2-SNAPSHOT</hosttracker.api.version>
     <hosttracker.implementation.version>0.5.2-SNAPSHOT</hosttracker.implementation.version>
     <hosttracker.northbound.version>0.4.2-SNAPSHOT</hosttracker.northbound.version>
+    <hosttracker.shell.version>1.0.0-SNAPSHOT</hosttracker.shell.version>
     <hosttracker_new.api.version>0.4.2-SNAPSHOT</hosttracker_new.api.version>
     <hosttracker_new.implementation.version>0.4.2-SNAPSHOT</hosttracker_new.implementation.version>
     <httpservice-bridge.northbound.version>0.0.2-SNAPSHOT</httpservice-bridge.northbound.version>
     <northbound.jolokia.version>1.4.2-SNAPSHOT</northbound.jolokia.version>
     <opendaylight-l2-types.version>2013.08.27.4-SNAPSHOT</opendaylight-l2-types.version>
     <osgi-brandfragment.web.version>0.0.2-SNAPSHOT</osgi-brandfragment.web.version>
+    <pax.exam.version>4.0.0</pax.exam.version>
     <parboiled.version>1.1.6</parboiled.version>
     <parboiled.scala.version>1.1.6</parboiled.scala.version>
     <propertymavenplugin.version>1.0-alpha-2</propertymavenplugin.version>
     <topology.northbound.version>0.4.2-SNAPSHOT</topology.northbound.version>
     <topology.web.version>0.4.2-SNAPSHOT</topology.web.version>
     <topologymanager.version>0.4.2-SNAPSHOT</topologymanager.version>
+    <topologymanager.shell.version>1.0.0-SNAPSHOT</topologymanager.shell.version>
     <troubleshoot.web.version>0.4.2-SNAPSHOT</troubleshoot.web.version>
     <typesafe.config.version>1.2.0</typesafe.config.version>
     <uncommons.maths.version>1.2.2</uncommons.maths.version>
         <artifactId>sal-rest-connector-config</artifactId>
         <version>${mdsal.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>config-netty-config</artifactId>
+        <version>${config.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>md-sal-config</artifactId>
+        <version>${mdsal.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>netconf-config</artifactId>
+        <version>${netconf.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.opendaylight.controller</groupId>
+        <artifactId>netconf-connector-config</artifactId>
+        <version>${netconf.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.opendaylight.controller</groupId>
         <artifactId>sal-rest-docgen</artifactId>
         <artifactId>sample-toaster-provider</artifactId>
         <version>${mdsal.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.opendaylight.controller.samples</groupId>
+        <artifactId>toaster-config</artifactId>
+        <version>${mdsal.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.opendaylight.controller.thirdparty</groupId>
         <artifactId>com.sun.jersey.jersey-servlet</artifactId>
index f706987..774bc7c 100644 (file)
       <artifactId>netty-event-executor-config</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
index a62bd7d..a05d02c 100644 (file)
@@ -7,12 +7,19 @@
  */
 package org.opendaylight.protocol.framework;
 
+import java.io.Closeable;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.google.common.base.Preconditions;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.Channel;
 import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
@@ -28,13 +35,6 @@ import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import io.netty.util.concurrent.Promise;
 
-import java.io.Closeable;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
  * start method that will handle sockets in different thread.
@@ -155,7 +155,7 @@ public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends
      */
     protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final PipelineInitializer<S> initializer) {
         final Bootstrap b = new Bootstrap();
-        final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<S>(executor, address, strategy, b);
+        final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<>(executor, address, strategy, b);
         b.option(ChannelOption.SO_KEEPALIVE, true).handler(
                 new ChannelInitializer<SocketChannel>() {
                     @Override
@@ -165,18 +165,36 @@ public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends
                 });
 
         customizeBootstrap(b);
+        setWorkerGroup(b);
+        setChannelFactory(b);
+
+        p.connect();
+        LOG.debug("Client created.");
+        return p;
+    }
 
+    private void setWorkerGroup(final Bootstrap b) {
         if (b.group() == null) {
             b.group(workerGroup);
         }
+    }
 
-        // There is no way to detect if this was already set by
-        // customizeBootstrap()
-        try {
-            b.channel(NioSocketChannel.class);
-        } catch (IllegalStateException e) {
-            LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
-        }
+    /**
+     * Create a client but use a pre-configured bootstrap.
+     * This method however replaces the ChannelInitializer in the bootstrap. All other configuration is preserved.
+     *
+     * @param address remote address
+     */
+    protected Future<S> createClient(final InetSocketAddress address, final ReconnectStrategy strategy, final Bootstrap bootstrap, final PipelineInitializer<S> initializer) {
+        final ProtocolSessionPromise<S> p = new ProtocolSessionPromise<>(executor, address, strategy, bootstrap);
+
+        bootstrap.handler(
+                new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    protected void initChannel(final SocketChannel ch) {
+                        initializer.initializeChannel(ch, p);
+                    }
+                });
 
         p.connect();
         LOG.debug("Client created.");
@@ -195,6 +213,9 @@ public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends
     }
 
     /**
+     *
+     * @deprecated use {@link org.opendaylight.protocol.framework.AbstractDispatcher#createReconnectingClient(java.net.InetSocketAddress, ReconnectStrategyFactory, org.opendaylight.protocol.framework.AbstractDispatcher.PipelineInitializer)} with only one reconnectStrategyFactory instead.
+     *
      * Creates a client.
      *
      * @param address remote address
@@ -204,15 +225,47 @@ public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends
      * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
      *         success if it indicates no further attempts should be made and failure if it reports an error
      */
+    @Deprecated
     protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
             final ReconnectStrategy reestablishStrategy, final PipelineInitializer<S> initializer) {
+        return createReconnectingClient(address, connectStrategyFactory, initializer);
+    }
 
-        final ReconnectPromise<S, L> p = new ReconnectPromise<S, L>(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, reestablishStrategy, initializer);
-        p.connect();
+    /**
+     * Creates a reconnecting client.
+     *
+     * @param address remote address
+     * @param connectStrategyFactory Factory for creating reconnection strategy for every reconnect attempt
+     *
+     * @return Future representing the reconnection task. It will report completion based on reestablishStrategy, e.g.
+     *         success if it indicates no further attempts should be made and failure if it reports an error
+     */
+    protected Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory connectStrategyFactory,
+            final PipelineInitializer<S> initializer) {
+        final Bootstrap b = new Bootstrap();
+
+        final ReconnectPromise<S, L> p = new ReconnectPromise<>(GlobalEventExecutor.INSTANCE, this, address, connectStrategyFactory, b, initializer);
+
+        b.option(ChannelOption.SO_KEEPALIVE, true);
 
+        customizeBootstrap(b);
+        setWorkerGroup(b);
+        setChannelFactory(b);
+
+        p.connect();
         return p;
     }
 
+    private void setChannelFactory(final Bootstrap b) {
+        // There is no way to detect if this was already set by
+        // customizeBootstrap()
+        try {
+            b.channel(NioSocketChannel.class);
+        } catch (final IllegalStateException e) {
+            LOG.trace("Not overriding channelFactory on bootstrap {}", b, e);
+        }
+    }
+
     /**
      * @deprecated Should only be used with {@link AbstractDispatcher#AbstractDispatcher()}
      */
@@ -225,5 +278,4 @@ public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends
             this.bossGroup.shutdownGracefully();
         }
     }
-
 }
index a78274c..a38db61 100644 (file)
@@ -7,6 +7,7 @@
  */
 package org.opendaylight.protocol.framework;
 
+import com.google.common.base.Preconditions;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -16,17 +17,12 @@ import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
 import io.netty.util.concurrent.Promise;
-
 import java.net.InetSocketAddress;
-
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Preconditions;
-
 @ThreadSafe
 final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends DefaultPromise<S> {
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolSessionPromise.class);
@@ -54,72 +50,12 @@ final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends Default
             LOG.debug("Promise {} attempting connect for {}ms", lock, timeout);
 
             this.b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout);
-            this.pending = this.b.connect(this.address).addListener(new ChannelFutureListener() {
-                @Override
-                public void operationComplete(final ChannelFuture cf) throws Exception {
-                    synchronized (lock) {
-
-                        LOG.debug("Promise {} connection resolved", lock);
-
-                        // Triggered when a connection attempt is resolved.
-                        Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(cf));
-
-                        /*
-                         * The promise we gave out could have been cancelled,
-                         * which cascades to the connect getting cancelled,
-                         * but there is a slight race window, where the connect
-                         * is already resolved, but the listener has not yet
-                         * been notified -- cancellation at that point won't
-                         * stop the notification arriving, so we have to close
-                         * the race here.
-                         */
-                        if (isCancelled()) {
-                            if (cf.isSuccess()) {
-                                LOG.debug("Closing channel for cancelled promise {}", lock);
-                                cf.channel().close();
-                            }
-                            return;
-                        }
-
-                        if (!cf.isSuccess()) {
-                            LOG.debug("Attempt to connect to {} failed", ProtocolSessionPromise.this.address, cf.cause());
-
-                            final Future<Void> rf = ProtocolSessionPromise.this.strategy.scheduleReconnect(cf.cause());
-                            rf.addListener(new FutureListener<Void>() {
-                                @Override
-                                public void operationComplete(final Future<Void> sf) {
-                                    synchronized (lock) {
-                                        // Triggered when a connection attempt is to be made.
-                                        Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(sf));
-
-                                        /*
-                                         * The promise we gave out could have been cancelled,
-                                         * which cascades to the reconnect attempt getting
-                                         * cancelled, but there is a slight race window, where
-                                         * the reconnect attempt is already enqueued, but the
-                                         * listener has not yet been notified -- if cancellation
-                                         * happens at that point, we need to catch it here.
-                                         */
-                                        if (!isCancelled()) {
-                                            if (sf.isSuccess()) {
-                                                connect();
-                                            } else {
-                                                setFailure(sf.cause());
-                                            }
-                                        }
-                                    }
-                                }
-                            });
-
-                            ProtocolSessionPromise.this.pending = rf;
-                        } else {
-                            LOG.debug("Promise {} connection successful", lock);
-                        }
-                    }
-                }
-            });
+            final ChannelFuture connectFuture = this.b.connect(this.address);
+            // Add listener that attempts reconnect by invoking this method again.
+            connectFuture.addListener(new BootstrapConnectListener(lock));
+            this.pending = connectFuture;
         } catch (final Exception e) {
-            LOG.info("Failed to connect to {}", e);
+            LOG.info("Failed to connect to {}", address, e);
             setFailure(e);
         }
     }
@@ -140,4 +76,79 @@ final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends Default
         this.strategy.reconnectSuccessful();
         return super.setSuccess(result);
     }
+
+    private class BootstrapConnectListener implements ChannelFutureListener {
+        private final Object lock;
+
+        public BootstrapConnectListener(final Object lock) {
+            this.lock = lock;
+        }
+
+        @Override
+        public void operationComplete(final ChannelFuture cf) throws Exception {
+            synchronized (lock) {
+
+                LOG.debug("Promise {} connection resolved", lock);
+
+                // Triggered when a connection attempt is resolved.
+                Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(cf));
+
+                /*
+                 * The promise we gave out could have been cancelled,
+                 * which cascades to the connect getting cancelled,
+                 * but there is a slight race window, where the connect
+                 * is already resolved, but the listener has not yet
+                 * been notified -- cancellation at that point won't
+                 * stop the notification arriving, so we have to close
+                 * the race here.
+                 */
+                if (isCancelled()) {
+                    if (cf.isSuccess()) {
+                        LOG.debug("Closing channel for cancelled promise {}", lock);
+                        cf.channel().close();
+                    }
+                    return;
+                }
+
+                if(cf.isSuccess()) {
+                    LOG.debug("Promise {} connection successful", lock);
+                    return;
+                }
+
+                LOG.debug("Attempt to connect to {} failed", ProtocolSessionPromise.this.address, cf.cause());
+
+                final Future<Void> rf = ProtocolSessionPromise.this.strategy.scheduleReconnect(cf.cause());
+                rf.addListener(new ReconnectingStrategyListener());
+                ProtocolSessionPromise.this.pending = rf;
+            }
+        }
+
+        private class ReconnectingStrategyListener implements FutureListener<Void> {
+            @Override
+            public void operationComplete(final Future<Void> sf) {
+                synchronized (lock) {
+                    // Triggered when a connection attempt is to be made.
+                    Preconditions.checkState(ProtocolSessionPromise.this.pending.equals(sf));
+
+                    /*
+                     * The promise we gave out could have been cancelled,
+                     * which cascades to the reconnect attempt getting
+                     * cancelled, but there is a slight race window, where
+                     * the reconnect attempt is already enqueued, but the
+                     * listener has not yet been notified -- if cancellation
+                     * happens at that point, we need to catch it here.
+                     */
+                    if (!isCancelled()) {
+                        if (sf.isSuccess()) {
+                            connect();
+                        } else {
+                            setFailure(sf.cause());
+                        }
+                    }
+                }
+            }
+        }
+
+    }
+
 }
index 1fa6a81..fe1012f 100644 (file)
  */
 package org.opendaylight.protocol.framework;
 
-import io.netty.channel.ChannelFuture;
+import com.google.common.base.Preconditions;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.FutureListener;
 import io.netty.util.concurrent.Promise;
-
-import java.io.Closeable;
 import java.net.InetSocketAddress;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.opendaylight.protocol.framework.AbstractDispatcher.PipelineInitializer;
-
-import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
+    private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
+
     private final AbstractDispatcher<S, L> dispatcher;
     private final InetSocketAddress address;
     private final ReconnectStrategyFactory strategyFactory;
-    private final ReconnectStrategy strategy;
-    private final PipelineInitializer<S> initializer;
+    private final Bootstrap b;
+    private final AbstractDispatcher.PipelineInitializer<S> initializer;
     private Future<?> pending;
 
-    private final AtomicBoolean negotiationFinished = new AtomicBoolean(false);
-
     public ReconnectPromise(final EventExecutor executor, final AbstractDispatcher<S, L> dispatcher, final InetSocketAddress address,
-            final ReconnectStrategyFactory connectStrategyFactory, final ReconnectStrategy reestablishStrategy,
-            final PipelineInitializer<S> initializer) {
+                            final ReconnectStrategyFactory connectStrategyFactory, final Bootstrap b, final AbstractDispatcher.PipelineInitializer<S> initializer) {
         super(executor);
+        this.b = b;
+        this.initializer = Preconditions.checkNotNull(initializer);
         this.dispatcher = Preconditions.checkNotNull(dispatcher);
         this.address = Preconditions.checkNotNull(address);
         this.strategyFactory = Preconditions.checkNotNull(connectStrategyFactory);
-        this.strategy = Preconditions.checkNotNull(reestablishStrategy);
-        this.initializer = Preconditions.checkNotNull(initializer);
     }
 
-    // FIXME: BUG-190: refactor
-
     synchronized void connect() {
-        negotiationFinished.set(false);
-
         final ReconnectStrategy cs = this.strategyFactory.createReconnectStrategy();
-        final ReconnectStrategy rs = new ReconnectStrategy() {
-            @Override
-            public Future<Void> scheduleReconnect(final Throwable cause) {
-                return cs.scheduleReconnect(cause);
-            }
 
-            @Override
-            public void reconnectSuccessful() {
-                cs.reconnectSuccessful();
-            }
-
-            @Override
-            public int getConnectTimeout() throws Exception {
-                final int cst = cs.getConnectTimeout();
-                final int rst = ReconnectPromise.this.strategy.getConnectTimeout();
-
-                if (cst == 0) {
-                    return rst;
-                }
-                if (rst == 0) {
-                    return cst;
-                }
-                return Math.min(cst, rst);
-            }
-        };
-
-        final Future<S> cf = this.dispatcher.createClient(this.address, rs, new PipelineInitializer<S>() {
+        // Set up a client with pre-configured bootstrap, but add a closed channel handler into the pipeline to support reconnect attempts
+        pending = this.dispatcher.createClient(this.address, cs, b, new AbstractDispatcher.PipelineInitializer<S>() {
             @Override
             public void initializeChannel(final SocketChannel channel, final Promise<S> promise) {
-                addChannelClosedListener(channel.closeFuture());
                 initializer.initializeChannel(channel, promise);
+
+                // add closed channel handler
+                channel.pipeline().addFirst(new ClosedChannelHandler(ReconnectPromise.this));
             }
         });
+    }
 
-        final Object lock = this;
-        this.pending = cf;
+    /**
+     *
+     * @return true if initial connection was established successfully, false if initial connection failed due to e.g. Connection refused, Negotiation failed
+     */
+    private boolean isInitialConnectFinished() {
+        Preconditions.checkNotNull(pending);
+        return pending.isDone() && pending.isSuccess();
+    }
 
-        cf.addListener(new FutureListener<S>() {
+    @Override
+    public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
+        if (super.cancel(mayInterruptIfRunning)) {
+            Preconditions.checkNotNull(pending);
+            this.pending.cancel(mayInterruptIfRunning);
+            return true;
+        }
 
-            @Override
-            public void operationComplete(final Future<S> future) {
-                synchronized (lock) {
-                    if (!future.isSuccess()) {
-                        final Future<Void> rf = ReconnectPromise.this.strategy.scheduleReconnect(cf.cause());
-
-                        if(rf == null) {
-                            // This should reflect: no more reconnecting strategies, enough
-                            // Currently all reconnect strategies fail with exception, should return null
-                            return;
-                        }
-
-                        ReconnectPromise.this.pending = rf;
-
-                        rf.addListener(new FutureListener<Void>() {
-                            @Override
-                            public void operationComplete(final Future<Void> sf) {
-                                synchronized (lock) {
-                                    /*
-                                     * The promise we gave out could have been cancelled,
-                                     * which cascades to the reconnect attempt getting
-                                     * cancelled, but there is a slight race window, where
-                                     * the reconnect attempt is already enqueued, but the
-                                     * listener has not yet been notified -- if cancellation
-                                     * happens at that point, we need to catch it here.
-                                     */
-                                    if (!isCancelled()) {
-                                        if (sf.isSuccess()) {
-                                            connect();
-                                        } else {
-                                            setFailure(sf.cause());
-                                        }
-                                    }
-                                }
-                            }
-                        });
-                    } else {
-                        /*
-                         *  FIXME: BUG-190: we have a slight race window with cancellation
-                         *         here. Analyze and define its semantics.
-                         */
-                        ReconnectPromise.this.strategy.reconnectSuccessful();
-                        negotiationFinished.set(true);
-                    }
-                }
-            }
-        });
+        return false;
     }
 
-    private final ClosedChannelListener closedChannelListener = new ClosedChannelListener();
-
-    class ClosedChannelListener implements Closeable, FutureListener<Void> {
+    /**
+     * Channel handler that responds to channelInactive event and reconnects the session.
+     * Only if the initial connection was successfully established and promise was not canceled.
+     */
+    private static final class ClosedChannelHandler extends ChannelInboundHandlerAdapter {
+        private final ReconnectPromise<?, ?> promise;
 
-        private final AtomicBoolean stop = new AtomicBoolean(false);
+        public ClosedChannelHandler(final ReconnectPromise<?, ?> promise) {
+            this.promise = promise;
+        }
 
         @Override
-        public void operationComplete(final Future<Void> future) throws Exception {
-            if (stop.get()) {
+        public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
+            if (promise.isCancelled()) {
                 return;
             }
 
-            // Start reconnecting crashed session after negotiation was successful
-            if (!negotiationFinished.get()) {
+            // Check if initial connection was fully finished. If the session was dropped during negotiation, reconnect will not happen.
+            // Session can be dropped during negotiation on purpose by the client side and would make no sense to initiate reconnect
+            if (promise.isInitialConnectFinished() == false) {
                 return;
             }
 
-            connect();
-        }
-
-        @Override
-        public void close() {
-            this.stop.set(true);
+            LOG.debug("Reconnecting after connection to {} was dropped", promise.address);
+            promise.connect();
         }
     }
 
-    private void addChannelClosedListener(final ChannelFuture channelFuture) {
-        channelFuture.addListener(closedChannelListener);
-    }
-
-    @Override
-    public synchronized boolean cancel(final boolean mayInterruptIfRunning) {
-        closedChannelListener.close();
-
-        if (super.cancel(mayInterruptIfRunning)) {
-            this.pending.cancel(mayInterruptIfRunning);
-            return true;
-        }
-
-        return false;
-    }
 }
index 3c429fc..a756a0d 100644 (file)
@@ -10,7 +10,7 @@ package org.opendaylight.protocol.framework;
 import java.util.EventListener;
 
 /**
- * Listener that receives session state informations. This interface should be
+ * Listener that receives session state information. This interface should be
  * implemented by a protocol specific abstract class, that is extended by
  * a final class that implements the methods.
  */
index bead1ee..63026e3 100644 (file)
@@ -9,6 +9,14 @@ package org.opendaylight.protocol.framework;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -16,50 +24,139 @@ import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import io.netty.util.concurrent.Promise;
-
+import io.netty.util.concurrent.SucceededFuture;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 public class ServerTest {
     SimpleDispatcher clientDispatcher, dispatcher;
 
-    final SimpleSessionListener pce = new SimpleSessionListener();
-
     SimpleSession session = null;
 
     ChannelFuture server = null;
 
     InetSocketAddress serverAddress;
     private NioEventLoopGroup eventLoopGroup;
-
+    // Dedicated loop group for server, needed for testing reconnection client
+    // With dedicated server group we can simulate session drop by shutting only the server group down
+    private NioEventLoopGroup serverLoopGroup;
 
     @Before
     public void setUp() {
         final int port = 10000 + (int)(10000 * Math.random());
         serverAddress = new InetSocketAddress("127.0.0.1", port);
         eventLoopGroup = new NioEventLoopGroup();
+        serverLoopGroup = new NioEventLoopGroup();
+    }
+
+    @After
+    public void tearDown() throws IOException, InterruptedException, ExecutionException {
+        if(server != null) {
+            this.server.channel().close();
+        }
+        this.eventLoopGroup.shutdownGracefully().get();
+        this.serverLoopGroup.shutdownGracefully().get();
+        try {
+            Thread.sleep(500);
+        } catch (final InterruptedException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     @Test
-    public void testConnectionEstablished() throws Exception {
+    public void testConnectionRefused() throws Exception {
+        this.clientDispatcher = getClientDispatcher();
+
+        final ReconnectStrategy mockReconnectStrategy = getMockedReconnectStrategy();
+
+        this.clientDispatcher.createClient(this.serverAddress,
+                mockReconnectStrategy, new SessionListenerFactory<SimpleSessionListener>() {
+                    @Override
+                    public SimpleSessionListener getSessionListener() {
+                        return new SimpleSessionListener();
+                    }
+                });
+
+        Mockito.verify(mockReconnectStrategy, timeout(5000).atLeast(2)).scheduleReconnect(any(Throwable.class));
+    }
+
+    @Test
+    public void testConnectionReestablishInitial() throws Exception {
+        this.clientDispatcher = getClientDispatcher();
+
+        final ReconnectStrategy mockReconnectStrategy = getMockedReconnectStrategy();
+
+        this.clientDispatcher.createClient(this.serverAddress,
+                mockReconnectStrategy, new SessionListenerFactory<SimpleSessionListener>() {
+                    @Override
+                    public SimpleSessionListener getSessionListener() {
+                        return new SimpleSessionListener();
+                    }
+                });
+
+        Mockito.verify(mockReconnectStrategy, timeout(5000).atLeast(2)).scheduleReconnect(any(Throwable.class));
+
+        final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+        this.dispatcher = getServerDispatcher(p);
+
+        this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
+            @Override
+            public SimpleSessionListener getSessionListener() {
+                return new SimpleSessionListener();
+            }
+        });
+
+        this.server.get();
+
+        assertEquals(true, p.get(3, TimeUnit.SECONDS));
+    }
+
+    @Test
+    public void testConnectionDrop() throws Exception {
         final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
 
-        this.dispatcher = new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
+        this.dispatcher = getServerDispatcher(p);
 
+        this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
             @Override
-            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
-                    final Channel channel, final Promise<SimpleSession> promise) {
-                p.setSuccess(true);
-                return new SimpleSessionNegotiator(promise, channel);
+            public SimpleSessionListener getSessionListener() {
+                return new SimpleSessionListener();
             }
-        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+        });
+
+        this.server.get();
+
+        this.clientDispatcher = getClientDispatcher();
+
+        final ReconnectStrategy reconnectStrategy = getMockedReconnectStrategy();
+        this.session = this.clientDispatcher.createClient(this.serverAddress,
+                reconnectStrategy, new SessionListenerFactory<SimpleSessionListener>() {
+                    @Override
+                    public SimpleSessionListener getSessionListener() {
+                        return new SimpleSessionListener();
+                    }
+                }).get(6, TimeUnit.SECONDS);
+
+        assertEquals(true, p.get(3, TimeUnit.SECONDS));
+
+        shutdownServer();
+
+        // No reconnect should be scheduled after server drops connection with not-reconnecting client
+        verify(reconnectStrategy, times(0)).scheduleReconnect(any(Throwable.class));
+    }
+
+    @Test
+    public void testConnectionReestablishAfterDrop() throws Exception {
+        final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+
+        this.dispatcher = getServerDispatcher(p);
 
         this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
             @Override
@@ -70,13 +167,42 @@ public class ServerTest {
 
         this.server.get();
 
-        this.clientDispatcher = new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
+        this.clientDispatcher = getClientDispatcher();
+
+        final ReconnectStrategyFactory reconnectStrategyFactory = mock(ReconnectStrategyFactory.class);
+        final ReconnectStrategy reconnectStrategy = getMockedReconnectStrategy();
+        doReturn(reconnectStrategy).when(reconnectStrategyFactory).createReconnectStrategy();
+
+        this.clientDispatcher.createReconnectingClient(this.serverAddress,
+                reconnectStrategyFactory, new SessionListenerFactory<SimpleSessionListener>() {
+                    @Override
+                    public SimpleSessionListener getSessionListener() {
+                        return new SimpleSessionListener();
+                    }
+                });
+
+        assertEquals(true, p.get(3, TimeUnit.SECONDS));
+        shutdownServer();
+
+        verify(reconnectStrategyFactory, timeout(20000).atLeast(2)).createReconnectStrategy();
+    }
+
+    @Test
+    public void testConnectionEstablished() throws Exception {
+        final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+
+        this.dispatcher = getServerDispatcher(p);
+
+        this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
             @Override
-            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
-                    final Channel channel, final Promise<SimpleSession> promise) {
-                return new SimpleSessionNegotiator(promise, channel);
+            public SimpleSessionListener getSessionListener() {
+                return new SimpleSessionListener();
             }
-        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+        });
+
+        this.server.get();
+
+        this.clientDispatcher = getClientDispatcher();
 
         this.session = this.clientDispatcher.createClient(this.serverAddress,
                 new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE, 5000), new SessionListenerFactory<SimpleSessionListener>() {
@@ -93,15 +219,7 @@ public class ServerTest {
     public void testConnectionFailed() throws IOException, InterruptedException, ExecutionException, TimeoutException {
         final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
 
-        this.dispatcher = new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
-
-            @Override
-            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
-                    final Channel channel, final Promise<SimpleSession> promise) {
-                p.setSuccess(true);
-                return new SimpleSessionNegotiator(promise, channel);
-            }
-        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+        this.dispatcher = getServerDispatcher(p);
 
         this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
             @Override
@@ -112,13 +230,7 @@ public class ServerTest {
 
         this.server.get();
 
-        this.clientDispatcher = new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
-            @Override
-            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
-                    final Channel channel, final Promise<SimpleSession> promise) {
-                return new SimpleSessionNegotiator(promise, channel);
-            }
-        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+        this.clientDispatcher = getClientDispatcher();
 
         this.session = this.clientDispatcher.createClient(this.serverAddress,
                 new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE, 5000), new SessionListenerFactory<SimpleSessionListener>() {
@@ -138,14 +250,89 @@ public class ServerTest {
         assertFalse(session.isSuccess());
     }
 
-    @After
-    public void tearDown() throws IOException, InterruptedException {
-        this.server.channel().close();
-        this.eventLoopGroup.shutdownGracefully();
-        try {
-            Thread.sleep(500);
-        } catch (final InterruptedException e) {
-            throw new RuntimeException(e);
-        }
+    @Test
+    public void testNegotiationFailedNoReconnect() throws Exception {
+        final Promise<Boolean> p = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
+
+        this.dispatcher = getServerDispatcher(p);
+
+        this.server = this.dispatcher.createServer(this.serverAddress, new SessionListenerFactory<SimpleSessionListener>() {
+            @Override
+            public SimpleSessionListener getSessionListener() {
+                return new SimpleSessionListener();
+            }
+        });
+
+        this.server.get();
+
+        this.clientDispatcher = new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
+            @Override
+            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
+                                                                         final Channel channel, final Promise<SimpleSession> promise) {
+
+                return new SimpleSessionNegotiator(promise, channel) {
+                    @Override
+                    protected void startNegotiation() throws Exception {
+                        negotiationFailed(new IllegalStateException("Negotiation failed"));
+                    }
+                };
+            }
+        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+
+        final ReconnectStrategyFactory reconnectStrategyFactory = mock(ReconnectStrategyFactory.class);
+        final ReconnectStrategy reconnectStrategy = getMockedReconnectStrategy();
+        doReturn(reconnectStrategy).when(reconnectStrategyFactory).createReconnectStrategy();
+
+        this.clientDispatcher.createReconnectingClient(this.serverAddress,
+                reconnectStrategyFactory, new SessionListenerFactory<SimpleSessionListener>() {
+                    @Override
+                    public SimpleSessionListener getSessionListener() {
+                        return new SimpleSessionListener();
+                    }
+                });
+
+
+        // Only one strategy should be created for initial connect, no more = no reconnects
+        verify(reconnectStrategyFactory, times(1)).createReconnectStrategy();
     }
+
+    private SimpleDispatcher getClientDispatcher() {
+        return new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
+            @Override
+            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
+                                                                         final Channel channel, final Promise<SimpleSession> promise) {
+                return new SimpleSessionNegotiator(promise, channel);
+            }
+        }, new DefaultPromise<SimpleSession>(GlobalEventExecutor.INSTANCE), eventLoopGroup);
+    }
+
+    private ReconnectStrategy getMockedReconnectStrategy() throws Exception {
+        final ReconnectStrategy mockReconnectStrategy = mock(ReconnectStrategy.class);
+        final Future<Void> future = new SucceededFuture<>(GlobalEventExecutor.INSTANCE, null);
+        doReturn(future).when(mockReconnectStrategy).scheduleReconnect(any(Throwable.class));
+        doReturn(5000).when(mockReconnectStrategy).getConnectTimeout();
+        doNothing().when(mockReconnectStrategy).reconnectSuccessful();
+        return mockReconnectStrategy;
+    }
+
+
+    private void shutdownServer() throws InterruptedException, ExecutionException {
+        // Shutdown server
+        server.channel().close().get();
+        // Closing server channel does not close established connections, eventLoop has to be closed as well to simulate dropped session
+        serverLoopGroup.shutdownGracefully().get();
+    }
+
+    private SimpleDispatcher getServerDispatcher(final Promise<Boolean> p) {
+        return new SimpleDispatcher(new SessionNegotiatorFactory<SimpleMessage, SimpleSession, SimpleSessionListener>() {
+
+            @Override
+            public SessionNegotiator<SimpleSession> getSessionNegotiator(final SessionListenerFactory<SimpleSessionListener> factory,
+                                                                         final Channel channel, final Promise<SimpleSession> promise) {
+                p.setSuccess(true);
+                return new SimpleSessionNegotiator(promise, channel);
+            }
+        }, null, serverLoopGroup);
+    }
+
 }
index 12aac9e..d837385 100644 (file)
@@ -54,6 +54,10 @@ public class SimpleDispatcher extends AbstractDispatcher<SimpleSession, SimpleSe
         return super.createClient(address, strategy, new SimplePipelineInitializer(listenerFactory));
     }
 
+    public Future<Void> createReconnectingClient(final InetSocketAddress address, final ReconnectStrategyFactory strategy, final SessionListenerFactory<SimpleSessionListener> listenerFactory) {
+        return super.createReconnectingClient(address, strategy, new SimplePipelineInitializer(listenerFactory));
+    }
+
     public ChannelFuture createServer(final InetSocketAddress address, final SessionListenerFactory<SimpleSessionListener> listenerFactory) {
         return super.createServer(address, new SimplePipelineInitializer(listenerFactory));
     }
index ec46219..6050f7c 100644 (file)
@@ -14,7 +14,7 @@ import org.opendaylight.yangtools.sal.binding.generator.api.ClassLoadingStrategy
 import org.opendaylight.yangtools.sal.binding.generator.impl.RuntimeGeneratedMappingServiceImpl;
 import org.opendaylight.yangtools.yang.data.impl.codec.BindingIndependentMappingService;
 import org.opendaylight.yangtools.yang.data.impl.codec.CodecRegistry;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.osgi.framework.BundleContext;
 
 /**
@@ -31,7 +31,7 @@ public class CodecRegistryProvider implements AutoCloseable {
     public CodecRegistryProvider(final ClassLoadingStrategy classLoadingStrategy, final BundleContext context) {
         service = new RuntimeGeneratedMappingServiceImpl(CLASS_POOL, classLoadingStrategy);
         registration = OsgiRegistrationUtil.registerService(context, service,
-                SchemaServiceListener.class, BindingIndependentMappingService.class);
+                SchemaContextListener.class, BindingIndependentMappingService.class);
     }
 
     public CodecRegistry getCodecRegistry() {
diff --git a/opendaylight/config/config-netty-config/pom.xml b/opendaylight/config/config-netty-config/pom.xml
new file mode 100644 (file)
index 0000000..8dc31dc
--- /dev/null
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.opendaylight.controller</groupId>
+    <artifactId>config-subsystem</artifactId>
+    <version>0.2.5-SNAPSHOT</version>
+  </parent>
+  <artifactId>config-netty-config</artifactId>
+  <description>Configuration files for sal-rest-connector</description>
+  <packaging>jar</packaging>
+  <build>
+       <plugins>
+               <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>build-helper-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>attach-artifacts</id>
+            <goals>
+              <goal>attach-artifact</goal>
+            </goals>
+            <phase>package</phase>
+            <configuration>
+              <artifacts>
+                <artifact>
+                  <file>${project.build.directory}/classes/initial/00-netty.xml</file>
+                  <type>xml</type>
+                  <classifier>config</classifier>
+                </artifact>
+              </artifacts>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>
index 66bb01f..26fac47 100644 (file)
@@ -39,6 +39,7 @@
     <module>shutdown-impl</module>
     <module>netconf-config-dispatcher</module>
     <module>config-module-archetype</module>
+    <module>config-netty-config</module>
   </modules>
 
   <dependencies>
index 1d579eb..7c3289a 100644 (file)
             <phase>generate-resources</phase>
             <configuration>
                <outputDirectory>${project.build.directory}/configuration</outputDirectory>
-               <includeArtifactIds>sal-rest-connector-config</includeArtifactIds>
+               <includeArtifactIds>sal-rest-connector-config,config-netty-config,md-sal-config,netconf-config,toaster-config,netconf-connector-config</includeArtifactIds>
                <includes>**\/*.xml</includes>
                <excludeTransitive>true</excludeTransitive>
                <ignorePermissions>false</ignorePermissions>
           <groupId>org.opendaylight.controller</groupId>
           <artifactId>sal-rest-connector-config</artifactId>
         </dependency>
+        <dependency>
+          <groupId>org.opendaylight.controller</groupId>
+          <artifactId>config-netty-config</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>org.opendaylight.controller</groupId>
+          <artifactId>md-sal-config</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>org.opendaylight.controller</groupId>
+          <artifactId>netconf-config</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>org.opendaylight.controller</groupId>
+          <artifactId>netconf-connector-config</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>org.opendaylight.controller.samples</groupId>
+          <artifactId>toaster-config</artifactId>
+        </dependency>
         <dependency>
           <groupId>org.opendaylight.controller</groupId>
           <artifactId>sal-rest-docgen</artifactId>
                 <phase>generate-resources</phase>
                 <configuration>
                    <outputDirectory>${project.build.directory}/configuration</outputDirectory>
-                   <includeArtifactIds>sal-rest-connector-config</includeArtifactIds>
+                   <includeArtifactIds>sal-rest-connector-config,config-netty-config,md-sal-config,netconf-config,toaster-config,netconf-connector-config</includeArtifactIds>
                    <includes>**\/*.xml</includes>
                    <excludeTransitive>true</excludeTransitive>
                    <ignorePermissions>false</ignorePermissions>
diff --git a/opendaylight/hosttracker/api/src/main/java/org/opendaylight/controller/hosttracker/IHostTrackerShell.java b/opendaylight/hosttracker/api/src/main/java/org/opendaylight/controller/hosttracker/IHostTrackerShell.java
new file mode 100644 (file)
index 0000000..5d11be4
--- /dev/null
@@ -0,0 +1,9 @@
+package org.opendaylight.controller.hosttracker;
+
+import java.util.List;
+
+public interface IHostTrackerShell{
+
+    public List<String> dumpPendingArpReqList();
+    public List<String> dumpFailedArpReqList();
+}
\ No newline at end of file
index 825b245..7e52771 100644 (file)
@@ -16,6 +16,7 @@ import java.util.Set;
 import org.apache.felix.dm.Component;
 import org.opendaylight.controller.clustering.services.ICacheUpdateAware;
 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
+import org.opendaylight.controller.hosttracker.IHostTrackerShell;
 import org.opendaylight.controller.hosttracker.IfHostListener;
 import org.opendaylight.controller.hosttracker.IfIptoHost;
 import org.opendaylight.controller.hosttracker.IfNewHostNotify;
@@ -78,6 +79,7 @@ public class Activator extends ComponentActivatorAbstractBase {
                             IInventoryListener.class.getName(),
                             IfIptoHost.class.getName(),
                             IfHostListener.class.getName(),
+                            IHostTrackerShell.class.getName(),
                             ITopologyManagerAware.class.getName(),
                             ICacheUpdateAware.class.getName() }, props);
 
index ce49b59..f728b35 100644 (file)
@@ -39,6 +39,7 @@ import org.opendaylight.controller.clustering.services.IClusterContainerServices
 import org.opendaylight.controller.clustering.services.IClusterServices;
 import org.opendaylight.controller.hosttracker.HostIdFactory;
 import org.opendaylight.controller.hosttracker.IHostId;
+import org.opendaylight.controller.hosttracker.IHostTrackerShell;
 import org.opendaylight.controller.hosttracker.IPHostId;
 import org.opendaylight.controller.hosttracker.IPMacHostId;
 import org.opendaylight.controller.hosttracker.IfHostListener;
@@ -100,7 +101,7 @@ import org.slf4j.LoggerFactory;
  *
  */
 
-public class HostTracker implements IfIptoHost, IfHostListener, ISwitchManagerAware, IInventoryListener,
+public class HostTracker implements IfIptoHost, IfHostListener, IHostTrackerShell, ISwitchManagerAware, IInventoryListener,
         ITopologyManagerAware, ICacheUpdateAware<IHostId, HostNodeConnector>, CommandProvider {
     static final String ACTIVE_HOST_CACHE = "hosttracker.ActiveHosts";
     static final String INACTIVE_HOST_CACHE = "hosttracker.InactiveHosts";
@@ -1618,4 +1619,24 @@ public class HostTracker implements IfIptoHost, IfHostListener, ISwitchManagerAw
         IHostId id = HostIdFactory.create(addr, null);
         return getHostNetworkHierarchy(id);
     }
+
+    @Override
+    public List<String> dumpPendingArpReqList() {
+        ARPPending arphost;
+        List<String> arpReq = new ArrayList<String>();
+        for (Entry<IHostId, ARPPending> entry : ARPPendingList.entrySet()) {
+            arpReq.add(entry.getValue().getHostId().toString());
+        }
+        return arpReq;
+    }
+
+    @Override
+    public List<String> dumpFailedArpReqList() {
+        ARPPending arphost;
+        List<String> arpReq = new ArrayList<String>();
+        for (Entry<IHostId, ARPPending> entry : failedARPReqList.entrySet()) {
+            arpReq.add(entry.getValue().getHostId().toString());
+        }
+        return arpReq;
+    }
 }
diff --git a/opendaylight/hosttracker/shell/pom.xml b/opendaylight/hosttracker/shell/pom.xml
new file mode 100644 (file)
index 0000000..3f73303
--- /dev/null
@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.opendaylight.controller</groupId>
+    <artifactId>commons.opendaylight</artifactId>
+    <version>1.4.2-SNAPSHOT</version>
+    <relativePath>../../commons/opendaylight</relativePath>
+  </parent>
+  <artifactId>hosttracker.shell</artifactId>
+  <version>${hosttracker.shell.version}</version>
+  <packaging>bundle</packaging>
+  <dependencies>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.karaf.shell</groupId>
+      <artifactId>org.apache.karaf.shell.console</artifactId>
+      <version>3.0.0</version>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.opendaylight.controller</groupId>
+      <artifactId>hosttracker.implementation</artifactId>
+      <version>${hosttracker.implementation.version}</version>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.felix</groupId>
+        <artifactId>maven-bundle-plugin</artifactId>
+        <version>${bundle.plugin.version}</version>
+        <configuration>
+          <instructions>
+            <Import-Package>org.apache.felix.service.command,
+              org.apache.karaf.shell.commands,
+              org.apache.karaf.shell.console,
+              *</Import-Package>
+          </instructions>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpFailedARPReqList.java b/opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpFailedARPReqList.java
new file mode 100644 (file)
index 0000000..ec9971f
--- /dev/null
@@ -0,0 +1,28 @@
+package org.opendaylight.controller.hosttracker.shell;
+/**
+* Copyright (c) 2014 Inocybe Technologies, 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
+*/
+
+import org.apache.felix.gogo.commands.Command;
+import org.apache.karaf.shell.console.OsgiCommandSupport;
+import org.opendaylight.controller.hosttracker.IHostTrackerShell;
+
+@Command(scope = "hosttracker", name = "dumpFailedARPReqList", description="Display the dump failed ARPReqList")
+public class DumpFailedARPReqList extends OsgiCommandSupport{
+
+    private IHostTrackerShell hostTracker;
+
+    @Override
+    protected Object doExecute() throws Exception {
+        System.out.print(hostTracker.dumpFailedArpReqList());
+        return null;
+    }
+
+    public void setHostTracker(IHostTrackerShell hostTracker){
+        this.hostTracker = hostTracker;
+    }
+}
diff --git a/opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpPendingARPReqList.java b/opendaylight/hosttracker/shell/src/main/java/org/opendaylight/controller/hosttracker/shell/DumpPendingARPReqList.java
new file mode 100644 (file)
index 0000000..7f52a55
--- /dev/null
@@ -0,0 +1,28 @@
+package org.opendaylight.controller.hosttracker.shell;
+/**
+* Copyright (c) 2014 Inocybe Technologies, 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
+*/
+
+import org.apache.felix.gogo.commands.Command;
+import org.apache.karaf.shell.console.OsgiCommandSupport;
+import org.opendaylight.controller.hosttracker.IHostTrackerShell;
+
+@Command(scope = "hosttracker", name = "dumpPendingARPReqList", description="Display the dump pending ARPReqList")
+public class DumpPendingARPReqList extends OsgiCommandSupport{
+
+    private IHostTrackerShell hostTracker;
+
+    @Override
+    protected Object doExecute() throws Exception {
+        System.out.print(hostTracker.dumpPendingArpReqList());
+        return null;
+    }
+
+    public void setHostTracker(IHostTrackerShell hostTracker){
+        this.hostTracker = hostTracker;
+    }
+}
\ No newline at end of file
diff --git a/opendaylight/hosttracker/shell/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/opendaylight/hosttracker/shell/src/main/resources/OSGI-INF/blueprint/blueprint.xml
new file mode 100644 (file)
index 0000000..ba79b5d
--- /dev/null
@@ -0,0 +1,17 @@
+<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0">
+    <reference id="HostTrackerRef" interface="org.opendaylight.controller.hosttracker.IHostTrackerShell"/>
+
+    <command-bundle xmlns="http://karaf.apache.org/xmlns/shell/v1.1.0">
+
+    <command>
+       <action class="org.opendaylight.controller.hosttracker.shell.DumpFailedARPReqList">
+          <property name="hostTracker" ref="HostTrackerRef"/>
+       </action>
+    </command>
+    <command>
+       <action class="org.opendaylight.controller.hosttracker.shell.DumpPendingARPReqList">
+          <property name="hostTracker" ref="HostTrackerRef"/>
+       </action>
+    </command>
+    </command-bundle>
+</blueprint>
diff --git a/opendaylight/hosttracker/shell/src/test/java/org/opendaylight/controller/hosttracker/shell/HostTrackerShellTest.java b/opendaylight/hosttracker/shell/src/test/java/org/opendaylight/controller/hosttracker/shell/HostTrackerShellTest.java
new file mode 100644 (file)
index 0000000..a0a5a2e
--- /dev/null
@@ -0,0 +1,47 @@
+package org.opendaylight.controller.hosttracker.shell;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import org.opendaylight.controller.hosttracker.IHostTrackerShell;
+
+public class HostTrackerShellTest {
+
+    private final long COMMAND_TIMEOUT = 1000;
+    private IHostTrackerShell hostTracker;
+
+    @Test
+    public void testDumpPendingARPReqList() throws Exception {
+        DumpPendingARPReqList dumpPendTest = new DumpPendingARPReqList();
+        hostTracker = mock(IHostTrackerShell.class);
+        List<String> failedList = new ArrayList<String>(Arrays.asList("a", "b", "c"));
+        when(hostTracker.dumpPendingArpReqList()).thenReturn(failedList);
+        dumpPendTest.setHostTracker(hostTracker);
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(baos));
+        dumpPendTest.doExecute();
+        //Assert.assertTrue(true);
+        Assert.assertEquals("[a, b, c]", baos.toString());
+    }
+
+    @Test
+    public void testDumpFailedARPReqList() throws Exception {
+        DumpFailedARPReqList dumpFailTest = new DumpFailedARPReqList();
+        hostTracker = mock(IHostTrackerShell.class);
+        List<String> failedList = new ArrayList<String>(Arrays.asList("a", "b", "c"));
+        when(hostTracker.dumpFailedArpReqList()).thenReturn(failedList);
+        dumpFailTest.setHostTracker(hostTracker);
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(baos));
+        dumpFailTest.doExecute();
+        //Assert.assertTrue(true);
+        Assert.assertEquals("[a, b, c]", baos.toString());
+    }
+}
diff --git a/opendaylight/md-sal/md-sal-config/pom.xml b/opendaylight/md-sal/md-sal-config/pom.xml
new file mode 100644 (file)
index 0000000..2e19b5a
--- /dev/null
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.opendaylight.controller</groupId>
+    <artifactId>sal-parent</artifactId>
+    <version>1.1-SNAPSHOT</version>
+  </parent>
+  <artifactId>md-sal-config</artifactId>
+  <description>Configuration files for md-sal</description>
+  <packaging>jar</packaging>
+  <build>
+    <plugins>
+        <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>build-helper-maven-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>attach-artifacts</id>
+            <goals>
+              <goal>attach-artifact</goal>
+            </goals>
+            <phase>package</phase>
+            <configuration>
+              <artifacts>
+                <artifact>
+                  <file>${project.build.directory}/classes/initial/01-md-sal.xml</file>
+                  <type>xml</type>
+                  <classifier>config</classifier>
+                </artifact>
+              </artifacts>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>
index 3f02765..02fbde8 100644 (file)
@@ -32,6 +32,9 @@
 
     <module>sal-binding-util</module>
 
+    <!-- Configuration File -->
+    <module>md-sal-config</module>
+
     <!-- Samples -->
     <module>samples</module>
 
@@ -69,6 +72,9 @@
     <!-- sal-distributed-datastore -->
     <module>sal-distributed-datastore</module>
 
+    <!-- XSQL -->
+    <module>sal-dom-xsql</module>
+
     <!-- Yang Test Models for MD-SAL -->
     <module>sal-test-model</module>
 
index 641ec05..aa100df 100644 (file)
@@ -11,10 +11,12 @@ package org.opendaylight.controller.cluster.example;
 import akka.actor.ActorRef;
 import akka.actor.Props;
 import akka.japi.Creator;
+import com.google.common.base.Optional;
 import org.opendaylight.controller.cluster.example.messages.KeyValue;
 import org.opendaylight.controller.cluster.example.messages.KeyValueSaved;
 import org.opendaylight.controller.cluster.example.messages.PrintRole;
 import org.opendaylight.controller.cluster.example.messages.PrintState;
+import org.opendaylight.controller.cluster.raft.ConfigParams;
 import org.opendaylight.controller.cluster.raft.RaftActor;
 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
 
@@ -31,15 +33,17 @@ public class ExampleActor extends RaftActor {
     private long persistIdentifier = 1;
 
 
-    public ExampleActor(String id, Map<String, String> peerAddresses) {
-        super(id, peerAddresses);
+    public ExampleActor(String id, Map<String, String> peerAddresses,
+        Optional<ConfigParams> configParams) {
+        super(id, peerAddresses, configParams);
     }
 
-    public static Props props(final String id, final Map<String, String> peerAddresses){
+    public static Props props(final String id, final Map<String, String> peerAddresses,
+        final Optional<ConfigParams> configParams){
         return Props.create(new Creator<ExampleActor>(){
 
             @Override public ExampleActor create() throws Exception {
-                return new ExampleActor(id, peerAddresses);
+                return new ExampleActor(id, peerAddresses, configParams);
             }
         });
     }
@@ -56,10 +60,12 @@ public class ExampleActor extends RaftActor {
             }
 
         } else if (message instanceof PrintState) {
-            LOG.debug("State of the node:"+getId() + " has = "+state.size() + " entries");
+            LOG.debug("State of the node:{} has entries={}, {}",
+                getId(), state.size(), getReplicatedLogState());
 
         } else if (message instanceof PrintRole) {
-            LOG.debug(getId() + " = " + getRaftState());
+            LOG.debug("{} = {}, Peers={}", getId(), getRaftState(),getPeers());
+
         } else {
             super.onReceiveCommand(message);
         }
@@ -83,6 +89,7 @@ public class ExampleActor extends RaftActor {
     @Override protected void applySnapshot(Object snapshot) {
         state.clear();
         state.putAll((HashMap) snapshot);
+        LOG.debug("Snapshot applied to state :" + ((HashMap) snapshot).size());
     }
 
     @Override public void onReceiveRecover(Object message) {
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleConfigParamsImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/ExampleConfigParamsImpl.java
new file mode 100644 (file)
index 0000000..d11377d
--- /dev/null
@@ -0,0 +1,20 @@
+/*
+ * 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.example;
+
+import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
+
+/**
+ * Implementation of ConfigParams for Example
+ */
+public class ExampleConfigParamsImpl extends DefaultConfigParamsImpl {
+    @Override
+    public long getSnapshotBatchCount() {
+        return 50;
+    }
+}
index a148ed4..0e5d643 100644 (file)
@@ -11,7 +11,9 @@ package org.opendaylight.controller.cluster.example;
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import akka.actor.PoisonPill;
+import com.google.common.base.Optional;
 import org.opendaylight.controller.cluster.example.messages.KeyValue;
+import org.opendaylight.controller.cluster.raft.ConfigParams;
 
 import java.io.BufferedReader;
 import java.io.InputStreamReader;
@@ -34,15 +36,15 @@ public class Main {
     public static void main(String[] args) throws Exception{
         ActorRef example1Actor =
             actorSystem.actorOf(ExampleActor.props("example-1",
-                withoutPeer("example-1")), "example-1");
+                withoutPeer("example-1"), Optional.<ConfigParams>absent()), "example-1");
 
         ActorRef example2Actor =
             actorSystem.actorOf(ExampleActor.props("example-2",
-                withoutPeer("example-2")), "example-2");
+                withoutPeer("example-2"), Optional.<ConfigParams>absent()), "example-2");
 
         ActorRef example3Actor =
             actorSystem.actorOf(ExampleActor.props("example-3",
-                withoutPeer("example-3")), "example-3");
+                withoutPeer("example-3"), Optional.<ConfigParams>absent()), "example-3");
 
 
         List<ActorRef> examples = Arrays.asList(example1Actor, example2Actor, example3Actor);
@@ -74,7 +76,8 @@ public class Main {
                         String actorName = "example-" + i;
                         examples.add(i - 1,
                             actorSystem.actorOf(ExampleActor.props(actorName,
-                                withoutPeer(actorName)), actorName));
+                                withoutPeer(actorName), Optional.<ConfigParams>absent()),
+                                actorName));
                         System.out.println("Created actor : " + actorName);
                         continue;
                     }
index c2d0b3a..fd6e192 100644 (file)
@@ -2,8 +2,10 @@ package org.opendaylight.controller.cluster.example;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
+import com.google.common.base.Optional;
 import org.opendaylight.controller.cluster.example.messages.PrintRole;
 import org.opendaylight.controller.cluster.example.messages.PrintState;
+import org.opendaylight.controller.cluster.raft.ConfigParams;
 import org.opendaylight.controller.cluster.raft.client.messages.AddRaftPeer;
 import org.opendaylight.controller.cluster.raft.client.messages.RemoveRaftPeer;
 
@@ -11,14 +13,13 @@ import java.io.BufferedReader;
 import java.io.InputStreamReader;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * This is a test driver for testing akka-raft implementation
  * Its uses ExampleActors and threads to push content(key-vals) to these actors
  * Each ExampleActor can have one or more ClientActors. Each ClientActor spawns
- * a thread and starts push logs to the actor its assignged to.
+ * a thread and starts push logs to the actor its assigned to.
  */
 public class TestDriver {
 
@@ -26,7 +27,9 @@ public class TestDriver {
     private static Map<String, String> allPeers = new HashMap<>();
     private static Map<String, ActorRef> clientActorRefs  = new HashMap<String, ActorRef>();
     private static Map<String, ActorRef> actorRefs = new HashMap<String, ActorRef>();
-    private static LogGenerator logGenerator = new LogGenerator();;
+    private static LogGenerator logGenerator = new LogGenerator();
+    private int nameCounter = 0;
+    private static ConfigParams configParams = new ExampleConfigParamsImpl();
 
     /**
      * Create nodes, add clients and start logging.
@@ -35,6 +38,7 @@ public class TestDriver {
      *  createNodes:{num}
      *  addNodes:{num}
      *  stopNode:{nodeName}
+     *  reinstateNode:{nodeName}
      *  addClients:{num}
      *  addClientsToNode:{nodeName, num}
      *  startLogging
@@ -83,6 +87,10 @@ public class TestDriver {
                 String[] arr = command.split(":");
                 td.stopNode(arr[1]);
 
+            } else if (command.startsWith("reinstateNode")) {
+                String[] arr = command.split(":");
+                td.reinstateNode(arr[1]);
+
             } else if (command.startsWith("startLogging")) {
                 td.startAllLogging();
 
@@ -106,15 +114,19 @@ public class TestDriver {
         }
     }
 
+    public static ActorRef createExampleActor(String name) {
+        return actorSystem.actorOf(ExampleActor.props(name, withoutPeer(name),
+            Optional.of(configParams)), name);
+    }
+
     public void createNodes(int num) {
         for (int i=0; i < num; i++)  {
-            int rand = getUnusedRandom(num);
-            allPeers.put("example-"+rand, "akka://default/user/example-"+rand);
+            nameCounter = nameCounter + 1;
+            allPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
         }
 
         for (String s : allPeers.keySet())  {
-            ActorRef exampleActor = actorSystem.actorOf(
-                ExampleActor.props(s, withoutPeer(s)), s);
+            ActorRef exampleActor = createExampleActor(s);
             actorRefs.put(s, exampleActor);
             System.out.println("Created node:"+s);
 
@@ -125,15 +137,14 @@ public class TestDriver {
     public void addNodes(int num) {
         Map<String, String> newPeers = new HashMap<>();
         for (int i=0; i < num; i++)  {
-            int rand = getUnusedRandom(num);
-            newPeers.put("example-"+rand, "akka://default/user/example-"+rand);
-            allPeers.put("example-"+rand, "akka://default/user/example-"+rand);
+            nameCounter = nameCounter + 1;
+            newPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
+            allPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
 
         }
         Map<String, ActorRef> newActorRefs = new HashMap<String, ActorRef>(num);
         for (Map.Entry<String, String> entry : newPeers.entrySet())  {
-            ActorRef exampleActor = actorSystem.actorOf(
-                ExampleActor.props(entry.getKey(), withoutPeer(entry.getKey())), entry.getKey());
+            ActorRef exampleActor = createExampleActor(entry.getKey());
             newActorRefs.put(entry.getKey(), exampleActor);
 
             //now also add these new nodes as peers from the previous nodes
@@ -165,7 +176,7 @@ public class TestDriver {
     public void addClientsToNode(String actorName, int num) {
         ActorRef actorRef = actorRefs.get(actorName);
         for (int i=0; i < num; i++) {
-            String clientName = "client-" + i + "-" + actorRef;
+            String clientName = "client-" + i + "-" + actorName;
             clientActorRefs.put(clientName,
                 actorSystem.actorOf(ClientActor.props(actorRef), clientName));
             System.out.println("Added client-node:" + clientName);
@@ -174,11 +185,13 @@ public class TestDriver {
 
     public void stopNode(String actorName) {
         ActorRef actorRef = actorRefs.get(actorName);
-        String clientName = "client-"+actorName;
-        if(clientActorRefs.containsKey(clientName)) {
-            actorSystem.stop(clientActorRefs.get(clientName));
-            clientActorRefs.remove(clientName);
+
+        for (Map.Entry<String,ActorRef> entry : clientActorRefs.entrySet()) {
+            if (entry.getKey().endsWith(actorName)) {
+                actorSystem.stop(entry.getValue());
+            }
         }
+
         actorSystem.stop(actorRef);
         actorRefs.remove(actorName);
 
@@ -187,7 +200,21 @@ public class TestDriver {
         }
 
         allPeers.remove(actorName);
+    }
 
+    public void reinstateNode(String actorName) {
+        String address = "akka://default/user/"+actorName;
+        allPeers.put(actorName, address);
+
+        ActorRef exampleActor = createExampleActor(actorName);
+
+        for (ActorRef actor : actorRefs.values()) {
+            actor.tell(new AddRaftPeer(actorName, address), null);
+        }
+
+        actorRefs.put(actorName, exampleActor);
+
+        addClientsToNode(actorName, 1);
     }
 
     public void startAllLogging() {
@@ -232,14 +259,6 @@ public class TestDriver {
         return null;
     }
 
-    private int getUnusedRandom(int num) {
-        int rand = -1;
-        do {
-            rand = (new Random()).nextInt(num * num);
-        } while (allPeers.keySet().contains("example-"+rand));
-
-        return rand;
-    }
 
     private static Map<String, String> withoutPeer(String peerId) {
         Map<String, String> without = new ConcurrentHashMap<>(allPeers);
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
new file mode 100644 (file)
index 0000000..24bfa3d
--- /dev/null
@@ -0,0 +1,153 @@
+/*
+ * 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.raft;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Abstract class handling the mapping of
+ * logical LogEntry Index and the physical list index.
+ */
+public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
+
+    protected final List<ReplicatedLogEntry> journal;
+    protected final Object snapshot;
+    protected long snapshotIndex = -1;
+    protected long snapshotTerm = -1;
+
+    public AbstractReplicatedLogImpl(Object state, long snapshotIndex,
+        long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries) {
+        this.snapshot = state;
+        this.snapshotIndex = snapshotIndex;
+        this.snapshotTerm = snapshotTerm;
+        this.journal = new ArrayList<>(unAppliedEntries);
+    }
+
+
+    public AbstractReplicatedLogImpl() {
+        this.snapshot = null;
+        this.journal = new ArrayList<>();
+    }
+
+    protected int adjustedIndex(long logEntryIndex) {
+        if(snapshotIndex < 0){
+            return (int) logEntryIndex;
+        }
+        return (int) (logEntryIndex - (snapshotIndex + 1));
+    }
+
+    @Override
+    public ReplicatedLogEntry get(long logEntryIndex) {
+        int adjustedIndex = adjustedIndex(logEntryIndex);
+
+        if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
+            // physical index should be less than list size and >= 0
+            return null;
+        }
+
+        return journal.get(adjustedIndex);
+    }
+
+    @Override
+    public ReplicatedLogEntry last() {
+        if (journal.isEmpty()) {
+            return null;
+        }
+        // get the last entry directly from the physical index
+        return journal.get(journal.size() - 1);
+    }
+
+    @Override
+    public long lastIndex() {
+        if (journal.isEmpty()) {
+            // it can happen that after snapshot, all the entries of the
+            // journal are trimmed till lastApplied, so lastIndex = snapshotIndex
+            return snapshotIndex;
+        }
+        return last().getIndex();
+    }
+
+    @Override
+    public long lastTerm() {
+        if (journal.isEmpty()) {
+            // it can happen that after snapshot, all the entries of the
+            // journal are trimmed till lastApplied, so lastTerm = snapshotTerm
+            return snapshotTerm;
+        }
+        return last().getTerm();
+    }
+
+    @Override
+    public void removeFrom(long logEntryIndex) {
+        int adjustedIndex = adjustedIndex(logEntryIndex);
+        if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
+            // physical index should be less than list size and >= 0
+            return;
+        }
+        journal.subList(adjustedIndex , journal.size()).clear();
+    }
+
+    @Override
+    public void append(ReplicatedLogEntry replicatedLogEntry) {
+        journal.add(replicatedLogEntry);
+    }
+
+    @Override
+    public List<ReplicatedLogEntry> getFrom(long logEntryIndex) {
+        int adjustedIndex = adjustedIndex(logEntryIndex);
+        int size = journal.size();
+        List<ReplicatedLogEntry> entries = new ArrayList<>(100);
+        if (adjustedIndex >= 0 && adjustedIndex < size) {
+            // physical index should be less than list size and >= 0
+            entries.addAll(journal.subList(adjustedIndex, size));
+        }
+        return entries;
+    }
+
+    @Override
+    public long size() {
+       return journal.size();
+    }
+
+    @Override
+    public boolean isPresent(long logEntryIndex) {
+        if (logEntryIndex > lastIndex()) {
+            // if the request logical index is less than the last present in the list
+            return false;
+        }
+        int adjustedIndex = adjustedIndex(logEntryIndex);
+        return (adjustedIndex >= 0);
+    }
+
+    @Override
+    public boolean isInSnapshot(long logEntryIndex) {
+        return logEntryIndex <= snapshotIndex;
+    }
+
+    @Override
+    public Object getSnapshot() {
+        return snapshot;
+    }
+
+    @Override
+    public long getSnapshotIndex() {
+        return snapshotIndex;
+    }
+
+    @Override
+    public long getSnapshotTerm() {
+        return snapshotTerm;
+    }
+
+    @Override
+    public abstract void appendAndPersist(ReplicatedLogEntry replicatedLogEntry);
+
+    @Override
+    public abstract void removeFromAndPersist(long index);
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ConfigParams.java
new file mode 100644 (file)
index 0000000..4c6434a
--- /dev/null
@@ -0,0 +1,55 @@
+/*
+ * 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.raft;
+
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * Configuration Parameter interface for configuring the Raft consensus system
+ * <p/>
+ * Any component using this implementation might want to provide an implementation of
+ * this interface to configure
+ *
+ * A default implementation will be used if none is provided.
+ *
+ * @author Kamal Rameshan
+ */
+public interface ConfigParams {
+    /**
+     * The minimum number of entries to be present in the in-memory Raft log
+     * for a snapshot to be taken
+     *
+     * @return long
+     */
+    public long getSnapshotBatchCount();
+
+    /**
+     * The interval at which a heart beat message will be sent to the remote
+     * RaftActor
+     *
+     * @return FiniteDuration
+     */
+    public FiniteDuration getHeartBeatInterval();
+
+    /**
+     * The interval in which a new election would get triggered if no leader is found
+     *
+     * Normally its set to atleast twice the heart beat interval
+     *
+     * @return FiniteDuration
+     */
+    public FiniteDuration getElectionTimeOutInterval();
+
+    /**
+     * The maximum election time variance. The election is scheduled using both
+     * the Election Timeout and Variance
+     *
+     * @return int
+     */
+    public int getElectionTimeVariance();
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/DefaultConfigParamsImpl.java
new file mode 100644 (file)
index 0000000..c633337
--- /dev/null
@@ -0,0 +1,61 @@
+/*
+ * 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.raft;
+
+import scala.concurrent.duration.FiniteDuration;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Default implementation of the ConfigParams
+ *
+ * If no implementation is provided for ConfigParams, then this will be used.
+ */
+public class DefaultConfigParamsImpl implements ConfigParams {
+
+    private static final int SNAPSHOT_BATCH_COUNT = 100000;
+
+    /**
+     * The maximum election time variance
+     */
+    private static final int ELECTION_TIME_MAX_VARIANCE = 100;
+
+
+    /**
+     * The interval at which a heart beat message will be sent to the remote
+     * RaftActor
+     * <p/>
+     * Since this is set to 100 milliseconds the Election timeout should be
+     * at least 200 milliseconds
+     */
+    protected static final FiniteDuration HEART_BEAT_INTERVAL =
+        new FiniteDuration(100, TimeUnit.MILLISECONDS);
+
+
+    @Override
+    public long getSnapshotBatchCount() {
+        return SNAPSHOT_BATCH_COUNT;
+    }
+
+    @Override
+    public FiniteDuration getHeartBeatInterval() {
+        return HEART_BEAT_INTERVAL;
+    }
+
+
+    @Override
+    public FiniteDuration getElectionTimeOutInterval() {
+        // returns 2 times the heart beat interval
+        return HEART_BEAT_INTERVAL.$times(2);
+    }
+
+    @Override
+    public int getElectionTimeVariance() {
+        return ELECTION_TIME_MAX_VARIANCE;
+    }
+}
index b8e9653..70b85b4 100644 (file)
@@ -22,6 +22,7 @@ import akka.persistence.UntypedPersistentActor;
 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
+import com.google.common.base.Optional;
 import org.opendaylight.controller.cluster.raft.behaviors.Candidate;
 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
@@ -33,7 +34,6 @@ import org.opendaylight.controller.cluster.raft.client.messages.RemoveRaftPeer;
 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
 
 import java.io.Serializable;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
@@ -100,14 +100,22 @@ public abstract class RaftActor extends UntypedPersistentActor {
 
 
     public RaftActor(String id, Map<String, String> peerAddresses) {
+        this(id, peerAddresses, Optional.<ConfigParams>absent());
+    }
+
+    public RaftActor(String id, Map<String, String> peerAddresses,
+         Optional<ConfigParams> configParams) {
+
         context = new RaftActorContextImpl(this.getSelf(),
-            this.getContext(),
-            id, new ElectionTermImpl(),
-            -1, -1, replicatedLog, peerAddresses, LOG);
+            this.getContext(), id, new ElectionTermImpl(),
+            -1, -1, replicatedLog, peerAddresses,
+            (configParams.isPresent() ? configParams.get(): new DefaultConfigParamsImpl()),
+            LOG);
     }
 
     @Override public void onReceiveRecover(Object message) {
         if (message instanceof SnapshotOffer) {
+            LOG.debug("SnapshotOffer called..");
             SnapshotOffer offer = (SnapshotOffer) message;
             Snapshot snapshot = (Snapshot) offer.snapshot();
 
@@ -116,6 +124,13 @@ public abstract class RaftActor extends UntypedPersistentActor {
             // when we need to install it on a peer
             replicatedLog = new ReplicatedLogImpl(snapshot);
 
+            context.setReplicatedLog(replicatedLog);
+
+            LOG.debug("Applied snapshot to replicatedLog. " +
+                "snapshotIndex={}, snapshotTerm={}, journal-size={}",
+                replicatedLog.snapshotIndex, replicatedLog.snapshotTerm,
+                replicatedLog.size());
+
             // Apply the snapshot to the actors state
             applySnapshot(snapshot.getState());
 
@@ -127,7 +142,11 @@ public abstract class RaftActor extends UntypedPersistentActor {
             context.getTermInformation().update(((UpdateElectionTerm) message).getCurrentTerm(), ((UpdateElectionTerm) message).getVotedFor());
         } else if (message instanceof RecoveryCompleted) {
             LOG.debug(
-                "Last index in log : " + replicatedLog.lastIndex());
+                "RecoveryCompleted - Switching actor to Follower - " +
+                    "Last index in log:{}, snapshotIndex={}, snapshotTerm={}, " +
+                    "journal-size={}",
+                replicatedLog.lastIndex(), replicatedLog.snapshotIndex,
+                replicatedLog.snapshotTerm, replicatedLog.size());
             currentBehavior = switchBehavior(RaftState.Follower);
         }
     }
@@ -191,6 +210,15 @@ public abstract class RaftActor extends UntypedPersistentActor {
         }
     }
 
+    public java.util.Set<String> getPeers() {
+        return context.getPeerAddresses().keySet();
+    }
+
+    protected String getReplicatedLogState() {
+        return "snapshotIndex=" + context.getReplicatedLog().getSnapshotIndex()
+            + ", snapshotTerm=" + context.getReplicatedLog().getSnapshotTerm()
+            + ", im-mem journal size=" + context.getReplicatedLog().size();
+    }
 
 
     /**
@@ -343,85 +371,33 @@ public abstract class RaftActor extends UntypedPersistentActor {
     }
 
     private void trimPersistentData(long sequenceNumber) {
-        // Trim snapshots
+        // Trim akka snapshots
         // FIXME : Not sure how exactly the SnapshotSelectionCriteria is applied
         // For now guessing that it is ANDed.
         deleteSnapshots(new SnapshotSelectionCriteria(
-            sequenceNumber - 100000, 43200000));
+            sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
 
-        // Trim journal
+        // Trim akka journal
         deleteMessages(sequenceNumber);
     }
 
 
-    private class ReplicatedLogImpl implements ReplicatedLog {
-        private final List<ReplicatedLogEntry> journal;
-        private final Object snapshot;
-        private long snapshotIndex = -1;
-        private long snapshotTerm = -1;
+    private class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
 
         public ReplicatedLogImpl(Snapshot snapshot) {
-            this.snapshot = snapshot.getState();
-            this.snapshotIndex = snapshot.getLastAppliedIndex();
-            this.snapshotTerm = snapshot.getLastAppliedTerm();
-
-            this.journal = new ArrayList<>(snapshot.getUnAppliedEntries());
+            super(snapshot.getState(),
+                snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm(),
+                snapshot.getUnAppliedEntries());
         }
 
         public ReplicatedLogImpl() {
-            this.snapshot = null;
-            this.journal = new ArrayList<>();
+            super();
         }
 
-        @Override public ReplicatedLogEntry get(long index) {
-            int adjustedIndex = adjustedIndex(index);
+        @Override public void removeFromAndPersist(long logEntryIndex) {
+            int adjustedIndex = adjustedIndex(logEntryIndex);
 
-            if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
-                return null;
-            }
-
-            return journal.get(adjustedIndex);
-        }
-
-        @Override public ReplicatedLogEntry last() {
-            if (journal.size() == 0) {
-                return null;
-            }
-            return get(journal.size() - 1);
-        }
-
-        @Override public long lastIndex() {
-            if (journal.size() == 0) {
-                return -1;
-            }
-
-            return last().getIndex();
-        }
-
-        @Override public long lastTerm() {
-            if (journal.size() == 0) {
-                return -1;
-            }
-
-            return last().getTerm();
-        }
-
-
-        @Override public void removeFrom(long index) {
-            int adjustedIndex = adjustedIndex(index);
-
-            if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
-                return;
-            }
-
-            journal.subList(adjustedIndex , journal.size()).clear();
-        }
-
-
-        @Override public void removeFromAndPersist(long index) {
-            int adjustedIndex = adjustedIndex(index);
-
-            if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
+            if (adjustedIndex < 0) {
                 return;
             }
 
@@ -435,29 +411,6 @@ public abstract class RaftActor extends UntypedPersistentActor {
                     //FIXME : Doing nothing for now
                 }
             });
-
-
-        }
-
-        @Override public void append(
-            final ReplicatedLogEntry replicatedLogEntry) {
-            journal.add(replicatedLogEntry);
-        }
-
-        @Override public List<ReplicatedLogEntry> getFrom(long index) {
-            int adjustedIndex = adjustedIndex(index);
-
-            List<ReplicatedLogEntry> entries = new ArrayList<>(100);
-            if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
-                return entries;
-            }
-
-
-            for (int i = adjustedIndex;
-                 i < journal.size(); i++) {
-                entries.add(journal.get(i));
-            }
-            return entries;
         }
 
         @Override public void appendAndPersist(
@@ -482,20 +435,42 @@ public abstract class RaftActor extends UntypedPersistentActor {
                 new Procedure<ReplicatedLogEntry>() {
                     public void apply(ReplicatedLogEntry evt) throws Exception {
                         // FIXME : Tentatively create a snapshot every hundred thousand entries. To be tuned.
-                        if (size() > 100000) {
-                            ReplicatedLogEntry lastAppliedEntry =
-                                get(context.getLastApplied());
+                        if (journal.size() > context.getConfigParams().getSnapshotBatchCount()) {
+                            LOG.info("Initiating Snapshot Capture..");
                             long lastAppliedIndex = -1;
                             long lastAppliedTerm = -1;
+
+                            ReplicatedLogEntry lastAppliedEntry = get(context.getLastApplied());
                             if (lastAppliedEntry != null) {
                                 lastAppliedIndex = lastAppliedEntry.getIndex();
                                 lastAppliedTerm = lastAppliedEntry.getTerm();
                             }
 
-                            saveSnapshot(Snapshot.create(createSnapshot(),
+                            LOG.debug("Snapshot Capture logSize: {}", journal.size());
+                            LOG.debug("Snapshot Capture lastApplied:{} ", context.getLastApplied());
+                            LOG.debug("Snapshot Capture lastAppliedIndex:{}", lastAppliedIndex);
+                            LOG.debug("Snapshot Capture lastAppliedTerm:{}", lastAppliedTerm);
+
+                            // create a snapshot object from the state provided and save it
+                            // when snapshot is saved async, SaveSnapshotSuccess is raised.
+                            Snapshot sn = Snapshot.create(createSnapshot(),
                                 getFrom(context.getLastApplied() + 1),
                                 lastIndex(), lastTerm(), lastAppliedIndex,
-                                lastAppliedTerm));
+                                lastAppliedTerm);
+                            saveSnapshot(sn);
+
+                            LOG.info("Persisting of snapshot done:{}", sn.getLogMessage());
+
+                            //be greedy and remove entries from in-mem journal which are in the snapshot
+                            // and update snapshotIndex and snapshotTerm without waiting for the success,
+                            // TODO: damage-recovery to be done on failure
+                            journal.subList(0, (int) (lastAppliedIndex - snapshotIndex)).clear();
+                            snapshotIndex = lastAppliedIndex;
+                            snapshotTerm = lastAppliedTerm;
+
+                            LOG.info("Removed in-memory snapshotted entries, " +
+                                "adjusted snaphsotIndex:{}" +
+                                "and term:{}", snapshotIndex, lastAppliedTerm);
                         }
                         // Send message for replication
                         if (clientActor != null) {
@@ -509,46 +484,8 @@ public abstract class RaftActor extends UntypedPersistentActor {
             );
         }
 
-        @Override public long size() {
-            return journal.size() + snapshotIndex + 1;
-        }
-
-        @Override public boolean isPresent(long index) {
-            int adjustedIndex = adjustedIndex(index);
-
-            if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
-                return false;
-            }
-            return true;
-        }
-
-        @Override public boolean isInSnapshot(long index) {
-            return index <= snapshotIndex;
-        }
-
-        @Override public Object getSnapshot() {
-            return snapshot;
-        }
-
-        @Override public long getSnapshotIndex() {
-            return snapshotIndex;
-        }
-
-        @Override public long getSnapshotTerm() {
-            return snapshotTerm;
-        }
-
-        private int adjustedIndex(long index) {
-            if(snapshotIndex < 0){
-                return (int) index;
-            }
-            return (int) (index - snapshotIndex);
-        }
     }
 
-
-
-
     private static class DeleteEntries implements Serializable {
         private final int fromIndex;
 
@@ -609,6 +546,17 @@ public abstract class RaftActor extends UntypedPersistentActor {
         public long getLastAppliedTerm() {
             return lastAppliedTerm;
         }
+
+        public String getLogMessage() {
+            StringBuilder sb = new StringBuilder();
+            return sb.append("Snapshot={")
+                .append("lastTerm:" + this.getLastTerm()  + ", ")
+                .append("LastAppliedIndex:" + this.getLastAppliedIndex()  + ", ")
+                .append("LastAppliedTerm:" + this.getLastAppliedTerm()  + ", ")
+                .append("UnAppliedEntries size:" + this.getUnAppliedEntries().size()  + "}")
+                .toString();
+
+        }
     }
 
     private class ElectionTermImpl implements ElectionTerm {
index ae9431a..0eb4b73 100644 (file)
@@ -85,6 +85,12 @@ public interface RaftActorContext {
      */
     void setLastApplied(long lastApplied);
 
+    /**
+     *
+     * @param replicatedLog
+     */
+    public void setReplicatedLog(ReplicatedLog replicatedLog);
+
     /**
      * @return A representation of the log
      */
@@ -155,4 +161,9 @@ public interface RaftActorContext {
      * @param peerAddress
      */
     void setPeerAddress(String peerId, String peerAddress);
+
+    /**
+     * @return ConfigParams
+     */
+    public ConfigParams getConfigParams();
 }
index 833c8a9..25da371 100644 (file)
@@ -19,7 +19,7 @@ import java.util.Map;
 
 import static com.google.common.base.Preconditions.checkState;
 
-public class RaftActorContextImpl implements RaftActorContext{
+public class RaftActorContextImpl implements RaftActorContext {
 
     private final ActorRef actor;
 
@@ -33,16 +33,20 @@ public class RaftActorContextImpl implements RaftActorContext{
 
     private long lastApplied;
 
-    private final ReplicatedLog replicatedLog;
+    private ReplicatedLog replicatedLog;
 
     private final Map<String, String> peerAddresses;
 
     private final LoggingAdapter LOG;
 
+    private final ConfigParams configParams;
+
     public RaftActorContextImpl(ActorRef actor, UntypedActorContext context,
         String id,
         ElectionTerm termInformation, long commitIndex,
-        long lastApplied, ReplicatedLog replicatedLog, Map<String, String> peerAddresses, LoggingAdapter logger) {
+        long lastApplied, ReplicatedLog replicatedLog,
+        Map<String, String> peerAddresses, ConfigParams configParams,
+        LoggingAdapter logger) {
         this.actor = actor;
         this.context = context;
         this.id = id;
@@ -51,6 +55,7 @@ public class RaftActorContextImpl implements RaftActorContext{
         this.lastApplied = lastApplied;
         this.replicatedLog = replicatedLog;
         this.peerAddresses = peerAddresses;
+        this.configParams = configParams;
         this.LOG = logger;
     }
 
@@ -90,6 +95,10 @@ public class RaftActorContextImpl implements RaftActorContext{
         this.lastApplied = lastApplied;
     }
 
+    @Override public void setReplicatedLog(ReplicatedLog replicatedLog) {
+        this.replicatedLog = replicatedLog;
+    }
+
     @Override public ReplicatedLog getReplicatedLog() {
         return replicatedLog;
     }
@@ -110,6 +119,10 @@ public class RaftActorContextImpl implements RaftActorContext{
         return peerAddresses.get(peerId);
     }
 
+    @Override public ConfigParams getConfigParams() {
+        return configParams;
+    }
+
     @Override public void addToPeers(String name, String address) {
         peerAddresses.put(name, address);
     }
index f7281bb..0a553b4 100644 (file)
@@ -43,27 +43,6 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      */
     protected final RaftActorContext context;
 
-    /**
-     * The maximum election time variance
-     */
-    private static final int ELECTION_TIME_MAX_VARIANCE = 100;
-
-    /**
-     * The interval at which a heart beat message will be sent to the remote
-     * RaftActor
-     * <p/>
-     * Since this is set to 100 milliseconds the Election timeout should be
-     * at least 200 milliseconds
-     */
-    protected static final FiniteDuration HEART_BEAT_INTERVAL =
-        new FiniteDuration(100, TimeUnit.MILLISECONDS);
-
-    /**
-     * The interval in which a new election would get triggered if no leader is found
-     */
-    private static final long ELECTION_TIME_INTERVAL =
-        HEART_BEAT_INTERVAL.toMillis() * 2;
-
     /**
      *
      */
@@ -208,9 +187,9 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @return
      */
     protected FiniteDuration electionDuration() {
-        long variance = new Random().nextInt(ELECTION_TIME_MAX_VARIANCE);
-        return new FiniteDuration(ELECTION_TIME_INTERVAL + variance,
-            TimeUnit.MILLISECONDS);
+        long variance = new Random().nextInt(context.getConfigParams().getElectionTimeVariance());
+        return context.getConfigParams().getElectionTimeOutInterval().$plus(
+            new FiniteDuration(variance, TimeUnit.MILLISECONDS));
     }
 
     /**
index db62dfc..c8cd41d 100644 (file)
@@ -126,15 +126,10 @@ public class Follower extends AbstractRaftActorBehavior {
             int addEntriesFrom = 0;
             if (context.getReplicatedLog().size() > 0) {
 
-                // Find the entry up until which the one that is not in the
-                // follower's log
-                for (int i = 0;
-                     i < appendEntries.getEntries()
-                         .size(); i++, addEntriesFrom++) {
-                    ReplicatedLogEntry matchEntry =
-                        appendEntries.getEntries().get(i);
-                    ReplicatedLogEntry newEntry = context.getReplicatedLog()
-                        .get(matchEntry.getIndex());
+                // Find the entry up until which the one that is not in the follower's log
+                for (int i = 0;i < appendEntries.getEntries().size(); i++, addEntriesFrom++) {
+                    ReplicatedLogEntry matchEntry = appendEntries.getEntries().get(i);
+                    ReplicatedLogEntry newEntry = context.getReplicatedLog().get(matchEntry.getIndex());
 
                     if (newEntry == null) {
                         //newEntry not found in the log
index 53e47c2..2a44e8b 100644 (file)
@@ -112,8 +112,8 @@ public class Leader extends AbstractRaftActorBehavior {
         scheduleHeartBeat(new FiniteDuration(0, TimeUnit.SECONDS));
 
         scheduleInstallSnapshotCheck(
-            new FiniteDuration(HEART_BEAT_INTERVAL.length() * 1000,
-                HEART_BEAT_INTERVAL.unit())
+            new FiniteDuration(context.getConfigParams().getHeartBeatInterval().length() * 1000,
+                context.getConfigParams().getHeartBeatInterval().unit())
         );
 
     }
@@ -241,7 +241,7 @@ public class Leader extends AbstractRaftActorBehavior {
                     (InstallSnapshotReply) message);
             }
         } finally {
-            scheduleHeartBeat(HEART_BEAT_INTERVAL);
+            scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
         }
 
         return super.handleMessage(sender, message);
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java
new file mode 100644 (file)
index 0000000..ae8e525
--- /dev/null
@@ -0,0 +1,153 @@
+/*
+ * 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.raft;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
+import static org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockReplicatedLogEntry;
+/**
+*
+*/
+public class AbstractReplicatedLogImplTest {
+
+    private MockAbstractReplicatedLogImpl replicatedLogImpl;
+
+    @Before
+    public void setUp() {
+        replicatedLogImpl = new MockAbstractReplicatedLogImpl();
+    }
+
+    @After
+    public void tearDown() {
+        replicatedLogImpl.journal.clear();
+        replicatedLogImpl.setSnapshotIndex(-1);
+        replicatedLogImpl.setSnapshotTerm(-1);
+        replicatedLogImpl = null;
+    }
+
+    @Test
+    public void testIndexOperations() {
+        // create a set of initial entries in the in-memory log
+        replicatedLogImpl.append(new MockReplicatedLogEntry(1, 0, new MockPayload("A")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(1, 1, new MockPayload("B")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(1, 2, new MockPayload("C")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(2, 3, new MockPayload("D")));
+
+        // check if the values returned are correct, with snapshotIndex = -1
+        assertEquals("B", replicatedLogImpl.get(1).getData().toString());
+        assertEquals("D", replicatedLogImpl.last().getData().toString());
+        assertEquals(3, replicatedLogImpl.lastIndex());
+        assertEquals(2, replicatedLogImpl.lastTerm());
+        assertEquals(2, replicatedLogImpl.getFrom(2).size());
+        assertEquals(4, replicatedLogImpl.size());
+        assertTrue(replicatedLogImpl.isPresent(2));
+        assertFalse(replicatedLogImpl.isPresent(4));
+        assertFalse(replicatedLogImpl.isInSnapshot(2));
+
+        // now create a snapshot of 3 entries, with 1 unapplied entry left in the log
+        // It removes the entries which have made it to snapshot
+        // and updates the snapshot index and term
+        Map state = takeSnapshot(3);
+
+        // check the values after the snapshot.
+        // each index value passed in the test is the logical index (log entry index)
+        // which gets mapped to the list's physical index
+        assertEquals("D", replicatedLogImpl.get(3).getData().toString());
+        assertEquals("D", replicatedLogImpl.last().getData().toString());
+        assertNull(replicatedLogImpl.get(1));
+        assertEquals(3, replicatedLogImpl.lastIndex());
+        assertEquals(2, replicatedLogImpl.lastTerm());
+        assertEquals(0, replicatedLogImpl.getFrom(2).size());
+        assertEquals(1, replicatedLogImpl.size());
+        assertFalse(replicatedLogImpl.isPresent(2));
+        assertTrue(replicatedLogImpl.isPresent(3));
+        assertFalse(replicatedLogImpl.isPresent(4));
+        assertTrue(replicatedLogImpl.isInSnapshot(2));
+
+        // append few more entries
+        replicatedLogImpl.append(new MockReplicatedLogEntry(2, 4, new MockPayload("E")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(2, 5, new MockPayload("F")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(3, 6, new MockPayload("G")));
+        replicatedLogImpl.append(new MockReplicatedLogEntry(3, 7, new MockPayload("H")));
+
+        // check their values as well
+        assertEquals(5, replicatedLogImpl.size());
+        assertEquals("D", replicatedLogImpl.get(3).getData().toString());
+        assertEquals("E", replicatedLogImpl.get(4).getData().toString());
+        assertEquals("H", replicatedLogImpl.last().getData().toString());
+        assertEquals(3, replicatedLogImpl.lastTerm());
+        assertEquals(7, replicatedLogImpl.lastIndex());
+        assertTrue(replicatedLogImpl.isPresent(7));
+        assertFalse(replicatedLogImpl.isInSnapshot(7));
+        assertEquals(1, replicatedLogImpl.getFrom(7).size());
+        assertEquals(2, replicatedLogImpl.getFrom(6).size());
+
+        // take a second snapshot with 5 entries with 0 unapplied entries left in the log
+        state = takeSnapshot(5);
+
+        assertEquals(0, replicatedLogImpl.size());
+        assertNull(replicatedLogImpl.last());
+        assertNull(replicatedLogImpl.get(7));
+        assertNull(replicatedLogImpl.get(1));
+        assertFalse(replicatedLogImpl.isPresent(7));
+        assertTrue(replicatedLogImpl.isInSnapshot(7));
+        assertEquals(0, replicatedLogImpl.getFrom(7).size());
+        assertEquals(0, replicatedLogImpl.getFrom(6).size());
+
+    }
+
+    // create a snapshot for test
+    public Map takeSnapshot(int numEntries) {
+        Map map = new HashMap(numEntries);
+        List<ReplicatedLogEntry> entries = replicatedLogImpl.getEntriesTill(numEntries);
+        for (ReplicatedLogEntry entry : entries) {
+            map.put(entry.getIndex(), entry.getData().toString());
+        }
+
+        int term = (int) replicatedLogImpl.lastTerm();
+        int lastIndex = (int) entries.get(entries.size() - 1).getIndex();
+        entries.clear();
+        replicatedLogImpl.setSnapshotTerm(term);
+        replicatedLogImpl.setSnapshotIndex(lastIndex);
+
+        return map;
+
+    }
+    class MockAbstractReplicatedLogImpl extends AbstractReplicatedLogImpl {
+        @Override
+        public void appendAndPersist(ReplicatedLogEntry replicatedLogEntry) {
+        }
+
+        @Override
+        public void removeFromAndPersist(long index) {
+        }
+
+        public void setSnapshotIndex(long snapshotIndex) {
+            this.snapshotIndex = snapshotIndex;
+        }
+
+        public void setSnapshotTerm(long snapshotTerm) {
+            this.snapshotTerm = snapshotTerm;
+        }
+
+        public List<ReplicatedLogEntry> getEntriesTill(int index) {
+            return journal.subList(0, index);
+        }
+    }
+}
index 2e200cb..aa50fa7 100644 (file)
@@ -177,7 +177,10 @@ public class MockRaftActorContext implements RaftActorContext {
         this.peerAddresses = peerAddresses;
     }
 
-
+    @Override
+    public ConfigParams getConfigParams() {
+        return new DefaultConfigParamsImpl();
+    }
 
     public static class SimpleReplicatedLog implements ReplicatedLog {
         private final List<ReplicatedLogEntry> log = new ArrayList<>();
index 823a4d9..b0c2d74 100644 (file)
@@ -23,7 +23,7 @@ import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.impl.codec.BindingIndependentMappingService;
 import org.opendaylight.yangtools.yang.data.impl.codec.CodecRegistry;
 import org.opendaylight.yangtools.yang.data.impl.codec.DeserializationException;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.ServiceReference;
 import org.slf4j.Logger;
@@ -74,7 +74,7 @@ public final class RuntimeMappingModule extends
         }
 
         final RuntimeGeneratedMappingServiceImpl service = new RuntimeGeneratedMappingServiceImpl(SingletonHolder.CLASS_POOL);
-        bundleContext.registerService(SchemaServiceListener.class, service, new Hashtable<String,String>());
+        bundleContext.registerService(SchemaContextListener.class, service, new Hashtable<String,String>());
         return service;
     }
 
index 15e4a46..2d81b60 100644 (file)
@@ -41,7 +41,6 @@ import org.opendaylight.yangtools.yang.data.impl.codec.BindingIndependentMapping
 import org.opendaylight.yangtools.yang.data.impl.codec.DeserializationException;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,14 +57,14 @@ public abstract class AbstractForwardedDataBroker implements Delegator<DOMDataBr
     private final BindingToNormalizedNodeCodec codec;
     private BindingIndependentConnector connector;
     private ProviderSession context;
-    private final ListenerRegistration<SchemaServiceListener> schemaListenerRegistration;
+    private final ListenerRegistration<SchemaContextListener> schemaListenerRegistration;
 
     protected AbstractForwardedDataBroker(final DOMDataBroker domDataBroker,
             final BindingIndependentMappingService mappingService,final SchemaService schemaService) {
         this.domDataBroker = domDataBroker;
         this.mappingService = mappingService;
         this.codec = new BindingToNormalizedNodeCodec(mappingService);
-        this.schemaListenerRegistration = schemaService.registerSchemaServiceListener(this);
+        this.schemaListenerRegistration = schemaService.registerSchemaContextListener(this);
     }
 
     protected BindingToNormalizedNodeCodec getCodec() {
index df09f78..05651bf 100644 (file)
@@ -9,6 +9,7 @@ package org.opendaylight.controller.sal.binding.impl;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
 
 import org.opendaylight.controller.md.sal.binding.util.AbstractBindingSalProviderInstance;
 import org.opendaylight.controller.sal.binding.api.mount.MountProviderInstance;
@@ -20,6 +21,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 
 public class MountPointManagerImpl implements MountProviderService {
 
@@ -82,7 +84,7 @@ public class MountPointManagerImpl implements MountProviderService {
         RpcProviderRegistryImpl rpcRegistry = new RpcProviderRegistryImpl("mount");
         NotificationBrokerImpl notificationBroker = new NotificationBrokerImpl(getNotificationExecutor());
         DataBrokerImpl dataBroker = new DataBrokerImpl();
-        dataBroker.setExecutor(getDataCommitExecutor());
+        dataBroker.setExecutor(MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()));
         BindingMountPointImpl mountInstance = new BindingMountPointImpl(path, rpcRegistry, notificationBroker,
                 dataBroker);
         mountPoints.putIfAbsent(path, mountInstance);
diff --git a/opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/md/sal/binding/impl/test/Bug1333DataChangeListenerTest.java b/opendaylight/md-sal/sal-binding-broker/src/test/java/org/opendaylight/controller/md/sal/binding/impl/test/Bug1333DataChangeListenerTest.java
new file mode 100644 (file)
index 0000000..60d56db
--- /dev/null
@@ -0,0 +1,145 @@
+/*
+ * 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.md.sal.binding.impl.test;
+
+import static org.opendaylight.controller.md.sal.binding.test.AssertCollections.assertContains;
+import static org.opendaylight.controller.md.sal.binding.test.AssertCollections.assertEmpty;
+import static org.opendaylight.controller.md.sal.binding.test.AssertCollections.assertNotContains;
+import static org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.TOP_FOO_KEY;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.USES_ONE_KEY;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.USES_TWO_KEY;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.complexUsesAugment;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.path;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.top;
+import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.topLevelList;
+
+import org.junit.Test;
+import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
+import org.opendaylight.controller.md.sal.binding.test.AbstractDataChangeListenerTest;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.augment.rev140709.TreeComplexUsesAugment;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.list.rev140701.Top;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.list.rev140701.two.level.list.TopLevelList;
+import org.opendaylight.yangtools.yang.binding.DataObject;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+
+/**
+ *
+ * This testsuite tries to replicate bug 1333 and tests regresion of it
+ * using test-model with similar construction as one reported.
+ *
+ *
+ * See  https://bugs.opendaylight.org/show_bug.cgi?id=1333 for Bug Description
+ *
+ */
+public class Bug1333DataChangeListenerTest extends AbstractDataChangeListenerTest{
+
+    private static final InstanceIdentifier<Top> TOP_PATH = InstanceIdentifier.create(Top.class);
+
+    private static final InstanceIdentifier<?> AUGMENT_WILDCARD = TOP_PATH.child(TopLevelList.class).augmentation(
+            TreeComplexUsesAugment.class);
+
+    public void writeTopWithListItem(final LogicalDatastoreType store) {
+        ReadWriteTransaction tx = getDataBroker().newReadWriteTransaction();
+        Top topItem = top(topLevelList(TOP_FOO_KEY, complexUsesAugment(USES_ONE_KEY, USES_TWO_KEY)));
+        tx.put(store, TOP_PATH, topItem);
+        assertCommit(tx.submit());
+    }
+
+    public void deleteItem(final LogicalDatastoreType store, final InstanceIdentifier<?> path) {
+        ReadWriteTransaction tx = getDataBroker().newReadWriteTransaction();
+        tx.delete(store, path);
+        assertCommit(tx.submit());
+    }
+
+    @Test
+    public void writeTopWithListItemAugmentedListenTopSubtree() {
+        TestListener listener = createListener(CONFIGURATION,TOP_PATH, DataChangeScope.SUBTREE);
+        listener.startCapture();
+
+        writeTopWithListItem(CONFIGURATION);
+
+        AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();
+
+        assertContains(event.getCreatedData(), TOP_PATH);
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY));
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_ONE_KEY));
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));
+
+        assertEmpty(event.getUpdatedData());
+        assertEmpty(event.getRemovedPaths());
+    }
+
+    @Test
+    public void writeTopWithListItemAugmentedListenAugmentSubtreeWildcarded() {
+        TestListener listener = createListener(CONFIGURATION,AUGMENT_WILDCARD, DataChangeScope.SUBTREE);
+        listener.startCapture();
+        writeTopWithListItem(CONFIGURATION);
+
+        AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();
+
+        /*
+         * Event should not contain parent nodes
+         */
+        assertNotContains(event.getCreatedData(), TOP_PATH, path(TOP_FOO_KEY));
+
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_ONE_KEY));
+        assertContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));
+
+        assertEmpty(event.getUpdatedData());
+        assertEmpty(event.getRemovedPaths());
+    }
+
+    @Test
+    public void deleteAugmentChildListenTopSubtree() {
+        writeTopWithListItem(CONFIGURATION);
+        TestListener listener = createListener(CONFIGURATION, TOP_PATH, DataChangeScope.SUBTREE);
+        InstanceIdentifier<?> deletePath = path(TOP_FOO_KEY,USES_ONE_KEY);
+        deleteItem(CONFIGURATION,deletePath);
+
+        AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();
+
+
+        assertEmpty(event.getCreatedData());
+
+        assertContains(event.getRemovedPaths(), deletePath);
+
+        assertContains(event.getUpdatedData(), TOP_PATH);
+        assertContains(event.getUpdatedData(), path(TOP_FOO_KEY));
+        assertContains(event.getUpdatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
+
+        assertNotContains(event.getCreatedData(), path(TOP_FOO_KEY, USES_TWO_KEY));
+    }
+
+    @Test
+    public void deleteAugmentChildListenAugmentSubtreeWildcarded() {
+        writeTopWithListItem(CONFIGURATION);
+
+        TestListener listener = createListener(CONFIGURATION, AUGMENT_WILDCARD, DataChangeScope.SUBTREE);
+        InstanceIdentifier<?> deletePath = path(TOP_FOO_KEY,USES_ONE_KEY);
+        deleteItem(CONFIGURATION,deletePath);
+        AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event = listener.event();
+
+        assertEmpty(event.getCreatedData());
+
+        assertContains(event.getUpdatedData(), path(TOP_FOO_KEY, TreeComplexUsesAugment.class));
+
+        /*
+         * Event should not contain parent nodes
+         */
+        assertNotContains(event.getUpdatedData(), TOP_PATH, path(TOP_FOO_KEY));
+
+        assertContains(event.getRemovedPaths(), deletePath);
+    }
+
+}
index 79aa6b6..e0f6f35 100644 (file)
@@ -47,13 +47,13 @@ public class DataBrokerTestCustomizer {
 
     public DOMStore createConfigurationDatastore() {
         InMemoryDOMDataStore store = new InMemoryDOMDataStore("CFG", MoreExecutors.sameThreadExecutor());
-        schemaService.registerSchemaServiceListener(store);
+        schemaService.registerSchemaContextListener(store);
         return store;
     }
 
     public DOMStore createOperationalDatastore() {
         InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
-        schemaService.registerSchemaServiceListener(store);
+        schemaService.registerSchemaContextListener(store);
         return store;
     }
 
index e82c9d3..deb4a8a 100644 (file)
@@ -144,8 +144,8 @@ public class BindingTestContext implements AutoCloseable {
 
         biCompatibleBroker = new BackwardsCompatibleDataBroker(newDOMDataBroker,mockSchemaService);
 
-        mockSchemaService.registerSchemaServiceListener(configStore);
-        mockSchemaService.registerSchemaServiceListener(operStore);
+        mockSchemaService.registerSchemaContextListener(configStore);
+        mockSchemaService.registerSchemaContextListener(operStore);
         biDataLegacyBroker = biCompatibleBroker;
     }
 
@@ -246,7 +246,7 @@ public class BindingTestContext implements AutoCloseable {
     public void startBindingToDomMappingService() {
         checkState(classPool != null, "ClassPool needs to be present");
         mappingServiceImpl = new RuntimeGeneratedMappingServiceImpl(classPool);
-        mockSchemaService.registerSchemaServiceListener(mappingServiceImpl);
+        mockSchemaService.registerSchemaContextListener(mappingServiceImpl);
     }
 
     private void updateYangSchema(final ImmutableSet<YangModuleInfo> moduleInfos) {
index c8acbcd..63a4ffb 100644 (file)
@@ -13,14 +13,14 @@ import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 
 @SuppressWarnings("deprecation")
 public final class MockSchemaService implements SchemaService, SchemaContextProvider {
 
     private SchemaContext schemaContext;
 
-    ListenerRegistry<SchemaServiceListener> listeners = ListenerRegistry.create();
+    ListenerRegistry<SchemaContextListener> listeners = ListenerRegistry.create();
 
     @Override
     public void addModule(final Module module) {
@@ -38,8 +38,8 @@ public final class MockSchemaService implements SchemaService, SchemaContextProv
     }
 
     @Override
-    public ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(
-            final SchemaServiceListener listener) {
+    public ListenerRegistration<SchemaContextListener> registerSchemaContextListener(
+            final SchemaContextListener listener) {
         return listeners.register(listener);
     }
 
@@ -55,8 +55,8 @@ public final class MockSchemaService implements SchemaService, SchemaContextProv
 
     public synchronized void changeSchema(final SchemaContext newContext) {
         schemaContext = newContext;
-        for (ListenerRegistration<SchemaServiceListener> listener : listeners) {
+        for (ListenerRegistration<SchemaContextListener> listener : listeners) {
             listener.getInstance().onGlobalContextUpdated(schemaContext);
         }
     }
-}
\ No newline at end of file
+}
index 692d1b4..6d87271 100644 (file)
@@ -20,7 +20,7 @@ public class DistributedDataStoreFactory {
             new DistributedDataStore(actorSystem, name, new ClusterWrapperImpl(actorSystem),config );
        ShardStrategyFactory.setConfiguration(config);
         schemaService
-            .registerSchemaServiceListener(dataStore);
+            .registerSchemaContextListener(dataStore);
         return dataStore;
 
     }
index 57df201..6860872 100644 (file)
@@ -8,6 +8,7 @@
 
 package org.opendaylight.controller.md.cluster.datastore.model;
 
+import java.math.BigInteger;
 import org.opendaylight.yangtools.yang.common.QName;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
@@ -42,14 +43,14 @@ public class CarsModel {
         MapEntryNode altima =
             ImmutableNodes.mapEntryBuilder(CAR_QNAME, CAR_NAME_QNAME, "altima")
                 .withChild(ImmutableNodes.leafNode(CAR_NAME_QNAME, "altima"))
-                .withChild(ImmutableNodes.leafNode(CAR_PRICE_QNAME, 1000))
+                .withChild(ImmutableNodes.leafNode(CAR_PRICE_QNAME, new BigInteger("1000")))
                 .build();
 
         // Create an entry for the car accord
         MapEntryNode honda =
             ImmutableNodes.mapEntryBuilder(CAR_QNAME, CAR_NAME_QNAME, "accord")
                 .withChild(ImmutableNodes.leafNode(CAR_NAME_QNAME, "accord"))
-                .withChild(ImmutableNodes.leafNode(CAR_PRICE_QNAME, 2000))
+                .withChild(ImmutableNodes.leafNode(CAR_PRICE_QNAME, new BigInteger("2000")))
                 .build();
 
         cars.withChild(altima);
index 1b4020a..e637920 100644 (file)
@@ -42,14 +42,14 @@ public class PeopleModel {
         MapEntryNode jack =
             ImmutableNodes.mapEntryBuilder(PERSON_QNAME, PERSON_NAME_QNAME, "jack")
                 .withChild(ImmutableNodes.leafNode(PERSON_NAME_QNAME, "jack"))
-                .withChild(ImmutableNodes.leafNode(PERSON_AGE_QNAME, 100))
+                .withChild(ImmutableNodes.leafNode(PERSON_AGE_QNAME, 100L))
                 .build();
 
         // Create an entry for the person jill
         MapEntryNode jill =
             ImmutableNodes.mapEntryBuilder(PERSON_QNAME, PERSON_NAME_QNAME, "jill")
                 .withChild(ImmutableNodes.leafNode(PERSON_NAME_QNAME, "jill"))
-                .withChild(ImmutableNodes.leafNode(PERSON_AGE_QNAME, 200))
+                .withChild(ImmutableNodes.leafNode(PERSON_AGE_QNAME, 200L))
                 .build();
 
         cars.withChild(jack);
index d8fefcd..be8713c 100644 (file)
@@ -18,45 +18,45 @@ import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 public class SampleModelsTest {
     @Test
     public void testPeopleModel(){
-        NormalizedNode<?, ?> expected = PeopleModel.create();
+        final NormalizedNode<?, ?> expected = PeopleModel.create();
 
 
-        NormalizedNodeMessages.Container node =
+        final NormalizedNodeMessages.Container node =
             new NormalizedNodeToNodeCodec(SchemaContextHelper.full())
                 .encode(YangInstanceIdentifier.of(PeopleModel.BASE_QNAME),
                     expected);
 
-        NormalizedNodeMessages.Node normalizedNode =
+        final NormalizedNodeMessages.Node normalizedNode =
             node.getNormalizedNode();
 
-        NormalizedNode<?,?> actual = new NormalizedNodeToNodeCodec(SchemaContextHelper.full()).decode(YangInstanceIdentifier.of(PeopleModel.BASE_QNAME),
+        final NormalizedNode<?,?> actual = new NormalizedNodeToNodeCodec(SchemaContextHelper.full()).decode(YangInstanceIdentifier.of(PeopleModel.BASE_QNAME),
             normalizedNode);
 
 
-        Assert.assertEquals(expected.toString(), actual.toString());
+        Assert.assertEquals(expected, actual);
 
     }
 
 
     @Test
     public void testCarsModel(){
-        NormalizedNode<?, ?> expected = CarsModel.create();
+        final NormalizedNode<?, ?> expected = CarsModel.create();
 
 
-        NormalizedNodeMessages.Container node =
+        final NormalizedNodeMessages.Container node =
             new NormalizedNodeToNodeCodec(SchemaContextHelper.full())
                 .encode(YangInstanceIdentifier.of(CarsModel.BASE_QNAME),
                     expected);
 
-        NormalizedNodeMessages.Node normalizedNode =
+        final NormalizedNodeMessages.Node normalizedNode =
             node.getNormalizedNode();
 
-        NormalizedNode<?,?> actual = new NormalizedNodeToNodeCodec(SchemaContextHelper.full()).decode(
+        final NormalizedNode<?,?> actual = new NormalizedNodeToNodeCodec(SchemaContextHelper.full()).decode(
             YangInstanceIdentifier.of(CarsModel.BASE_QNAME),
             normalizedNode);
 
 
-        Assert.assertEquals(expected.toString(), actual.toString());
+        Assert.assertEquals(expected, actual);
 
     }
 }
index 34e5b1b..c3e979c 100644 (file)
@@ -11,7 +11,7 @@ import org.opendaylight.controller.sal.core.api.BrokerService;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 
 public interface SchemaService extends BrokerService {
 
@@ -42,5 +42,11 @@ public interface SchemaService extends BrokerService {
      */
     SchemaContext getGlobalContext();
 
-    ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(SchemaServiceListener listener);
+    /**
+     * Register a listener for changes in schema context.
+     *
+     * @param listener Listener which should be registered
+     * @return Listener registration handle
+     */
+    ListenerRegistration<SchemaContextListener> registerSchemaContextListener(SchemaContextListener listener);
 }
index 17b78f4..f1f16cd 100644 (file)
@@ -91,7 +91,7 @@ public final class DomBrokerImplModule extends org.opendaylight.controller.confi
         wrappedStore.changeDelegate(legacyStore);
         wrappedStore.setValidationEnabled(false);
 
-        schemaService.registerSchemaServiceListener(wrappedStore);
+        schemaService.registerSchemaContextListener(wrappedStore);
 
         dataService.registerConfigurationReader(rootPath, wrappedStore);
         dataService.registerCommitHandler(rootPath, wrappedStore);
index 69b17ee..667c0fc 100644 (file)
@@ -50,7 +50,7 @@ public final class DomInmemoryDataBrokerModule extends
            //we will default to InMemoryDOMDataStore creation
           operStore = new InMemoryDOMDataStore("DOM-OPER", storeExecutor);
           //here we will register the SchemaContext listener
-          getSchemaServiceDependency().registerSchemaServiceListener((InMemoryDOMDataStore)operStore);
+          getSchemaServiceDependency().registerSchemaContextListener((InMemoryDOMDataStore)operStore);
         }
 
         DOMStore configStore = getConfigDataStoreDependency();
@@ -58,7 +58,7 @@ public final class DomInmemoryDataBrokerModule extends
            //we will default to InMemoryDOMDataStore creation
            configStore = new InMemoryDOMDataStore("DOM-CFG", storeExecutor);
           //here we will register the SchemaContext listener
-          getSchemaServiceDependency().registerSchemaServiceListener((InMemoryDOMDataStore)configStore);
+          getSchemaServiceDependency().registerSchemaContextListener((InMemoryDOMDataStore)configStore);
         }
         ImmutableMap<LogicalDatastoreType, DOMStore> datastores = ImmutableMap
                 .<LogicalDatastoreType, DOMStore> builder().put(LogicalDatastoreType.OPERATIONAL, operStore)
index fbc418d..62b0264 100644 (file)
@@ -13,7 +13,7 @@ import org.opendaylight.yangtools.concepts.Delegator;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.osgi.framework.BundleContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -90,8 +90,8 @@ org.opendaylight.controller.config.yang.md.sal.dom.impl.AbstractSchemaServiceImp
         }
 
         @Override
-        public ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(final SchemaServiceListener arg0) {
-            return delegate.registerSchemaServiceListener(arg0);
+        public ListenerRegistration<SchemaContextListener> registerSchemaContextListener(final SchemaContextListener arg0) {
+            return delegate.registerSchemaContextListener(arg0);
         }
 
         @Override
index e8f8da5..dc122cf 100644 (file)
@@ -1,3 +1,10 @@
+/*
+ * 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.md.sal.dom.broker.impl.compat;
 
 import javax.annotation.concurrent.ThreadSafe;
@@ -21,18 +28,17 @@ import org.opendaylight.yangtools.yang.data.api.CompositeNode;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
 
 @ThreadSafe
 public class BackwardsCompatibleDataBroker implements DataProviderService {
 
     private final DOMDataBroker backingBroker;
     private volatile DataNormalizer normalizer;
-    private final ListenerRegistration<SchemaServiceListener> schemaReg;
+    private final ListenerRegistration<SchemaContextListener> schemaReg;
 
     public BackwardsCompatibleDataBroker(final DOMDataBroker newBiDataImpl, final SchemaService schemaService) {
         backingBroker = newBiDataImpl;
-        schemaReg = schemaService.registerSchemaServiceListener(new SchemaListener());
+        schemaReg = schemaService.registerSchemaContextListener(new SchemaListener());
     }
 
     @Override
index d837d75..61ea47e 100644 (file)
@@ -72,7 +72,7 @@ import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgum
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 
 public class BackwardsCompatibleMountPoint implements MountProvisionInstance, SchemaContextProvider, SchemaService {
 
@@ -83,7 +83,7 @@ public class BackwardsCompatibleMountPoint implements MountProvisionInstance, Sc
     private final NotificationPublishService notificationPublishService;
     private final RpcProvisionRegistry rpcs;
 
-    private final ListenerRegistry<SchemaServiceListener> schemaListenerRegistry = new ListenerRegistry<>();
+    private final ListenerRegistry<SchemaContextListener> schemaListenerRegistry = new ListenerRegistry<>();
 
     private SchemaContext schemaContext;
 
@@ -154,7 +154,7 @@ public class BackwardsCompatibleMountPoint implements MountProvisionInstance, Sc
     }
 
     @Override
-    public ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(final SchemaServiceListener listener) {
+    public ListenerRegistration<SchemaContextListener> registerSchemaContextListener(final SchemaContextListener listener) {
         return schemaListenerRegistry.register(listener);
     }
 
@@ -275,7 +275,7 @@ public class BackwardsCompatibleMountPoint implements MountProvisionInstance, Sc
     @Override
     public void setSchemaContext(final SchemaContext schemaContext) {
         this.schemaContext = schemaContext;
-        for (ListenerRegistration<SchemaServiceListener> schemaServiceListenerListenerRegistration : schemaListenerRegistry.getListeners()) {
+        for (ListenerRegistration<SchemaContextListener> schemaServiceListenerListenerRegistration : schemaListenerRegistry.getListeners()) {
             schemaServiceListenerListenerRegistration.getInstance().onGlobalContextUpdated(schemaContext);
         }
     }
index d8174c3..8263732 100644 (file)
@@ -23,7 +23,7 @@ import org.opendaylight.yangtools.concepts.Registration;
 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.opendaylight.yangtools.yang.parser.impl.util.URLSchemaContextResolver;
 import org.osgi.framework.Bundle;
 import org.osgi.framework.BundleContext;
@@ -40,15 +40,15 @@ import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
-public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvider, SchemaService, ServiceTrackerCustomizer<SchemaServiceListener, SchemaServiceListener>, AutoCloseable {
+public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvider, SchemaService, ServiceTrackerCustomizer<SchemaContextListener, SchemaContextListener>, AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(GlobalBundleScanningSchemaServiceImpl.class);
 
-    private final ListenerRegistry<SchemaServiceListener> listeners = new ListenerRegistry<>();
+    private final ListenerRegistry<SchemaContextListener> listeners = new ListenerRegistry<>();
     private final URLSchemaContextResolver contextResolver = new URLSchemaContextResolver();
     private final BundleScanner scanner = new BundleScanner();
     private final BundleContext context;
 
-    private ServiceTracker<SchemaServiceListener, SchemaServiceListener> listenerTracker;
+    private ServiceTracker<SchemaContextListener, SchemaContextListener> listenerTracker;
     private BundleTracker<Iterable<Registration>> bundleTracker;
     private boolean starting = true;
     private static GlobalBundleScanningSchemaServiceImpl instance;
@@ -81,7 +81,7 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
     public void start() {
         checkState(context != null);
 
-        listenerTracker = new ServiceTracker<>(context, SchemaServiceListener.class, GlobalBundleScanningSchemaServiceImpl.this);
+        listenerTracker = new ServiceTracker<>(context, SchemaContextListener.class, GlobalBundleScanningSchemaServiceImpl.this);
         bundleTracker = new BundleTracker<>(context, BundleEvent.RESOLVED | BundleEvent.UNRESOLVED, scanner);
         bundleTracker.open();
         listenerTracker.open();
@@ -115,7 +115,7 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
     }
 
     @Override
-    public synchronized ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(final SchemaServiceListener listener) {
+    public synchronized ListenerRegistration<SchemaContextListener> registerSchemaContextListener(final SchemaContextListener listener) {
         Optional<SchemaContext> potentialCtx = contextResolver.getSchemaContext();
         if(potentialCtx.isPresent()) {
             listener.onGlobalContextUpdated(potentialCtx.get());
@@ -137,7 +137,7 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
 
     private synchronized void updateContext(final SchemaContext snapshot) {
         Object[] services = listenerTracker.getServices();
-        for (ListenerRegistration<SchemaServiceListener> listener : listeners) {
+        for (ListenerRegistration<SchemaContextListener> listener : listeners) {
             try {
                 listener.getInstance().onGlobalContextUpdated(snapshot);
             } catch (Exception e) {
@@ -146,7 +146,7 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
         }
         if (services != null) {
             for (Object rawListener : services) {
-                SchemaServiceListener listener = (SchemaServiceListener) rawListener;
+                final SchemaContextListener listener = (SchemaContextListener) rawListener;
                 try {
                     listener.onGlobalContextUpdated(snapshot);
                 } catch (Exception e) {
@@ -213,9 +213,9 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
     }
 
     @Override
-    public synchronized SchemaServiceListener addingService(final ServiceReference<SchemaServiceListener> reference) {
+    public synchronized SchemaContextListener addingService(final ServiceReference<SchemaContextListener> reference) {
 
-        SchemaServiceListener listener = context.getService(reference);
+        SchemaContextListener listener = context.getService(reference);
         SchemaContext _ctxContext = getGlobalContext();
         if (getContext() != null && _ctxContext != null) {
             listener.onGlobalContextUpdated(_ctxContext);
@@ -234,12 +234,12 @@ public class GlobalBundleScanningSchemaServiceImpl implements SchemaContextProvi
     }
 
     @Override
-    public void modifiedService(final ServiceReference<SchemaServiceListener> reference, final SchemaServiceListener service) {
+    public void modifiedService(final ServiceReference<SchemaContextListener> reference, final SchemaContextListener service) {
         // NOOP
     }
 
     @Override
-    public void removedService(final ServiceReference<SchemaServiceListener> reference, final SchemaServiceListener service) {
+    public void removedService(final ServiceReference<SchemaContextListener> reference, final SchemaContextListener service) {
         context.ungetService(reference);
     }
 }
index 1d864ee..d8d2346 100644 (file)
@@ -8,10 +8,10 @@
 package org.opendaylight.controller.sal.dom.broker.osgi;
 
 import org.opendaylight.controller.sal.core.api.model.SchemaService;
-import org.opendaylight.yangtools.yang.model.api.SchemaServiceListener;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.model.api.Module;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.osgi.framework.ServiceReference;
 
 public class SchemaServiceProxy extends AbstractBrokerServiceProxy<SchemaService> implements SchemaService {
@@ -41,12 +41,9 @@ public class SchemaServiceProxy extends AbstractBrokerServiceProxy<SchemaService
     }
 
     @Override
-    public ListenerRegistration<SchemaServiceListener> registerSchemaServiceListener(SchemaServiceListener listener) {
-        ListenerRegistration<SchemaServiceListener> registration = getDelegate().registerSchemaServiceListener(listener);
+    public ListenerRegistration<SchemaContextListener> registerSchemaContextListener(SchemaContextListener listener) {
+        ListenerRegistration<SchemaContextListener> registration = getDelegate().registerSchemaContextListener(listener);
         addRegistration(registration);
         return registration;
     }
-
-
-
 }
index 399f49b..59cdc76 100644 (file)
@@ -20,7 +20,7 @@ public class XSQLModule extends org.opendaylight.yang.gen.v1.http.netconfcentral
     @Override
     public java.lang.AutoCloseable createInstance() {
         XSQLAdapter xsqlAdapter = XSQLAdapter.getInstance();
-        getSchemaServiceDependency().registerSchemaServiceListener(xsqlAdapter);
+        getSchemaServiceDependency().registerSchemaContextListener(xsqlAdapter);
         xsqlAdapter.setDataBroker(getAsyncDataBrokerDependency());
         XSQLProvider p = new XSQLProvider();
         p.buildXSQL(getDataBrokerDependency());
index 01a5989..805608d 100644 (file)
@@ -23,7 +23,7 @@ public class InMemoryConfigDataStoreProviderModule extends org.opendaylight.cont
     @Override
     public java.lang.AutoCloseable createInstance() {
       InMemoryDOMDataStore   ids = new InMemoryDOMDataStore("DOM-CFG", MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()));
-      getSchemaServiceDependency().registerSchemaServiceListener(ids);
+      getSchemaServiceDependency().registerSchemaContextListener(ids);
       return ids;
     }
 
index b39c9bb..f479558 100644 (file)
@@ -23,7 +23,7 @@ public class InMemoryOperationalDataStoreProviderModule extends org.opendaylight
     @Override
     public java.lang.AutoCloseable createInstance() {
       InMemoryDOMDataStore ids = new InMemoryDOMDataStore("DOM-OPER", MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()));
-      getOperationalSchemaServiceDependency().registerSchemaServiceListener(ids);
+      getOperationalSchemaServiceDependency().registerSchemaContextListener(ids);
       return ids;
     }
 
index 037bfb4..b75df80 100644 (file)
@@ -13,8 +13,10 @@ import static org.opendaylight.controller.config.api.JmxAttributeValidationExcep
 import java.io.File;
 import java.io.InputStream;
 import java.net.InetSocketAddress;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 
+import org.opendaylight.controller.config.api.JmxAttributeValidationException;
 import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
 import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
 import org.opendaylight.controller.netconf.client.conf.NetconfReconnectingClientConfiguration;
@@ -24,6 +26,7 @@ import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
 import org.opendaylight.controller.sal.connect.api.RemoteDeviceHandler;
 import org.opendaylight.controller.sal.connect.netconf.NetconfDevice;
 import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionCapabilities;
 import org.opendaylight.controller.sal.connect.netconf.sal.NetconfDeviceSalFacade;
 import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
 import org.opendaylight.controller.sal.core.api.Broker;
@@ -40,6 +43,8 @@ import org.osgi.framework.BundleContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Optional;
+
 /**
  *
  */
@@ -49,6 +54,7 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
 
     private static AbstractCachingSchemaSourceProvider<String, InputStream> GLOBAL_NETCONF_SOURCE_PROVIDER = null;
     private BundleContext bundleContext;
+    private Optional<NetconfSessionCapabilities> userCapabilities;
 
     public NetconfConnectorModule(final org.opendaylight.controller.config.api.ModuleIdentifier identifier, final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
         super(identifier, dependencyResolver);
@@ -82,9 +88,11 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
             checkNotNull(getPassword(), passwordJmxAttribute);
         }
 
+        userCapabilities = getUserCapabilities();
+
     }
 
-    private boolean isHostAddressPresent(Host address) {
+    private boolean isHostAddressPresent(final Host address) {
         return address.getDomainName() != null ||
                address.getIpAddress() != null && (address.getIpAddress().getIpv4Address() != null || address.getIpAddress().getIpv6Address() != null);
     }
@@ -98,10 +106,14 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         final Broker domBroker = getDomRegistryDependency();
         final BindingAwareBroker bindingBroker = getBindingRegistryDependency();
 
-        final RemoteDeviceHandler salFacade = new NetconfDeviceSalFacade(id, domBroker, bindingBroker, bundleContext, globalProcessingExecutor);
+        final RemoteDeviceHandler<NetconfSessionCapabilities> salFacade
+                = new NetconfDeviceSalFacade(id, domBroker, bindingBroker, bundleContext, globalProcessingExecutor);
         final NetconfDevice device =
                 NetconfDevice.createNetconfDevice(id, getGlobalNetconfSchemaProvider(), globalProcessingExecutor, salFacade);
-        final NetconfDeviceCommunicator listener = new NetconfDeviceCommunicator(id, device);
+
+        final NetconfDeviceCommunicator listener = userCapabilities.isPresent() ?
+                new NetconfDeviceCommunicator(id, device, userCapabilities.get()) : new NetconfDeviceCommunicator(id, device);
+
         final NetconfReconnectingClientConfiguration clientConfig = getClientConfig(listener);
 
         final NetconfClientDispatcher dispatcher = getClientDispatcherDependency();
@@ -116,6 +128,26 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         };
     }
 
+    private Optional<NetconfSessionCapabilities> getUserCapabilities() {
+        if(getYangModuleCapabilities() == null) {
+            return Optional.absent();
+        }
+
+        final List<String> capabilities = getYangModuleCapabilities().getCapability();
+        if(capabilities == null || capabilities.isEmpty()) {
+            return Optional.absent();
+        }
+
+        final NetconfSessionCapabilities parsedOverrideCapabilities = NetconfSessionCapabilities.fromStrings(capabilities);
+        JmxAttributeValidationException.checkCondition(
+                parsedOverrideCapabilities.getNonModuleCaps().isEmpty(),
+                "Capabilities to override can only contain module based capabilities, non-module capabilities will be retrieved from the device," +
+                        " configured non-module capabilities: " + parsedOverrideCapabilities.getNonModuleCaps(),
+                yangModuleCapabilitiesJmxAttribute);
+
+        return Optional.of(parsedOverrideCapabilities);
+    }
+
     private synchronized AbstractCachingSchemaSourceProvider<String, InputStream> getGlobalNetconfSchemaProvider() {
         if(GLOBAL_NETCONF_SOURCE_PROVIDER == null) {
             final String storageFile = "cache/schema";
@@ -175,8 +207,8 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         if(getAddress().getDomainName() != null) {
             return new InetSocketAddress(getAddress().getDomainName().getValue(), getPort().getValue());
         } else {
-            IpAddress ipAddress = getAddress().getIpAddress();
-            String ip = ipAddress.getIpv4Address() != null ? ipAddress.getIpv4Address().getValue() : ipAddress.getIpv6Address().getValue();
+            final IpAddress ipAddress = getAddress().getIpAddress();
+            final String ip = ipAddress.getIpv4Address() != null ? ipAddress.getIpv4Address().getValue() : ipAddress.getIpv6Address().getValue();
             return new InetSocketAddress(ip, getPort().getValue());
         }
     }
index de4ac7a..07d3c08 100644 (file)
@@ -118,6 +118,7 @@ public final class NetconfDevice implements RemoteDevice<NetconfSessionCapabilit
                 // Unable to initialize device, set as disconnected
                 logger.error("{}: Initialization failed", id, t);
                 salFacade.onDeviceDisconnected();
+                // TODO ssh connection is still open if sal initialization fails
             }
         });
     }
index 8045f8c..3871cdf 100644 (file)
@@ -7,13 +7,19 @@
  */
 package org.opendaylight.controller.sal.connect.netconf.listener;
 
+import com.google.common.base.Optional;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
 import java.util.ArrayDeque;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-
 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
 import org.opendaylight.controller.netconf.api.NetconfMessage;
 import org.opendaylight.controller.netconf.api.NetconfTerminationReason;
@@ -36,31 +42,35 @@ import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.FutureListener;
-
 public class NetconfDeviceCommunicator implements NetconfClientSessionListener, RemoteDeviceCommunicator<NetconfMessage> {
 
     private static final Logger logger = LoggerFactory.getLogger(NetconfDeviceCommunicator.class);
 
     private final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice;
+    private final Optional<NetconfSessionCapabilities> overrideNetconfCapabilities;
     private final RemoteDeviceId id;
     private final Lock sessionLock = new ReentrantLock();
 
+    private final Queue<Request> requests = new ArrayDeque<>();
+    private NetconfClientSession session;
+
+    public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice,
+            final NetconfSessionCapabilities netconfSessionCapabilities) {
+        this(id, remoteDevice, Optional.of(netconfSessionCapabilities));
+    }
+
     public NetconfDeviceCommunicator(final RemoteDeviceId id,
-            final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice) {
+                                     final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice) {
+        this(id, remoteDevice, Optional.<NetconfSessionCapabilities>absent());
+    }
+
+    private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice,
+            final Optional<NetconfSessionCapabilities> overrideNetconfCapabilities) {
         this.id = id;
         this.remoteDevice = remoteDevice;
+        this.overrideNetconfCapabilities = overrideNetconfCapabilities;
     }
 
-    private final Queue<Request> requests = new ArrayDeque<>();
-    private NetconfClientSession session;
-
     @Override
     public void onSessionUp(final NetconfClientSession session) {
         sessionLock.lock();
@@ -68,10 +78,15 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
             logger.debug("{}: Session established", id);
             this.session = session;
 
-            final NetconfSessionCapabilities netconfSessionCapabilities =
+            NetconfSessionCapabilities netconfSessionCapabilities =
                                              NetconfSessionCapabilities.fromNetconfSession(session);
             logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionCapabilities);
 
+            if(overrideNetconfCapabilities.isPresent()) {
+                netconfSessionCapabilities = netconfSessionCapabilities.replaceModuleCaps(overrideNetconfCapabilities.get());
+                logger.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id, netconfSessionCapabilities);
+            }
+
             remoteDevice.onRemoteSessionUp(netconfSessionCapabilities, this);
         }
         finally {
@@ -223,7 +238,7 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
                 return;
             }
 
-            request.future.set( RpcResultBuilder.<NetconfMessage>success( message ).build() );
+            request.future.set( RpcResultBuilder.success( message ).build() );
         }
     }
 
index 8964a80..1a7d90e 100644 (file)
@@ -8,6 +8,7 @@ import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 
+import com.google.common.collect.Sets;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
@@ -19,6 +20,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public final class NetconfSessionCapabilities {
+
     private static final class ParameterMatcher {
         private final Predicate<String> predicate;
         private final int skipLength;
@@ -57,10 +59,10 @@ public final class NetconfSessionCapabilities {
     };
 
     private final Set<QName> moduleBasedCaps;
-    private final Set<String> capabilities;
+    private final Set<String> nonModuleCaps;
 
-    private NetconfSessionCapabilities(final Set<String> capabilities, final Set<QName> moduleBasedCaps) {
-        this.capabilities = Preconditions.checkNotNull(capabilities);
+    private NetconfSessionCapabilities(final Set<String> nonModuleCaps, final Set<QName> moduleBasedCaps) {
+        this.nonModuleCaps = Preconditions.checkNotNull(nonModuleCaps);
         this.moduleBasedCaps = Preconditions.checkNotNull(moduleBasedCaps);
     }
 
@@ -68,30 +70,49 @@ public final class NetconfSessionCapabilities {
         return moduleBasedCaps;
     }
 
-    public boolean containsCapability(final String capability) {
-        return capabilities.contains(capability);
+    public Set<String> getNonModuleCaps() {
+        return nonModuleCaps;
+    }
+
+    public boolean containsNonModuleCapability(final String capability) {
+        return nonModuleCaps.contains(capability);
     }
 
-    public boolean containsCapability(final QName capability) {
+    public boolean containsModuleCapability(final QName capability) {
         return moduleBasedCaps.contains(capability);
     }
 
     @Override
     public String toString() {
         return Objects.toStringHelper(this)
-                .add("capabilities", capabilities)
+                .add("capabilities", nonModuleCaps)
+                .add("moduleBasedCapabilities", moduleBasedCaps)
                 .add("rollback", isRollbackSupported())
                 .add("monitoring", isMonitoringSupported())
                 .toString();
     }
 
     public boolean isRollbackSupported() {
-        return containsCapability(NetconfMessageTransformUtil.NETCONF_ROLLBACK_ON_ERROR_URI.toString());
+        return containsNonModuleCapability(NetconfMessageTransformUtil.NETCONF_ROLLBACK_ON_ERROR_URI.toString());
+    }
+
+    public boolean isCandidateSupported() {
+        return containsNonModuleCapability(NetconfMessageTransformUtil.NETCONF_CANDIDATE_URI.toString());
     }
 
     public boolean isMonitoringSupported() {
-        return containsCapability(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING)
-                || containsCapability(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING.getNamespace().toString());
+        return containsModuleCapability(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING)
+                || containsNonModuleCapability(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING.getNamespace().toString());
+    }
+
+    public NetconfSessionCapabilities replaceModuleCaps(final NetconfSessionCapabilities netconfSessionModuleCapabilities) {
+        final Set<QName> moduleBasedCaps = Sets.newHashSet(netconfSessionModuleCapabilities.getModuleBasedCaps());
+
+        // Preserve monitoring module, since it indicates support for ietf-netconf-monitoring
+        if(containsModuleCapability(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING)) {
+            moduleBasedCaps.add(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING);
+        }
+        return new NetconfSessionCapabilities(getNonModuleCaps(), moduleBasedCaps);
     }
 
     public static NetconfSessionCapabilities fromNetconfSession(final NetconfClientSession session) {
@@ -100,6 +121,7 @@ public final class NetconfSessionCapabilities {
 
     public static NetconfSessionCapabilities fromStrings(final Collection<String> capabilities) {
         final Set<QName> moduleBasedCaps = new HashSet<>();
+        final Set<String> nonModuleCaps = Sets.newHashSet(capabilities);
 
         for (final String capability : capabilities) {
             final int qmark = capability.indexOf('?');
@@ -117,6 +139,7 @@ public final class NetconfSessionCapabilities {
             String revision = REVISION_PARAM.from(queryParams);
             if (revision != null) {
                 moduleBasedCaps.add(QName.create(namespace, revision, moduleName));
+                nonModuleCaps.remove(capability);
                 continue;
             }
 
@@ -136,8 +159,9 @@ public final class NetconfSessionCapabilities {
 
             // FIXME: do we really want to continue here?
             moduleBasedCaps.add(QName.create(namespace, revision, moduleName));
+            nonModuleCaps.remove(capability);
         }
 
-        return new NetconfSessionCapabilities(ImmutableSet.copyOf(capabilities), ImmutableSet.copyOf(moduleBasedCaps));
+        return new NetconfSessionCapabilities(ImmutableSet.copyOf(nonModuleCaps), ImmutableSet.copyOf(moduleBasedCaps));
     }
 }
index ee0c8b7..f3a9acd 100644 (file)
@@ -42,7 +42,7 @@ final class NetconfDeviceDataBroker implements DOMDataBroker {
 
     @Override
     public DOMDataReadOnlyTransaction newReadOnlyTransaction() {
-        return new NetconfDeviceReadOnlyTx(rpc, normalizer);
+        return new NetconfDeviceReadOnlyTx(rpc, normalizer, id);
     }
 
     @Override
@@ -52,8 +52,7 @@ final class NetconfDeviceDataBroker implements DOMDataBroker {
 
     @Override
     public DOMDataWriteTransaction newWriteOnlyTransaction() {
-        // FIXME detect if candidate is supported, true is hardcoded
-        return new NetconfDeviceWriteOnlyTx(id, rpc, normalizer, true, netconfSessionPreferences.isRollbackSupported());
+        return new NetconfDeviceWriteOnlyTx(id, rpc, normalizer, netconfSessionPreferences.isCandidateSupported(), netconfSessionPreferences.isRollbackSupported());
     }
 
     @Override
index 3248453..9ef44f6 100644 (file)
@@ -15,6 +15,7 @@ import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessag
 
 import com.google.common.base.Function;
 import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
@@ -22,6 +23,7 @@ import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizat
 import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizer;
 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
 import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
 import org.opendaylight.controller.sal.core.api.RpcImplementation;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
@@ -32,16 +34,19 @@ import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction {
 
     private static final Logger LOG  = LoggerFactory.getLogger(NetconfDeviceReadOnlyTx.class);
 
     private final RpcImplementation rpc;
     private final DataNormalizer normalizer;
+    private final RemoteDeviceId id;
 
-    public NetconfDeviceReadOnlyTx(final RpcImplementation rpc, final DataNormalizer normalizer) {
+    public NetconfDeviceReadOnlyTx(final RpcImplementation rpc, final DataNormalizer normalizer, final RemoteDeviceId id) {
         this.rpc = rpc;
         this.normalizer = normalizer;
+        this.id = id;
     }
 
     public ListenableFuture<Optional<NormalizedNode<?, ?>>> readConfigurationData(final YangInstanceIdentifier path) {
@@ -51,6 +56,8 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
         return Futures.transform(future, new Function<RpcResult<CompositeNode>, Optional<NormalizedNode<?, ?>>>() {
             @Override
             public Optional<NormalizedNode<?, ?>> apply(final RpcResult<CompositeNode> result) {
+                checkReadSuccess(result, path);
+
                 final CompositeNode data = result.getResult().getFirstCompositeByName(NETCONF_DATA_QNAME);
                 final CompositeNode node = (CompositeNode) findNode(data, path);
 
@@ -61,6 +68,11 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
         });
     }
 
+    private void checkReadSuccess(final RpcResult<CompositeNode> result, final YangInstanceIdentifier path) {
+        LOG.warn("{}: Unable to read data: {}, errors: {}", id, path, result.getErrors());
+        Preconditions.checkArgument(result.isSuccessful(), "%s: Unable to read data: %s, errors: %s", id, path, result.getErrors());
+    }
+
     private Optional<NormalizedNode<?, ?>> transform(final YangInstanceIdentifier path, final CompositeNode node) {
         if(node == null) {
             return Optional.absent();
@@ -68,7 +80,7 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
         try {
             return Optional.<NormalizedNode<?, ?>>of(normalizer.toNormalized(path, node).getValue());
         } catch (final Exception e) {
-            LOG.error("Unable to normalize data for {}, data: {}", path, node, e);
+            LOG.error("{}: Unable to normalize data for {}, data: {}", id, path, node, e);
             throw e;
         }
     }
@@ -79,6 +91,8 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
         return Futures.transform(future, new Function<RpcResult<CompositeNode>, Optional<NormalizedNode<?, ?>>>() {
             @Override
             public Optional<NormalizedNode<?, ?>> apply(final RpcResult<CompositeNode> result) {
+                checkReadSuccess(result, path);
+
                 final CompositeNode data = result.getResult().getFirstCompositeByName(NETCONF_DATA_QNAME);
                 final CompositeNode node = (CompositeNode) findNode(data, path);
 
@@ -123,7 +137,7 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
 
     @Override
     public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
-        final YangInstanceIdentifier legacyPath = toLegacyPath(normalizer, path);
+        final YangInstanceIdentifier legacyPath = toLegacyPath(normalizer, path, id);
 
         switch (store) {
             case CONFIGURATION : {
@@ -134,14 +148,14 @@ public final class NetconfDeviceReadOnlyTx implements DOMDataReadOnlyTransaction
             }
         }
 
-        throw new IllegalArgumentException(String.format("Cannot read data %s for %s datastore, unknown datastore type", path, store));
+        throw new IllegalArgumentException(String.format("%s, Cannot read data %s for %s datastore, unknown datastore type", id, path, store));
     }
 
-    static YangInstanceIdentifier toLegacyPath(final DataNormalizer normalizer, final YangInstanceIdentifier path) {
+    static YangInstanceIdentifier toLegacyPath(final DataNormalizer normalizer, final YangInstanceIdentifier path, final RemoteDeviceId id) {
         try {
             return normalizer.toLegacy(path);
         } catch (final DataNormalizationException e) {
-            throw new IllegalArgumentException("Cannot normalize path " + path, e);
+            throw new IllegalArgumentException(id + ": Cannot normalize path " + path, e);
         }
     }
 
index c8d9028..87f5477 100644 (file)
@@ -8,10 +8,11 @@
 
 package org.opendaylight.controller.sal.connect.netconf.sal.tx;
 
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.DISCARD_CHANGES_RPC_CONTENT;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_CANDIDATE_QNAME;
-import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_COMMIT_QNAME;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_CONFIG_QNAME;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_DEFAULT_OPERATION_QNAME;
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_DISCARD_CHANGES_QNAME;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_EDIT_CONFIG_QNAME;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_ERROR_OPTION_QNAME;
 import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_OPERATION_QNAME;
@@ -26,13 +27,14 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
-import javax.annotation.Nullable;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
@@ -57,97 +59,114 @@ import org.opendaylight.yangtools.yang.data.impl.util.CompositeNodeBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class NetconfDeviceWriteOnlyTx implements DOMDataWriteTransaction {
+public class NetconfDeviceWriteOnlyTx implements DOMDataWriteTransaction, FutureCallback<RpcResult<TransactionStatus>> {
 
     private static final Logger LOG  = LoggerFactory.getLogger(NetconfDeviceWriteOnlyTx.class);
 
     private final RemoteDeviceId id;
     private final RpcImplementation rpc;
     private final DataNormalizer normalizer;
+
     private final boolean rollbackSupported;
+    private final boolean candidateSupported;
     private final CompositeNode targetNode;
 
+    // Allow commit to be called only once
+    private final AtomicBoolean finished = new AtomicBoolean(false);
+
     public NetconfDeviceWriteOnlyTx(final RemoteDeviceId id, final RpcImplementation rpc, final DataNormalizer normalizer, final boolean candidateSupported, final boolean rollbackOnErrorSupported) {
         this.id = id;
         this.rpc = rpc;
         this.normalizer = normalizer;
-        this.targetNode = getTargetNode(candidateSupported);
+
+        this.candidateSupported = candidateSupported;
+        this.targetNode = getTargetNode(this.candidateSupported);
         this.rollbackSupported = rollbackOnErrorSupported;
     }
 
-    // FIXME add logging
-
     @Override
     public boolean cancel() {
-        if(isCommitted()) {
+        if(isFinished()) {
             return false;
         }
 
         return discardChanges();
     }
 
-    private boolean isCommitted() {
-        // TODO 732
-        return true;
+    private boolean isFinished() {
+        return finished.get();
     }
 
     private boolean discardChanges() {
-        // TODO 732
+        finished.set(true);
+
+        if(candidateSupported) {
+            sendDiscardChanges();
+        }
         return true;
     }
 
     // TODO should the edit operations be blocking ?
+    // TODO should the discard-changes operations be blocking ?
 
     @Override
     public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkNotFinished();
         Preconditions.checkArgument(store == LogicalDatastoreType.CONFIGURATION, "Can merge only configuration, not %s", store);
 
         try {
-            final YangInstanceIdentifier legacyPath = NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path);
+            final YangInstanceIdentifier legacyPath = NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path, id);
             final CompositeNode legacyData = normalizer.toLegacy(path, data);
-            sendEditRpc(createEditConfigStructure(legacyPath, Optional.of(ModifyAction.REPLACE), Optional.fromNullable(legacyData)), Optional.of(ModifyAction.NONE));
+            sendEditRpc(
+                    createEditConfigStructure(legacyPath, Optional.of(ModifyAction.REPLACE), Optional.fromNullable(legacyData)), Optional.of(ModifyAction.NONE));
         } catch (final ExecutionException e) {
-            LOG.warn("Error putting data to {}, data: {}, discarding changes", path, data, e);
+            LOG.warn("{}: Error putting data to {}, data: {}, discarding changes", id, path, data, e);
             discardChanges();
-            throw new RuntimeException("Error while replacing " + path, e);
+            throw new RuntimeException(id + ": Error while replacing " + path, e);
         }
     }
 
+    private void checkNotFinished() {
+        Preconditions.checkState(isFinished() == false, "%s: Transaction %s already finished", id, getIdentifier());
+    }
+
     @Override
     public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        Preconditions.checkArgument(store == LogicalDatastoreType.CONFIGURATION, "Can merge only configuration, not %s", store);
+        checkNotFinished();
+        Preconditions.checkArgument(store == LogicalDatastoreType.CONFIGURATION, "%s: Can merge only configuration, not %s", id, store);
 
         try {
-            final YangInstanceIdentifier legacyPath = NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path);
+            final YangInstanceIdentifier legacyPath = NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path, id);
             final CompositeNode legacyData = normalizer.toLegacy(path, data);
             sendEditRpc(
                     createEditConfigStructure(legacyPath, Optional.<ModifyAction> absent(), Optional.fromNullable(legacyData)), Optional.<ModifyAction> absent());
         } catch (final ExecutionException e) {
-            LOG.warn("Error merging data to {}, data: {}, discarding changes", path, data, e);
+            LOG.warn("{}: Error merging data to {}, data: {}, discarding changes", id, path, data, e);
             discardChanges();
-            throw new RuntimeException("Error while merging " + path, e);
+            throw new RuntimeException(id + ": Error while merging " + path, e);
         }
     }
 
     @Override
     public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
-        Preconditions.checkArgument(store == LogicalDatastoreType.CONFIGURATION, "Can merge only configuration, not %s", store);
+        checkNotFinished();
+        Preconditions.checkArgument(store == LogicalDatastoreType.CONFIGURATION, "%s: Can merge only configuration, not %s", id, store);
 
         try {
-            sendEditRpc(createEditConfigStructure(NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path), Optional.of(ModifyAction.DELETE), Optional.<CompositeNode>absent()), Optional.of(ModifyAction.NONE));
+            sendEditRpc(
+                    createEditConfigStructure(NetconfDeviceReadOnlyTx.toLegacyPath(normalizer, path, id), Optional.of(ModifyAction.DELETE), Optional.<CompositeNode>absent()), Optional.of(ModifyAction.NONE));
         } catch (final ExecutionException e) {
-            LOG.warn("Error deleting data {}, discarding changes", path, e);
+            LOG.warn("{}: Error deleting data {}, discarding changes", id, path, e);
             discardChanges();
-            throw new RuntimeException("Error while deleting " + path, e);
+            throw new RuntimeException(id + ": Error while deleting " + path, e);
         }
     }
 
     @Override
     public CheckedFuture<Void, TransactionCommitFailedException> submit() {
         final ListenableFuture<Void> commmitFutureAsVoid = Futures.transform(commit(), new Function<RpcResult<TransactionStatus>, Void>() {
-            @Nullable
             @Override
-            public Void apply(@Nullable final RpcResult<TransactionStatus> input) {
+            public Void apply(final RpcResult<TransactionStatus> input) {
                 return null;
             }
         });
@@ -162,25 +181,46 @@ public class NetconfDeviceWriteOnlyTx implements DOMDataWriteTransaction {
 
     @Override
     public ListenableFuture<RpcResult<TransactionStatus>> commit() {
-        // FIXME do not allow commit if closed or failed
+        checkNotFinished();
+        finished.set(true);
 
-        final ListenableFuture<RpcResult<CompositeNode>> rpcResult = rpc.invokeRpc(NetconfMessageTransformUtil.NETCONF_COMMIT_QNAME, getCommitRequest());
-        return Futures.transform(rpcResult, new Function<RpcResult<CompositeNode>, RpcResult<TransactionStatus>>() {
-            @Override
-            public RpcResult<TransactionStatus> apply(@Nullable final RpcResult<CompositeNode> input) {
-                if(input.isSuccessful()) {
-                    return RpcResultBuilder.success(TransactionStatus.COMMITED).build();
-                } else {
-                    final RpcResultBuilder<TransactionStatus> failed = RpcResultBuilder.failed();
-                    for (final RpcError rpcError : input.getErrors()) {
-                        failed.withError(rpcError.getErrorType(), rpcError.getTag(), rpcError.getMessage(), rpcError.getApplicationTag(), rpcError.getInfo(), rpcError.getCause());
+        if(candidateSupported == false) {
+            return Futures.immediateFuture(RpcResultBuilder.success(TransactionStatus.COMMITED).build());
+        }
+
+        final ListenableFuture<RpcResult<CompositeNode>> rpcResult = rpc.invokeRpc(
+                NetconfMessageTransformUtil.NETCONF_COMMIT_QNAME, NetconfMessageTransformUtil.COMMIT_RPC_CONTENT);
+
+        final ListenableFuture<RpcResult<TransactionStatus>> transformed = Futures.transform(rpcResult,
+                new Function<RpcResult<CompositeNode>, RpcResult<TransactionStatus>>() {
+                    @Override
+                    public RpcResult<TransactionStatus> apply(final RpcResult<CompositeNode> input) {
+                        if (input.isSuccessful()) {
+                            return RpcResultBuilder.success(TransactionStatus.COMMITED).build();
+                        } else {
+                            final RpcResultBuilder<TransactionStatus> failed = RpcResultBuilder.failed();
+                            for (final RpcError rpcError : input.getErrors()) {
+                                failed.withError(rpcError.getErrorType(), rpcError.getTag(), rpcError.getMessage(),
+                                        rpcError.getApplicationTag(), rpcError.getInfo(), rpcError.getCause());
+                            }
+                            return failed.build();
+                        }
                     }
-                    return failed.build();
-                }
-            }
-        });
+                });
 
-        // FIXME 732 detect commit failure
+        Futures.addCallback(transformed, this);
+        return transformed;
+    }
+
+    @Override
+    public void onSuccess(final RpcResult<TransactionStatus> result) {
+        LOG.debug("{}: Write successful, transaction: {}", id, getIdentifier());
+    }
+
+    @Override
+    public void onFailure(final Throwable t) {
+        LOG.warn("{}: Write failed, transaction {}, discarding changes", id, getIdentifier(), t);
+        discardChanges();
     }
 
     private void sendEditRpc(final CompositeNode editStructure, final Optional<ModifyAction> defaultOperation) throws ExecutionException {
@@ -200,6 +240,22 @@ public class NetconfDeviceWriteOnlyTx implements DOMDataWriteTransaction {
         }
     }
 
+    private void sendDiscardChanges() {
+        final ListenableFuture<RpcResult<CompositeNode>> discardFuture = rpc.invokeRpc(NETCONF_DISCARD_CHANGES_QNAME, DISCARD_CHANGES_RPC_CONTENT);
+        Futures.addCallback(discardFuture, new FutureCallback<RpcResult<CompositeNode>>() {
+            @Override
+            public void onSuccess(final RpcResult<CompositeNode> result) {
+                LOG.debug("{}: Discarding transaction: {}", id, getIdentifier());
+            }
+
+            @Override
+            public void onFailure(final Throwable t) {
+                LOG.error("{}: Discarding changes failed, transaction: {}. Device configuration might be corrupted", id, getIdentifier(), t);
+                throw new RuntimeException(id + ": Discarding changes failed, transaction " + getIdentifier(), t);
+            }
+        });
+    }
+
     private CompositeNode createEditConfigStructure(final YangInstanceIdentifier dataPath, final Optional<ModifyAction> operation,
                                                     final Optional<CompositeNode> lastChildOverride) {
         Preconditions.checkArgument(Iterables.isEmpty(dataPath.getPathArguments()) == false, "Instance identifier with empty path %s", dataPath);
@@ -298,13 +354,6 @@ public class NetconfDeviceWriteOnlyTx implements DOMDataWriteTransaction {
         }
     }
 
-    private ImmutableCompositeNode getCommitRequest() {
-        final CompositeNodeBuilder<ImmutableCompositeNode> commitInput = ImmutableCompositeNode.builder();
-        commitInput.setQName(NETCONF_COMMIT_QNAME);
-        return commitInput.toInstance();
-    }
-
-
     @Override
     public Object getIdentifier() {
         return this;
index a6924d9..d3faddd 100644 (file)
@@ -35,6 +35,7 @@ import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.Node;
 import org.opendaylight.yangtools.yang.data.impl.CompositeNodeTOImpl;
 import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode;
+import org.opendaylight.yangtools.yang.data.impl.NodeFactory;
 import org.opendaylight.yangtools.yang.data.impl.SimpleNodeTOImpl;
 import org.opendaylight.yangtools.yang.data.impl.util.CompositeNodeBuilder;
 import org.opendaylight.yangtools.yang.model.api.DataNodeContainer;
@@ -45,8 +46,7 @@ import org.w3c.dom.Element;
 
 public class NetconfMessageTransformUtil {
 
-    private NetconfMessageTransformUtil() {
-    }
+    private NetconfMessageTransformUtil() {}
 
     public static final QName IETF_NETCONF_MONITORING = QName.create("urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring", "2010-10-04", "ietf-netconf-monitoring");
     public static URI NETCONF_URI = URI.create("urn:ietf:params:xml:ns:netconf:base:1.0");
@@ -66,14 +66,27 @@ public class NetconfMessageTransformUtil {
     public static QName NETCONF_DEFAULT_OPERATION_QNAME = QName.create(NETCONF_OPERATION_QNAME, "default-operation");
     public static QName NETCONF_EDIT_CONFIG_QNAME = QName.create(NETCONF_QNAME, "edit-config");
     public static QName NETCONF_GET_CONFIG_QNAME = QName.create(NETCONF_QNAME, "get-config");
+    public static QName NETCONF_DISCARD_CHANGES_QNAME = QName.create(NETCONF_QNAME, "discard-changes");
     public static QName NETCONF_TYPE_QNAME = QName.create(NETCONF_QNAME, "type");
     public static QName NETCONF_FILTER_QNAME = QName.create(NETCONF_QNAME, "filter");
     public static QName NETCONF_GET_QNAME = QName.create(NETCONF_QNAME, "get");
     public static QName NETCONF_RPC_QNAME = QName.create(NETCONF_QNAME, "rpc");
+
     public static URI NETCONF_ROLLBACK_ON_ERROR_URI = URI
             .create("urn:ietf:params:netconf:capability:rollback-on-error:1.0");
     public static String ROLLBACK_ON_ERROR_OPTION = "rollback-on-error";
 
+    public static URI NETCONF_CANDIDATE_URI = URI
+            .create("urn:ietf:params:netconf:capability:candidate:1.0");
+
+    // Discard changes message
+    public static final CompositeNode DISCARD_CHANGES_RPC_CONTENT =
+            NodeFactory.createImmutableCompositeNode(NETCONF_DISCARD_CHANGES_QNAME, null, Collections.<Node<?>>emptyList());
+
+    // Commit changes message
+    public static final CompositeNode COMMIT_RPC_CONTENT =
+            NodeFactory.createImmutableCompositeNode(NETCONF_COMMIT_QNAME, null, Collections.<Node<?>>emptyList());
+
     public static Node<?> toFilterStructure(final YangInstanceIdentifier identifier) {
         Node<?> previous = null;
         if (Iterables.isEmpty(identifier.getPathArguments())) {
@@ -269,5 +282,4 @@ public class NetconfMessageTransformUtil {
             return it.toInstance();
         }
     }
-
 }
index 6bad479..e13398b 100644 (file)
@@ -58,6 +58,14 @@ module odl-sal-netconf-connector-cfg {
                 type string;
             }
 
+            container yang-module-capabilities {
+                leaf-list capability {
+                    type string;
+                    description "Set a list of capabilities to override capabilities provided in device's hello message.
+                                 Can be used for devices that do not report any yang modules in their hello message";
+                }
+            }
+
             container dom-registry {
                 uses config:service-ref {
                     refine type {
index 391bf9c..001b9a8 100644 (file)
@@ -8,35 +8,35 @@
 
 package org.opendaylight.controller.sal.connect.netconf.listener;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.RPC_REPLY_KEY;
+import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.ListenableFuture;
 import io.netty.channel.ChannelFuture;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
-
 import java.io.ByteArrayInputStream;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-
 import javax.xml.parsers.DocumentBuilderFactory;
 import javax.xml.parsers.ParserConfigurationException;
-
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.same;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.RPC_REPLY_KEY;
-import static org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0;
-
 import org.apache.commons.lang3.StringUtils;
 import org.junit.Before;
 import org.junit.Test;
@@ -56,10 +56,6 @@ import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 
-import com.google.common.base.Strings;
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.ListenableFuture;
-
 public class NetconfDeviceCommunicatorTest {
 
     @Mock
@@ -129,9 +125,9 @@ public class NetconfDeviceCommunicatorTest {
         verify( mockDevice ).onRemoteSessionUp( netconfSessionCapabilities.capture(), eq( communicator ) );
 
         NetconfSessionCapabilities actualCapabilites = netconfSessionCapabilities.getValue();
-        assertEquals( "containsCapability", true, actualCapabilites.containsCapability(
-                                NetconfMessageTransformUtil.NETCONF_ROLLBACK_ON_ERROR_URI.toString() ) );
-        assertEquals( "containsCapability", true, actualCapabilites.containsCapability( testCapability ) );
+        assertEquals( "containsModuleCapability", true, actualCapabilites.containsNonModuleCapability(
+                NetconfMessageTransformUtil.NETCONF_ROLLBACK_ON_ERROR_URI.toString()) );
+        assertEquals( "containsModuleCapability", false, actualCapabilites.containsNonModuleCapability(testCapability) );
         assertEquals( "getModuleBasedCaps", Sets.newHashSet(
                             QName.create( "urn:opendaylight:params:xml:ns:test", "2014-06-02", "test-module" )),
                       actualCapabilites.getModuleBasedCaps() );
diff --git a/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilitiesTest.java b/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilitiesTest.java
new file mode 100644 (file)
index 0000000..87947b5
--- /dev/null
@@ -0,0 +1,50 @@
+package org.opendaylight.controller.sal.connect.netconf.listener;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import org.junit.Test;
+import org.junit.matchers.JUnitMatchers;
+import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
+import org.opendaylight.yangtools.yang.common.QName;
+
+public class NetconfSessionCapabilitiesTest {
+
+    @Test
+    public void testMerge() throws Exception {
+        final List<String> caps1 = Lists.newArrayList(
+                "namespace:1?module=module1&revision=2012-12-12",
+                "namespace:2?module=module2&amp;revision=2012-12-12",
+                "urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring?module=ietf-netconf-monitoring&amp;revision=2010-10-04",
+                "urn:ietf:params:netconf:base:1.0",
+                "urn:ietf:params:netconf:capability:rollback-on-error:1.0"
+        );
+        final NetconfSessionCapabilities sessionCaps1 = NetconfSessionCapabilities.fromStrings(caps1);
+        assertCaps(sessionCaps1, 2, 3);
+
+        final List<String> caps2 = Lists.newArrayList(
+                "namespace:3?module=module3&revision=2012-12-12",
+                "namespace:4?module=module4&revision=2012-12-12",
+                "randomNonModuleCap"
+        );
+        final NetconfSessionCapabilities sessionCaps2 = NetconfSessionCapabilities.fromStrings(caps2);
+        assertCaps(sessionCaps2, 1, 2);
+
+        final NetconfSessionCapabilities merged = sessionCaps1.replaceModuleCaps(sessionCaps2);
+        assertCaps(merged, 2, 2 + 1 /*Preserved monitoring*/);
+        for (final QName qName : sessionCaps2.getModuleBasedCaps()) {
+            assertThat(merged.getModuleBasedCaps(), JUnitMatchers.hasItem(qName));
+        }
+        assertThat(merged.getModuleBasedCaps(), JUnitMatchers.hasItem(NetconfMessageTransformUtil.IETF_NETCONF_MONITORING));
+
+        assertThat(merged.getNonModuleCaps(), JUnitMatchers.hasItem("urn:ietf:params:netconf:base:1.0"));
+        assertThat(merged.getNonModuleCaps(), JUnitMatchers.hasItem("urn:ietf:params:netconf:capability:rollback-on-error:1.0"));
+    }
+
+    private void assertCaps(final NetconfSessionCapabilities sessionCaps1, final int nonModuleCaps, final int moduleCaps) {
+        assertEquals(nonModuleCaps, sessionCaps1.getNonModuleCaps().size());
+        assertEquals(moduleCaps, sessionCaps1.getModuleBasedCaps().size());
+    }
+}
diff --git a/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/tx/NetconfDeviceWriteOnlyTxTest.java b/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/tx/NetconfDeviceWriteOnlyTxTest.java
new file mode 100644 (file)
index 0000000..a65e426
--- /dev/null
@@ -0,0 +1,79 @@
+package org.opendaylight.controller.sal.connect.netconf.sal.tx;
+
+import static junit.framework.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.DISCARD_CHANGES_RPC_CONTENT;
+
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.Futures;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizer;
+import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
+import org.opendaylight.controller.sal.core.api.RpcImplementation;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
+import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+
+public class NetconfDeviceWriteOnlyTxTest {
+
+    private final RemoteDeviceId id = new RemoteDeviceId("test-mount");
+
+    @Mock
+    private RpcImplementation rpc;
+    @Mock
+    private DataNormalizer normalizer;
+    private YangInstanceIdentifier yangIId;
+
+    @Before
+    public void setUp() throws Exception {
+        MockitoAnnotations.initMocks(this);
+
+        doReturn(Futures.<RpcResult<CompositeNode>>immediateFailedFuture(new IllegalStateException("Failed tx")))
+        .doReturn(Futures.immediateFuture(RpcResultBuilder.<CompositeNode>success().build()))
+                .when(rpc).invokeRpc(any(QName.class), any(CompositeNode.class));
+
+        yangIId = YangInstanceIdentifier.builder().node(QName.create("namespace", "2012-12-12", "name")).build();
+        doReturn(yangIId).when(normalizer).toLegacy(yangIId);
+    }
+
+    @Test
+    public void testDiscardCahnges() {
+        final NetconfDeviceWriteOnlyTx tx = new NetconfDeviceWriteOnlyTx(id, rpc, normalizer, true, true);
+        final CheckedFuture<Void, TransactionCommitFailedException> submitFuture = tx.submit();
+        try {
+            submitFuture.checkedGet();
+        } catch (final TransactionCommitFailedException e) {
+            // verify discard changes was sent
+            verify(rpc).invokeRpc(NetconfMessageTransformUtil.NETCONF_DISCARD_CHANGES_QNAME, DISCARD_CHANGES_RPC_CONTENT);
+            return;
+        }
+
+        fail("Submit should fail");
+    }
+
+
+    @Test
+    public void testDiscardCahngesNotSentWithoutCandidate() {
+        doReturn(Futures.immediateFuture(RpcResultBuilder.<CompositeNode>success().build()))
+        .doReturn(Futures.<RpcResult<CompositeNode>>immediateFailedFuture(new IllegalStateException("Failed tx")))
+                .when(rpc).invokeRpc(any(QName.class), any(CompositeNode.class));
+
+        final NetconfDeviceWriteOnlyTx tx = new NetconfDeviceWriteOnlyTx(id, rpc, normalizer, false, true);
+        tx.delete(LogicalDatastoreType.CONFIGURATION, yangIId);
+        verify(rpc).invokeRpc(eq(NetconfMessageTransformUtil.NETCONF_EDIT_CONFIG_QNAME), any(CompositeNode.class));
+        verifyNoMoreInteractions(rpc);
+    }
+
+}
index d0be2cb..4791979 100644 (file)
@@ -24,5 +24,9 @@ machine
 
 6. !!!WARNING!!! - never edit the generated sources files of protocol buffer
 
-7. !!!NOTE!!! if you are planning to add new .proto file  option java_package should begin with
+7. !!!IMPORTANT!!! if you are planning to add new .proto file  option java_package should begin with
    org.opendaylight.controller.protobuff.messages to properly exclude from sonar.
+
+8. !!!IMPORTANT!!! for any new .proto file added you need to create corresponding version-compatibility-serialized-data
+   serialized test file under test/resources and a test case for one message in your .proto file.
+   Please follow the instruction in readme.txt in that folder
index ce12080..03d632b 100644 (file)
@@ -1,3 +1,13 @@
+/*
+ *
+ *  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.node;
 
 import com.google.common.base.Preconditions;
index 7b6f5c4..14ca75b 100644 (file)
@@ -1,9 +1,11 @@
 /*
- * 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
+ *  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.node;
index fc4b395..17bdb36 100644 (file)
@@ -1,3 +1,13 @@
+/*
+ *
+ *  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.node;
 
 import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
index fb15312..68d3c59 100644 (file)
@@ -1,3 +1,13 @@
+/*
+ *
+ *  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.node;
 
 import com.google.common.base.Preconditions;