Skip to content

#62 getRpcReply hang issue is fixed. #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

#62 getRpcReply hang issue is fixed. #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions src/main/java/net/juniper/netconf/NetconfSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,9 +136,10 @@ String getRpcReply(String rpc) throws IOException {
int promptPosition;
while ((promptPosition = rpcReply.indexOf(NetconfConstants.DEVICE_PROMPT)) < 0 &&
(timeoutNotExceeded = (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime) < commandTimeout))) {
int charsRead = in.read(buffer, 0, buffer.length);
if (charsRead < 0) throw new NetconfException("Input Stream has been closed during reading.");
rpcReply.append(buffer, 0, charsRead);
if (in.ready()) {
Copy link
Contributor

@GregDThomas GregDThomas Jul 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is a bad idea. It looks to me it will end up in a tight loop burning CPU until timeout or the device sends some data.

Copy link
Author

@beserb beserb Jul 18, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right. I have investigated a bit further and it seems like having a timeout control using another thread is a valid solution. I have implemented it, now all tests pass and hang issue seems resolved. However, as I don't have much experience with threads I will investigate this solution a bit more and commit later on.

Do you have any other idea or any objection?

Copy link
Contributor

@GregDThomas GregDThomas Jul 18, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Without thinking about it too much, a CompletableFuture is probably what you want to be looking at, but there are no doubt other options.

Copy link
Contributor

@GregDThomas GregDThomas Jul 18, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Though there may well be a better way using NIO, but can’t remember the full context of this code off the top of my head , so unsure if it could easily be applied in this case)

int charsRead = in.read(buffer, 0, buffer.length);
rpcReply.append(buffer, 0, charsRead);
}
}

if (!timeoutNotExceeded)
Expand Down
74 changes: 32 additions & 42 deletions src/test/java/net/juniper/netconf/NetconfSessionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,24 +119,11 @@ public void GIVEN_createSession_WHEN_timeoutExceeded_THEN_throwSocketTimeoutExce
.hasMessage("Command timeout limit was exceeded: 1000");
}

@Test
public void GIVEN_createSession_WHEN_connectionClose_THEN_throwSocketTimeoutException() {
Thread thread = new Thread(() -> {
try {
outPipe.write(FAKE_RPC_REPLY.getBytes());
Thread.sleep(200);
outPipe.flush();
Thread.sleep(200);
outPipe.close();
} catch (IOException | InterruptedException e) {
log.error("error =", e);
}
});
thread.start();

assertThatThrownBy(() -> createNetconfSession(COMMAND_TIMEOUT))
.isInstanceOf(NetconfException.class)
.hasMessage("Input Stream has been closed during reading.");
@Test(timeout = 2000)
public void GIVEN_createSession_WHEN_noResponseTimeoutExceeded_THEN_throwSocketTimeoutException() {
assertThatThrownBy(() -> createNetconfSession(1000))
.isInstanceOf(SocketTimeoutException.class)
.hasMessage("Command timeout limit was exceeded: 1000");
}

@Test
Expand Down Expand Up @@ -287,35 +274,42 @@ private static void mockResponse(final InputStream is, final String message) thr
@Test
public void loadTextConfigurationWillSucceedIfResponseIsOk() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(true)
.build();
mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

netconfSession.loadTextConfiguration("some config", "some type");

verify(is, times(2)).read(any(), anyInt(), anyInt());
}

private void mockChannelInputStream(InputStream is, String replyXml, Channel mockChannel) throws IOException {
mockResponse(is, replyXml);
when(is.available())
.thenReturn((replyXml + NetconfConstants.DEVICE_PROMPT).length())
.thenReturn(0);
when(mockChannel.getInputStream())
.thenReturn(is);
}

@Test
public void loadTextConfigurationWillFailIfResponseIsNotOk() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(false)
.build();
mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

assertThrows(LoadException.class,
() -> netconfSession.loadTextConfiguration("some config", "some type"));
Expand All @@ -326,17 +320,16 @@ public void loadTextConfigurationWillFailIfResponseIsNotOk() throws Exception {
@Test
public void loadTextConfigurationWillFailIfResponseIsOkWithErrors() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(true)
.error(RpcError.builder().errorSeverity(RpcError.ErrorSeverity.ERROR).build())
.build();
mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

assertThrows(LoadException.class,
() -> netconfSession.loadTextConfiguration("some config", "some type"));
Expand All @@ -347,16 +340,15 @@ public void loadTextConfigurationWillFailIfResponseIsOkWithErrors() throws Excep
@Test
public void loadXmlConfigurationWillSucceedIfResponseIsOk() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(true)
.build();
mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

netconfSession.loadXMLConfiguration("some config", "merge");

Expand All @@ -366,17 +358,16 @@ public void loadXmlConfigurationWillSucceedIfResponseIsOk() throws Exception {
@Test
public void loadXmlConfigurationWillFailIfResponseIsNotOk() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(false)
.build();

mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

assertThrows(LoadException.class,
() -> netconfSession.loadXMLConfiguration("some config", "merge"));
Expand All @@ -387,17 +378,16 @@ public void loadXmlConfigurationWillFailIfResponseIsNotOk() throws Exception {
@Test
public void loadXmlConfigurationWillFailIfResponseIsOkWithErrors() throws Exception {

final String hello = "<hello/>";
final InputStream is = mock(InputStream.class);
when(mockChannel.getInputStream())
.thenReturn(is);
mockResponse(is, "<hello/>");
mockChannelInputStream(is, hello, mockChannel);
final NetconfSession netconfSession = createNetconfSession(100);

final RpcReply rpcReply = RpcReply.builder()
.ok(true)
.error(RpcError.builder().errorSeverity(RpcError.ErrorSeverity.ERROR).build())
.build();
mockResponse(is, rpcReply.getXml());
mockChannelInputStream(is, rpcReply.getXml(), mockChannel);

assertThrows(LoadException.class,
() -> netconfSession.loadXMLConfiguration("some config", "merge"));
Expand Down