aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt2
-rw-r--r--ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt63
-rw-r--r--ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImplTest.kt11
3 files changed, 42 insertions, 34 deletions
diff --git a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
index 6ef4f41fb..aa156e2a8 100644
--- a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
+++ b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfDeviceCommunicator.kt
@@ -57,7 +57,7 @@ class NetconfDeviceCommunicator(private var inputStream: InputStream,
while (!socketClosed) {
val cInt = bufferReader.read()
if (cInt == -1) {
- log.error("$deviceInfo: Received cInt = -1")
+ log.debug("$deviceInfo: Received end of stream, closing socket.")
socketClosed = true
}
val c = cInt.toChar()
diff --git a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
index 7e56e3e51..13c1bfa37 100644
--- a/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
+++ b/ms/blueprintsprocessor/functions/netconf-executor/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImpl.kt
@@ -80,11 +80,11 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
RpcStatus.FAILURE, true)) {
rpcService.closeSession(true)
}
-
- session.close()
- // Closes the socket which should interrupt the streamHandler
- channel.close()
- client.close()
+ try {
+ close()
+ } catch (ioe: IOException) {
+ log.warn("$deviceInfo: Error closing session($sessionId) for host($deviceInfo)", ioe)
+ }
}
override fun reconnect() {
@@ -98,8 +98,8 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
checkAndReestablish()
try {
- return streamHandler.sendMessage(formattedRequest, messageId).get(replyTimeout.toLong(), TimeUnit.SECONDS)
-// replies.remove(messageId)
+ return streamHandler.getFutureFromSendMessage(streamHandler.sendMessage(formattedRequest, messageId),
+ replyTimeout.toLong(), TimeUnit.SECONDS)
} catch (e: InterruptedException) {
Thread.currentThread().interrupt()
throw NetconfException("$deviceInfo: Interrupted while waiting for reply for request: $formattedRequest", e)
@@ -109,10 +109,7 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
} catch (e: ExecutionException) {
log.warn("$deviceInfo: Closing session($sessionId) due to unexpected Error", e)
try {
- session.close()
- // Closes the socket which should interrupt the streamHandler
- channel.close()
- client.close()
+ close()
} catch (ioe: IOException) {
log.warn("$deviceInfo: Error closing session($sessionId) for host($deviceInfo)", ioe)
}
@@ -138,20 +135,23 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
override fun checkAndReestablish() {
try {
- if (client.isClosed) {
- log.info("Trying to restart the whole SSH connection with {}", deviceInfo)
- clearReplies()
- startConnection()
- } else if (session.isClosed) {
- log.info("Trying to restart the session with {}", deviceInfo)
- clearReplies()
- startSession()
- } else if (channel.isClosed) {
- log.info("Trying to reopen the channel with {}", deviceInfo)
- clearReplies()
- openChannel()
- } else {
- return
+ when {
+ client.isClosed -> {
+ log.info("Trying to restart the whole SSH connection with {}", deviceInfo)
+ clearReplies()
+ startConnection()
+ }
+ session.isClosed -> {
+ log.info("Trying to restart the session with {}", deviceInfo)
+ clearReplies()
+ startSession()
+ }
+ channel.isClosed -> {
+ log.info("Trying to reopen the channel with {}", deviceInfo)
+ clearReplies()
+ openChannel()
+ }
+ else -> return
}
} catch (e: IOException) {
log.error("Can't reopen connection for device {} error: {}", deviceInfo, e.message)
@@ -258,7 +258,7 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
val capabilityMatcher = NetconfMessageUtils.CAPABILITY_REGEX_PATTERN.matcher(serverHelloResponse)
while (capabilityMatcher.find()) {
- deviceCapabilities.plus(capabilityMatcher.group(1))
+ deviceCapabilities.add(capabilityMatcher.group(1))
}
}
@@ -293,6 +293,17 @@ class NetconfSessionImpl(private val deviceInfo: DeviceInfo, private val rpcServ
}
/**
+ * Closes the session/channel/client
+ */
+ @Throws(IOException::class)
+ private fun close() {
+ session.close()
+ // Closes the socket which should interrupt the streamHandler
+ channel.close()
+ client.close()
+ }
+
+ /**
* Internal function for accessing replies for testing.
*/
internal fun getReplies() = replies
diff --git a/ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImplTest.kt b/ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImplTest.kt
index 1f526f445..f5fd5410a 100644
--- a/ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImplTest.kt
+++ b/ms/blueprintsprocessor/functions/netconf-executor/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/netconf/executor/core/NetconfSessionImplTest.kt
@@ -168,7 +168,6 @@ class NetconfSessionImplTest {
}
- @Ignore //TODO undo close method removal
@Test
fun `disconnect wraps exception from ssh closing error`() {
val netconfSessionSpy = spyk(netconfSession, recordPrivateCalls = true)
@@ -295,7 +294,6 @@ class NetconfSessionImplTest {
}
}
- @Ignore //TODO revert back on getFutureFromSendMessage
@Test
fun `syncRpc throws NetconfException if TimeoutException is caught`() {
val expectedExceptionMsg = "$deviceInfo: Timed out while waiting for reply for request $formattedRequest after ${deviceInfo.replyTimeout} sec."
@@ -310,23 +308,23 @@ class NetconfSessionImplTest {
}
}
- @Ignore
@Test
fun `syncRpc throws NetconfException if ExecutionException is caught`() {
val expectedExceptionMsg = "$deviceInfo: Closing session $sessionId for request $formattedRequest"
assertFailsWith(exceptionClass = NetconfException::class, message = expectedExceptionMsg) {
- val netconfSessionSpy = spyk(netconfSession)
+ val netconfSessionSpy = spyk(netconfSession, recordPrivateCalls = false)
val futureRet: CompletableFuture<String> = CompletableFuture.completedFuture(futureMsg)
every { netconfCommunicator.sendMessage(any(), any()) } returns futureRet
every { netconfCommunicator.getFutureFromSendMessage(any(), any(), any()) } throws
- ExecutionException("exec exception", Exception("nested exception")) //TODO revert getFutureFromSendMessage back
+ ExecutionException("exec exception", Exception("nested exception"))
+ every { netconfSessionSpy["close"]() as Unit } just Runs
every { netconfSessionSpy.checkAndReestablish() } just Runs
+ netconfSessionSpy.setSession(mockClientSession)
//call the method
netconfSessionSpy.syncRpc("0", "0")
}
}
- @Ignore //TODO revert back on getFutureFromSendMessage
@Test
fun `syncRpc throws NetconfException if caught ExecutionException and failed to close SSH session`() {
val expectedExceptionMsg = "$deviceInfo: Closing session $sessionId for request $formattedRequest"
@@ -525,7 +523,6 @@ class NetconfSessionImplTest {
verify { mockSshClient.close() }
}
- @Ignore
@Test
fun `disconnect wraps IOException if channel doesn't close`() { //this test is equivalent to others
every { rpcService.closeSession(false) } returns SUCCESSFUL_DEVICE_RESPONSE