package org.opendaylight.netconf.topology.singleton.impl;
-import akka.actor.ActorRef;
-import akka.util.Timeout;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
-import io.netty.util.concurrent.EventExecutor;
import java.math.BigDecimal;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.Optional;
import javax.annotation.Nullable;
import org.opendaylight.aaa.encrypt.AAAEncryptionService;
-import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
import org.opendaylight.netconf.api.NetconfMessage;
import org.opendaylight.netconf.client.NetconfClientSessionListener;
import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfiguration;
import org.opendaylight.netconf.client.conf.NetconfReconnectingClientConfigurationBuilder;
+import org.opendaylight.netconf.nettyutil.ReconnectStrategyFactory;
+import org.opendaylight.netconf.nettyutil.TimedReconnectStrategyFactory;
import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler;
import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.LoginPasswordHandler;
import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
import org.opendaylight.netconf.sal.connect.netconf.sal.NetconfKeystoreAdapter;
import org.opendaylight.netconf.sal.connect.netconf.schema.YangLibrarySchemaYangSourceProvider;
import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
+import org.opendaylight.netconf.sal.connect.util.SslHandlerFactoryImpl;
import org.opendaylight.netconf.topology.singleton.api.RemoteDeviceConnector;
import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfConnectorDTO;
import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
-import org.opendaylight.protocol.framework.ReconnectStrategy;
-import org.opendaylight.protocol.framework.ReconnectStrategyFactory;
-import org.opendaylight.protocol.framework.TimedReconnectStrategy;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Host;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Uri;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.OdlHelloMessageCapabilities;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.parameters.Protocol;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapability.CapabilityOrigin;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.Credentials;
import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.credentials.credentials.KeyAuth;
private final NetconfTopologySetup netconfTopologyDeviceSetup;
private final RemoteDeviceId remoteDeviceId;
- private final DOMMountPointService mountService;
- private final Timeout actorResponseWaitTime;
private final String privateKeyPath;
private final String privateKeyPassphrase;
private final AAAEncryptionService encryptionService;
private final NetconfKeystoreAdapter keystoreAdapter;
public RemoteDeviceConnectorImpl(final NetconfTopologySetup netconfTopologyDeviceSetup,
- final RemoteDeviceId remoteDeviceId, final Timeout actorResponseWaitTime,
- final DOMMountPointService mountService) {
+ final RemoteDeviceId remoteDeviceId) {
this.netconfTopologyDeviceSetup = Preconditions.checkNotNull(netconfTopologyDeviceSetup);
this.remoteDeviceId = remoteDeviceId;
- this.actorResponseWaitTime = actorResponseWaitTime;
- this.mountService = mountService;
this.privateKeyPath = netconfTopologyDeviceSetup.getPrivateKeyPath();
this.privateKeyPassphrase = netconfTopologyDeviceSetup.getPrivateKeyPassphrase();
this.encryptionService = netconfTopologyDeviceSetup.getEncryptionService();
}
@Override
- public void startRemoteDeviceConnection(final ActorRef deviceContextActorRef) {
+ public void startRemoteDeviceConnection(final RemoteDeviceHandler<NetconfSessionPreferences> deviceHandler) {
- final NetconfNode netconfNode = netconfTopologyDeviceSetup.getNode().getAugmentation(NetconfNode.class);
+ final NetconfNode netconfNode = netconfTopologyDeviceSetup.getNode().augmentation(NetconfNode.class);
final NodeId nodeId = netconfTopologyDeviceSetup.getNode().getNodeId();
Preconditions.checkNotNull(netconfNode.getHost());
Preconditions.checkNotNull(netconfNode.getPort());
- Preconditions.checkNotNull(netconfNode.isTcpOnly());
- this.deviceCommunicatorDTO = createDeviceCommunicator(nodeId, netconfNode, deviceContextActorRef);
+ this.deviceCommunicatorDTO = createDeviceCommunicator(nodeId, netconfNode, deviceHandler);
final NetconfDeviceCommunicator deviceCommunicator = deviceCommunicatorDTO.getCommunicator();
final NetconfClientSessionListener netconfClientSessionListener = deviceCommunicatorDTO.getSessionListener();
final NetconfReconnectingClientConfiguration clientConfig =
@Override
public void onFailure(@Nullable final Throwable throwable) {
- LOG.error("{}: Connector failed, {}", remoteDeviceId, throwable);
+ LOG.error("{}: Connector failed", remoteDeviceId, throwable);
}
}, MoreExecutors.directExecutor());
}
@SuppressWarnings("checkstyle:IllegalCatch")
@Override
public void stopRemoteDeviceConnection() {
- Preconditions.checkNotNull(deviceCommunicatorDTO, remoteDeviceId + ": Device communicator was not created.");
- try {
- deviceCommunicatorDTO.close();
- } catch (final Exception e) {
- LOG.error("{}: Error at closing device communicator.", remoteDeviceId, e);
+ if (deviceCommunicatorDTO != null) {
+ try {
+ deviceCommunicatorDTO.close();
+ } catch (final Exception e) {
+ LOG.error("{}: Error at closing device communicator.", remoteDeviceId, e);
+ }
}
}
@VisibleForTesting
NetconfConnectorDTO createDeviceCommunicator(final NodeId nodeId, final NetconfNode node,
- final ActorRef deviceContextActorRef) {
+ final RemoteDeviceHandler<NetconfSessionPreferences> deviceHandler) {
//setup default values since default value is not supported in mdsal
final long defaultRequestTimeoutMillis = node.getDefaultRequestTimeoutMillis() == null
? NetconfTopologyUtils.DEFAULT_REQUEST_TIMEOUT_MILLIS : node.getDefaultRequestTimeoutMillis();
final boolean reconnectOnChangedSchema = node.isReconnectOnChangedSchema() == null
? NetconfTopologyUtils.DEFAULT_RECONNECT_ON_CHANGED_SCHEMA : node.isReconnectOnChangedSchema();
- RemoteDeviceHandler<NetconfSessionPreferences> salFacade = new MasterSalFacade(remoteDeviceId,
- netconfTopologyDeviceSetup.getActorSystem(), deviceContextActorRef, actorResponseWaitTime,
- mountService, netconfTopologyDeviceSetup.getDataBroker());
+ RemoteDeviceHandler<NetconfSessionPreferences> salFacade = deviceHandler;
if (keepaliveDelay > 0) {
LOG.info("{}: Adding keepalive facade.", remoteDeviceId);
salFacade = new KeepaliveSalFacade(remoteDeviceId, salFacade,
- netconfTopologyDeviceSetup.getKeepaliveExecutor().getExecutor(), keepaliveDelay,
+ netconfTopologyDeviceSetup.getKeepaliveExecutor(), keepaliveDelay,
defaultRequestTimeoutMillis);
}
final NetconfDevice.SchemaResourcesDTO schemaResourcesDTO = netconfTopologyDeviceSetup.getSchemaResourcesDTO();
-
// pre register yang library sources as fallback schemas to schema registry
final List<SchemaSourceRegistration<YangTextSchemaSource>> registeredYangLibSources = Lists.newArrayList();
if (node.getYangLibrary() != null) {
device = new NetconfDeviceBuilder()
.setReconnectOnSchemasChange(reconnectOnChangedSchema)
.setSchemaResourcesDTO(schemaResourcesDTO)
- .setGlobalProcessingExecutor(netconfTopologyDeviceSetup.getProcessingExecutor().getExecutor())
+ .setGlobalProcessingExecutor(netconfTopologyDeviceSetup.getProcessingExecutor())
.setId(remoteDeviceId)
.setSalFacade(salFacade)
.build();
LOG.info("{}: Concurrent rpc limit is smaller than 1, no limit will be enforced.", remoteDeviceId);
}
- return new NetconfConnectorDTO(
- userCapabilities.isPresent() ? new NetconfDeviceCommunicator(remoteDeviceId, device,
- new UserPreferences(userCapabilities.get(),
- Objects.isNull(node.getYangModuleCapabilities())
- ? false : node.getYangModuleCapabilities().isOverride(),
- Objects.isNull(node.getNonModuleCapabilities())
- ? false : node.getNonModuleCapabilities().isOverride()), rpcMessageLimit)
- : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit), salFacade);
+ NetconfDeviceCommunicator netconfDeviceCommunicator =
+ userCapabilities.isPresent() ? new NetconfDeviceCommunicator(remoteDeviceId, device,
+ new UserPreferences(userCapabilities.get(),
+ Objects.isNull(node.getYangModuleCapabilities())
+ ? false : node.getYangModuleCapabilities().isOverride(),
+ Objects.isNull(node.getNonModuleCapabilities())
+ ? false : node.getNonModuleCapabilities().isOverride()), rpcMessageLimit)
+ : new NetconfDeviceCommunicator(remoteDeviceId, device, rpcMessageLimit);
+
+ if (salFacade instanceof KeepaliveSalFacade) {
+ ((KeepaliveSalFacade)salFacade).setListener(netconfDeviceCommunicator);
+ }
+ return new NetconfConnectorDTO(netconfDeviceCommunicator, salFacade);
}
- private Optional<NetconfSessionPreferences> getUserCapabilities(final NetconfNode node) {
+ private static Optional<NetconfSessionPreferences> getUserCapabilities(final NetconfNode node) {
if (node.getYangModuleCapabilities() == null && node.getNonModuleCapabilities() == null) {
return Optional.empty();
}
}
//TODO: duplicate code
- private InetSocketAddress getSocketAddress(final Host host, final int port) {
+ private static InetSocketAddress getSocketAddress(final Host host, final int port) {
if (host.getDomainName() != null) {
return new InetSocketAddress(host.getDomainName().getValue(), port);
} else {
? NetconfTopologyUtils.DEFAULT_MAX_CONNECTION_ATTEMPTS : node.getMaxConnectionAttempts();
final int betweenAttemptsTimeoutMillis = node.getBetweenAttemptsTimeoutMillis() == null
? NetconfTopologyUtils.DEFAULT_BETWEEN_ATTEMPTS_TIMEOUT_MILLIS : node.getBetweenAttemptsTimeoutMillis();
+ final boolean isTcpOnly = node.isTcpOnly() == null
+ ? NetconfTopologyUtils.DEFAULT_IS_TCP_ONLY : node.isTcpOnly();
final BigDecimal sleepFactor = node.getSleepFactor() == null
? NetconfTopologyUtils.DEFAULT_SLEEP_FACTOR : node.getSleepFactor();
final ReconnectStrategyFactory sf =
new TimedReconnectStrategyFactory(netconfTopologyDeviceSetup.getEventExecutor(), maxConnectionAttempts,
betweenAttemptsTimeoutMillis, sleepFactor);
- final ReconnectStrategy strategy = sf.createReconnectStrategy();
- final AuthenticationHandler authHandler = getHandlerFromCredentials(node.getCredentials());
- return NetconfReconnectingClientConfigurationBuilder.create()
+ final NetconfReconnectingClientConfigurationBuilder reconnectingClientConfigurationBuilder;
+ final Protocol protocol = node.getProtocol();
+ if (isTcpOnly) {
+ reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
+ .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TCP)
+ .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
+ } else if (protocol == null || protocol.getName() == Protocol.Name.SSH) {
+ reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
+ .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH)
+ .withAuthHandler(getHandlerFromCredentials(node.getCredentials()));
+ } else if (protocol.getName() == Protocol.Name.TLS) {
+ reconnectingClientConfigurationBuilder = NetconfReconnectingClientConfigurationBuilder.create()
+ .withSslHandlerFactory(new SslHandlerFactoryImpl(keystoreAdapter, protocol.getSpecification()))
+ .withProtocol(NetconfClientConfiguration.NetconfClientProtocol.TLS);
+ } else {
+ throw new IllegalStateException("Unsupported protocol type: " + protocol.getName());
+ }
+
+ final List<Uri> odlHelloCapabilities = getOdlHelloCapabilities(node);
+ if (odlHelloCapabilities != null) {
+ reconnectingClientConfigurationBuilder.withOdlHelloCapabilities(odlHelloCapabilities);
+ }
+
+ return reconnectingClientConfigurationBuilder
.withAddress(socketAddress)
.withConnectionTimeoutMillis(clientConnectionTimeoutMillis)
- .withReconnectStrategy(strategy)
- .withAuthHandler(authHandler)
- .withProtocol(node.isTcpOnly()
- ? NetconfClientConfiguration.NetconfClientProtocol.TCP
- : NetconfClientConfiguration.NetconfClientProtocol.SSH)
+ .withReconnectStrategy(sf.createReconnectStrategy())
.withConnectStrategyFactory(sf)
.withSessionListener(listener)
.build();
}
+ private static List<Uri> getOdlHelloCapabilities(final NetconfNode node) {
+ final OdlHelloMessageCapabilities helloCapabilities = node.getOdlHelloMessageCapabilities();
+ return helloCapabilities != null ? helloCapabilities.getCapability() : null;
+ }
+
private AuthenticationHandler getHandlerFromCredentials(final Credentials credentials) {
if (credentials instanceof org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology
.rev150114.netconf.node.credentials.credentials.LoginPassword) {
}
throw new IllegalStateException("Unsupported credential type: " + credentials.getClass());
}
-
- private static final class TimedReconnectStrategyFactory implements ReconnectStrategyFactory {
- private final Long connectionAttempts;
- private final EventExecutor executor;
- private final double sleepFactor;
- private final int minSleep;
-
- TimedReconnectStrategyFactory(final EventExecutor executor, final Long maxConnectionAttempts,
- final int minSleep, final BigDecimal sleepFactor) {
- if (maxConnectionAttempts != null && maxConnectionAttempts > 0) {
- connectionAttempts = maxConnectionAttempts;
- } else {
- connectionAttempts = null;
- }
-
- this.sleepFactor = sleepFactor.doubleValue();
- this.executor = executor;
- this.minSleep = minSleep;
- }
-
- @Override
- public ReconnectStrategy createReconnectStrategy() {
- return new TimedReconnectStrategy(executor, minSleep,
- minSleep, sleepFactor, null /*maxSleep*/, connectionAttempts, null /*deadline*/);
- }
- }
}
Code Review / netconf.git / blobdiff