import static org.opendaylight.controller.sal.connect.netconf.NetconfMapping.toRpcMessage;
import static org.opendaylight.controller.sal.connect.netconf.NetconfMapping.wrap;
+import com.google.common.base.Preconditions;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.URI;
import org.opendaylight.controller.md.sal.common.api.data.DataModification;
import org.opendaylight.controller.md.sal.common.api.data.DataReader;
import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
+import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
import org.opendaylight.controller.sal.core.api.Broker.RpcRegistration;
import org.opendaylight.controller.sal.core.api.Provider;
NetconfDeviceListener listener;
+ private boolean rollbackSupported;
+
+ private NetconfClientConfiguration clientConfig;
+
public NetconfDevice(String name) {
this.name = name;
this.logger = LoggerFactory.getLogger(NetconfDevice.class + "#" + name);
checkState(schemaSourceProvider != null, "Schema Source Provider must be set.");
checkState(eventExecutor != null, "Event executor must be set.");
- listener = new NetconfDeviceListener(this);
+ Preconditions.checkArgument(clientConfig.getSessionListener() instanceof NetconfDeviceListener);
+ listener = (NetconfDeviceListener) clientConfig.getSessionListener();
logger.info("Starting NETCONF Client {} for address {}", name, socketAddress);
- dispatcher.createClient(socketAddress, listener, reconnectStrategy);
+ dispatcher.createClient(clientConfig);
}
Optional<SchemaContext> getSchemaContext() {
}
}
- void bringUp(SchemaSourceProvider<String> delegate, Set<QName> capabilities) {
- remoteSourceProvider = schemaSourceProvider.createInstanceFor(delegate);
- deviceContextProvider = new NetconfDeviceSchemaContextProvider(this, remoteSourceProvider);
- deviceContextProvider.createContextFromCapabilities(capabilities);
- if (mountInstance != null && getSchemaContext().isPresent()) {
- mountInstance.setSchemaContext(getSchemaContext().get());
- }
+ void bringUp(final SchemaSourceProvider<String> delegate, final Set<QName> capabilities, final boolean rollbackSupported) {
+ // This has to be called from separate thread, not from netty thread calling onSessionUp in DeviceListener.
+ // Reason: delegate.getSchema blocks thread when waiting for response
+ // however, if the netty thread is blocked, no incoming message can be processed
+ // ... netty should pick another thread from pool to process incoming message, but it does not http://netty.io/wiki/thread-model.html
+ // TODO redesign +refactor
+ processingExecutor.submit(new Runnable() {
+ @Override
+ public void run() {
+ NetconfDevice.this.rollbackSupported = rollbackSupported;
+ remoteSourceProvider = schemaSourceProvider.createInstanceFor(delegate);
+ deviceContextProvider = new NetconfDeviceSchemaContextProvider(NetconfDevice.this, remoteSourceProvider);
+ deviceContextProvider.createContextFromCapabilities(capabilities);
+ if (mountInstance != null && getSchemaContext().isPresent()) {
+ mountInstance.setSchemaContext(getSchemaContext().get());
+ }
- updateDeviceState(true, capabilities);
+ updateDeviceState(true, capabilities);
- if (mountInstance != null) {
- confReaderReg = mountInstance.registerConfigurationReader(ROOT_PATH, this);
- operReaderReg = mountInstance.registerOperationalReader(ROOT_PATH, this);
- commitHandlerReg = mountInstance.registerCommitHandler(ROOT_PATH, this);
+ if (mountInstance != null) {
+ confReaderReg = mountInstance.registerConfigurationReader(ROOT_PATH, NetconfDevice.this);
+ operReaderReg = mountInstance.registerOperationalReader(ROOT_PATH, NetconfDevice.this);
+ commitHandlerReg = mountInstance.registerCommitHandler(ROOT_PATH, NetconfDevice.this);
- List<RpcRegistration> rpcs = new ArrayList<>();
- // TODO same condition twice
- if (mountInstance != null && getSchemaContext().isPresent()) {
- for (RpcDefinition rpc : mountInstance.getSchemaContext().getOperations()) {
- rpcs.add(mountInstance.addRpcImplementation(rpc.getQName(), this));
+ List<RpcRegistration> rpcs = new ArrayList<>();
+ // TODO same condition twice
+ if (mountInstance != null && getSchemaContext().isPresent()) {
+ for (RpcDefinition rpc : mountInstance.getSchemaContext().getOperations()) {
+ rpcs.add(mountInstance.addRpcImplementation(rpc.getQName(), NetconfDevice.this));
+ }
+ }
+ rpcReg = rpcs;
}
}
- rpcReg = rpcs;
- }
+ });
}
private void updateDeviceState(boolean up, Set<QName> capabilities) {
@Override
public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
- return listener.sendRequest(toRpcMessage(rpc, input, getSchemaContext()));
+ return listener.sendRequest(toRpcMessage(rpc, input, getSchemaContext()), rpc);
}
@Override
public DataCommitTransaction<InstanceIdentifier, CompositeNode> requestCommit(
DataModification<InstanceIdentifier, CompositeNode> modification) {
NetconfDeviceTwoPhaseCommitTransaction twoPhaseCommit = new NetconfDeviceTwoPhaseCommitTransaction(this,
- modification, true);
+ modification, true, rollbackSupported);
try {
twoPhaseCommit.prepare();
} catch (InterruptedException e) {
public void setDispatcher(final NetconfClientDispatcher dispatcher) {
this.dispatcher = dispatcher;
}
+
+ public void setClientConfig(final NetconfClientConfiguration clientConfig) {
+ this.clientConfig = clientConfig;
+ }
+
}
class NetconfDeviceSchemaContextProvider {