[SSHD-449] SSH Exec channel with ClientChannel.Streaming.Async

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.14.0
• Fix Version/s: 2.0.0
• Flags: Important
• Assignee: Guillaume Nodet
• Reporter: Yarin Benado
• Created: 21/Apr/15 16:08
• Updated: 28/May/18 09:38
• Resolved: 28/May/18 09:38

Description

It looks like there is an issue with combining exec channel with ClientChannel.Streaming.Async.

You cannot use the channel.gerAsyncOut() to add a listener to it before the channel is opened, therefore open() call on the channel will fire the command to the server, and response might arrive before there are any listeners attached.
If a listener is not attached in-time (in the test below even 100 millis are suffice to cause it to miss the registration) the channel simply returns an empty output on the stream.

Here is a test added in ClientTest.java that reproduces the problem:

@Test
public void testExecAsyncClient() throws Exception {
    client.start();
    ClientSession session = client.connect("smx", "localhost", port).await().getSession();
    session.addPasswordIdentity("smx");
    session.auth().verify();
    final ByteArrayOutputStream baosOut = new ByteArrayOutputStream();
    final ByteArrayOutputStream baosErr = new ByteArrayOutputStream();

    final ChannelExec channel = session.createExecChannel("test");
    channel.setStreaming(ClientChannel.Streaming.Async);
    OpenFuture open = channel.open();

    
    Thread.sleep(100); // Removing this line will make the test succeed
    open.addListener(new SshFutureListener<OpenFuture>() {
        public void operationComplete(OpenFuture future) {
            channel.getAsyncOut().read(new Buffer())
                    .addListener(new SshFutureListener<IoReadFuture>() {
                        public void operationComplete(IoReadFuture future) {
                            try {
                                future.verify();
                                Buffer buffer = future.getBuffer();
                                baosOut.write(buffer.array(), buffer.rpos(), buffer.available());
                                buffer.rpos(buffer.rpos() + buffer.available());
                                buffer.compact();
                                channel.getAsyncOut().read(buffer).addListener(this);
                            } catch (IOException e) {
                                if (!channel.isClosing()) {
                                    e.printStackTrace();
                                    channel.close(true);
                                }
                            }
                        }
                    });
            channel.getAsyncErr().read(new Buffer())
                    .addListener(new SshFutureListener<IoReadFuture>() {
                        public void operationComplete(IoReadFuture future) {
                            try {
                                future.verify();
                                Buffer buffer = future.getBuffer();
                                baosErr.write(buffer.array(), buffer.rpos(), buffer.available());
                                buffer.rpos(buffer.rpos() + buffer.available());
                                buffer.compact();
                                channel.getAsyncErr().read(buffer).addListener(this);
                            } catch (IOException e) {
                                if (!channel.isClosing()) {
                                    e.printStackTrace();
                                    channel.close(true);
                                }
                            }
                        }
                    });
        }
    });

    channel.waitFor(ClientChannel.CLOSED, 0);

    assertFalse(baosErr.size() == 0);

    client.close(true);
}