Send EOF on channel Close (Fixes #143, #496, #553, #554)

Signed-off-by: Jeroen van Erp <[email protected]>

Author: hierynomus

src/main/java/net/schmizz/sshj/connection/channel/ChannelOutputStream.java

@@ -22,6 +22,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * {@link OutputStream} for channels. Buffers data upto the remote window's maximum packet size. Data can also be
@@ -36,7 +37,7 @@ public final class ChannelOutputStream extends OutputStream implements ErrorNoti
     private final DataBuffer buffer = new DataBuffer();
     private final byte[] b = new byte[1];
 
-    private boolean closed;
+    private AtomicBoolean closed;
     private SSHException error;
 
     private final class DataBuffer {
@@ -122,6 +123,7 @@ public ChannelOutputStream(Channel chan, Transport trans, Window.Remote win) {
         this.chan = chan;
         this.trans = trans;
         this.win = win;
+        this.closed = new AtomicBoolean(false);
     }
 
     @Override
@@ -151,24 +153,21 @@ public synchronized void notifyError(SSHException error) {
 
     private void checkClose() throws SSHException {
         // Check whether either the Stream is closed, or the underlying channel is closed
-        if (closed || !chan.isOpen()) {
-            if (error != null)
+        if (closed.get() || !chan.isOpen()) {
+            if (error != null) {
                 throw error;
-            else
+            } else {
                 throw new ConnectionException("Stream closed");
+            }
         }
     }
 
     @Override
     public synchronized void close() throws IOException {
         // Not closed yet, and underlying channel is open to flush the data to.
-        if (!closed && chan.isOpen()) {
-            try {
-                buffer.flush(false);
-//                trans.write(new SSHPacket(Message.CHANNEL_EOF).putUInt32(chan.getRecipient()));
-            } finally {
-                closed = true;
-            }
+        if (!closed.getAndSet(true) && chan.isOpen()) {
+            buffer.flush(false);
+            trans.write(new SSHPacket(Message.CHANNEL_EOF).putUInt32(chan.getRecipient()));
         }
     }
 

src/test/java/net/schmizz/sshj/LoadsOfConnects.java

@@ -20,6 +20,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import net.schmizz.sshj.common.IOUtils;
+import net.schmizz.sshj.connection.channel.direct.Session;
+
 import static org.junit.Assert.fail;
 
 public class LoadsOfConnects {
@@ -31,15 +34,23 @@ public class LoadsOfConnects {
     @Test
     public void loadsOfConnects() {
         try {
+            fixture.start();
             for (int i = 0; i < 1000; i++) {
                 log.info("Try " + i);
-                fixture.start();
-                fixture.setupConnectedDefaultClient();
+                SSHClient client = fixture.setupConnectedDefaultClient();
+                client.authPassword("test", "test");
+                Session s = client.startSession();
+                Session.Command c = s.exec("ls");
+                IOUtils.readFully(c.getErrorStream());
+                IOUtils.readFully(c.getInputStream());
+                c.close();
+                s.close();
                 fixture.stopClient();
-                fixture.stopServer();
             }
         } catch (Exception e) {
             fail(e.getMessage());
+        } finally {
+            fixture.stopServer();
         }
 
     }