import org.apache.sshd.client.channel.ClientChannel
import org.apache.sshd.client.session.ClientSession
import org.apache.sshd.common.FactoryManager
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.*
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.DeviceInfo
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfException
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfRpcService
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfSession
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.api.NetconfSessionListener
import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.NetconfMessageUtils
import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.RpcMessageUtils
import org.onap.ccsdk.cds.blueprintsprocessor.functions.netconf.executor.utils.RpcStatus
import org.slf4j.LoggerFactory
import java.io.IOException
-import java.util.*
-import java.util.concurrent.*
+import java.util.Collections
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcService: NetconfRpcService) :
NetconfSession {
override fun connect() {
try {
- log.info("$deviceInfo: Connecting to Netconf Device with timeouts C:${deviceInfo.connectTimeout}, " +
- "R:${deviceInfo.replyTimeout}, I:${deviceInfo.idleTimeout}")
+ log.info(
+ "$deviceInfo: Connecting to Netconf Device with timeouts C:${deviceInfo.connectTimeout}, " +
+ "R:${deviceInfo.replyTimeout}, I:${deviceInfo.idleTimeout}"
+ )
startConnection()
log.info("$deviceInfo: Connected to Netconf Device")
} catch (e: NetconfException) {
override fun disconnect() {
var retryNum = 3
- while(rpcService.closeSession(false).status
- .equals(RpcStatus.FAILURE, true) &&retryNum>0) {
+ while (rpcService.closeSession(false).status
+ .equals(RpcStatus.FAILURE, true) && retryNum > 0
+ ) {
log.error("disconnect: graceful disconnect failed, retrying $retryNum times...")
- retryNum--;
+ retryNum--
}
- //if we can't close the session, try to force terminate.
- if(retryNum == 0) {
+ // if we can't close the session, try to force terminate.
+ if (retryNum == 0) {
log.error("disconnect: trying to force-terminate the session.")
rpcService.closeSession(true)
}
checkAndReestablish()
try {
- return streamHandler.getFutureFromSendMessage(streamHandler.sendMessage(formattedRequest, messageId),
- replyTimeout.toLong(), TimeUnit.SECONDS)
+ return streamHandler.getFutureFromSendMessage(
+ streamHandler.sendMessage(formattedRequest, messageId),
+ replyTimeout.toLong(), TimeUnit.SECONDS
+ )
} catch (e: InterruptedException) {
throw NetconfException("$deviceInfo: Interrupted while waiting for reply for request: $formattedRequest", e)
} catch (e: TimeoutException) {
- throw NetconfException("$deviceInfo: Timed out while waiting for reply for request $formattedRequest after $replyTimeout sec.",
- e)
+ throw NetconfException(
+ "$deviceInfo: Timed out while waiting for reply for request $formattedRequest after $replyTimeout sec.",
+ e
+ )
} catch (e: ExecutionException) {
log.warn("$deviceInfo: Closing session($sessionId) due to unexpected Error", e)
try {
} catch (e: Exception) {
throw NetconfException("$deviceInfo: Failed to establish SSH session", e)
}
-
}
- //Needed to unit test connect method interacting with client.start in startClient() below
+ // Needed to unit test connect method interacting with client.start in startClient() below
private fun setupNewSSHClient() {
client = SshClient.setUpDefaultClient()
}
private fun authSession() {
session.addPasswordIdentity(deviceInfo.password)
session.auth().verify(connectionTimeout, TimeUnit.SECONDS)
- val event = session.waitFor(ImmutableSet.of(ClientSession.ClientSessionEvent.WAIT_AUTH,
- ClientSession.ClientSessionEvent.CLOSED, ClientSession.ClientSessionEvent.AUTHED), 0)
+ val event = session.waitFor(
+ ImmutableSet.of(
+ ClientSession.ClientSessionEvent.WAIT_AUTH,
+ ClientSession.ClientSessionEvent.CLOSED, ClientSession.ClientSessionEvent.AUTHED
+ ), 0
+ )
if (!event.contains(ClientSession.ClientSessionEvent.AUTHED)) {
throw NetconfException("$deviceInfo: Failed to authenticate session.")
}
private fun setupHandler() {
val sessionListener: NetconfSessionListener = NetconfSessionListenerImpl(this)
- streamHandler = NetconfDeviceCommunicator(channel.invertedOut, channel.invertedIn, deviceInfo,
- sessionListener, replies)
+ streamHandler = NetconfDeviceCommunicator(
+ channel.invertedOut, channel.invertedIn, deviceInfo,
+ sessionListener, replies
+ )
exchangeHelloMessage()
}
}
val capabilityMatcher = NetconfMessageUtils.CAPABILITY_REGEX_PATTERN.matcher(serverHelloResponse)
- while (capabilityMatcher.find()) { //TODO: refactor to add unit test easily for device capability accumulation.
+ while (capabilityMatcher.find()) { // TODO: refactor to add unit test easily for device capability accumulation.
deviceCapabilities.add(capabilityMatcher.group(1))
}
}
* internal function for accessing errorReplies for testing.
*/
internal fun getErrorReplies() = errorReplies
+
internal fun clearErrorReplies() = errorReplies.clear()
internal fun clearReplies() = replies.clear()
- internal fun setClient(client: SshClient) { this.client = client }
- internal fun setSession(session: ClientSession) { this.session = session }
- internal fun setChannel(channel: ClientChannel) { this.channel = channel }
+ internal fun setClient(client: SshClient) {
+ this.client = client
+ }
+
+ internal fun setSession(session: ClientSession) {
+ this.session = session
+ }
+
+ internal fun setChannel(channel: ClientChannel) {
+ this.channel = channel
+ }
}