You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by lg...@apache.org on 2015/08/25 10:26:04 UTC
mina-sshd git commit: [SSHD-557] MINA SSHD deadlocks upon multiple
concurrent HTTP downloads over locally-forwarded ports
Repository: mina-sshd
Updated Branches:
refs/heads/master 40195ab46 -> bda6cb099
[SSHD-557] MINA SSHD deadlocks upon multiple concurrent HTTP downloads over locally-forwarded ports
Project: http://git-wip-us.apache.org/repos/asf/mina-sshd/repo
Commit: http://git-wip-us.apache.org/repos/asf/mina-sshd/commit/bda6cb09
Tree: http://git-wip-us.apache.org/repos/asf/mina-sshd/tree/bda6cb09
Diff: http://git-wip-us.apache.org/repos/asf/mina-sshd/diff/bda6cb09
Branch: refs/heads/master
Commit: bda6cb099107900124479082390ae59613484127
Parents: 40195ab
Author: Lyor Goldstein <lg...@vmware.com>
Authored: Tue Aug 25 11:25:53 2015 +0300
Committer: Lyor Goldstein <lg...@vmware.com>
Committed: Tue Aug 25 11:25:53 2015 +0300
----------------------------------------------------------------------
.../client/channel/AbstractClientChannel.java | 40 +-
.../sshd/client/session/ClientSessionImpl.java | 51 ++-
.../sshd/common/channel/AbstractChannel.java | 36 +-
.../channel/ChannelAsyncOutputStream.java | 89 +++--
.../common/channel/ChannelPipedInputStream.java | 17 +-
.../sshd/common/channel/ChannelPipedSink.java | 2 +-
.../org/apache/sshd/common/channel/Window.java | 223 ++++++++---
.../org/apache/sshd/common/util/io/IoUtils.java | 17 +-
.../server/channel/AbstractServerChannel.java | 10 +-
.../sshd/server/channel/ChannelSession.java | 3 +-
.../test/java/org/apache/sshd/WindowTest.java | 345 ------------------
.../channel/ChannelPipedInputStreamTest.java | 17 +-
.../apache/sshd/common/channel/WindowTest.java | 365 +++++++++++++++++++
.../java/org/apache/sshd/server/ServerTest.java | 3 +-
.../sshd/server/channel/ChannelSessionTest.java | 19 +-
.../java/org/apache/sshd/util/BogusChannel.java | 3 +
sshd-core/src/test/resources/log4j.properties | 2 +-
17 files changed, 748 insertions(+), 494 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/client/channel/AbstractClientChannel.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/client/channel/AbstractClientChannel.java b/sshd-core/src/main/java/org/apache/sshd/client/channel/AbstractClientChannel.java
index 8eccb4e..1983a45 100644
--- a/sshd-core/src/main/java/org/apache/sshd/client/channel/AbstractClientChannel.java
+++ b/sshd-core/src/main/java/org/apache/sshd/client/channel/AbstractClientChannel.java
@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.sshd.client.future.DefaultOpenFuture;
import org.apache.sshd.client.future.OpenFuture;
import org.apache.sshd.common.Closeable;
+import org.apache.sshd.common.FactoryManager;
import org.apache.sshd.common.SshConstants;
import org.apache.sshd.common.SshException;
import org.apache.sshd.common.channel.AbstractChannel;
@@ -36,7 +37,9 @@ import org.apache.sshd.common.channel.ChannelListener;
import org.apache.sshd.common.channel.ChannelRequestHandler;
import org.apache.sshd.common.io.IoInputStream;
import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.session.Session;
import org.apache.sshd.common.util.GenericUtils;
+import org.apache.sshd.common.util.ValidateUtils;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
import org.apache.sshd.common.util.io.IoUtils;
@@ -70,6 +73,7 @@ public abstract class AbstractClientChannel extends AbstractChannel implements C
protected OpenFuture openFuture;
protected AbstractClientChannel(String type) {
+ super(true);
this.type = type;
this.streaming = Streaming.Sync;
addRequestHandler(new ExitStatusChannelRequestHandler());
@@ -194,8 +198,8 @@ public abstract class AbstractClientChannel extends AbstractChannel implements C
}
if ((cond & mask) != 0) {
if (log.isTraceEnabled()) {
- log.trace("WaitFor call returning on channel {}, mask={}, cond={}",
- this, mask, cond);
+ log.trace("WaitFor call returning on channel {}, mask=0x{}, cond=0x{}",
+ this, Integer.toHexString(mask), Integer.toHexString(cond));
}
return cond;
}
@@ -205,23 +209,38 @@ public abstract class AbstractClientChannel extends AbstractChannel implements C
} else {
timeout = t - System.currentTimeMillis();
if (timeout <= 0) {
+ if (log.isTraceEnabled()) {
+ log.trace("WaitFor call timeout on channel {}, mask=0x{}", this, Integer.toHexString(mask));
+ }
cond |= ClientChannel.TIMEOUT;
return cond;
}
}
}
+
+ if (log.isTraceEnabled()) {
+ log.trace("Waiting {} millis for lock on channel {}, mask={}, cond={}", timeout, this, mask, cond);
+ }
+
+ long nanoStart = System.nanoTime();
try {
- if (log.isTraceEnabled()) {
- log.trace("Waiting for lock on channel {}, mask={}, cond={}", this, mask, cond);
- }
- if (timeout > 0) {
+ if (timeout > 0L) {
lock.wait(timeout);
} else {
lock.wait();
}
- log.trace("Lock notified on channel {}", this);
+
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ if (log.isTraceEnabled()) {
+ log.trace("Lock notified on channel {} after {} nanos", this, nanoDuration);
+ }
} catch (InterruptedException e) {
- // Ignore
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ if (log.isTraceEnabled()) {
+ log.trace("waitFor({}) mask={} - ignoring interrupted exception after {} nanos", this, Integer.toHexString(mask), nanoDuration);
+ }
}
}
}
@@ -251,7 +270,10 @@ public abstract class AbstractClientChannel extends AbstractChannel implements C
@Override
public void handleOpenSuccess(int recipient, int rwSize, int packetSize, Buffer buffer) {
this.recipient = recipient;
- this.remoteWindow.init(rwSize, packetSize, session.getFactoryManager().getProperties());
+
+ Session s = getSession();
+ FactoryManager manager = ValidateUtils.checkNotNull(s.getFactoryManager(), "No factory manager");
+ this.remoteWindow.init(rwSize, packetSize, manager.getProperties());
ChannelListener listener = getChannelListenerProxy();
try {
doOpen();
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java b/sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java
index 2b112ea..d44a855 100644
--- a/sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java
+++ b/sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java
@@ -56,6 +56,7 @@ import org.apache.sshd.common.SshdSocketAddress;
import org.apache.sshd.common.cipher.BuiltinCiphers;
import org.apache.sshd.common.cipher.CipherNone;
import org.apache.sshd.common.config.keys.KeyUtils;
+import org.apache.sshd.common.forward.TcpipForwarder;
import org.apache.sshd.common.future.DefaultSshFuture;
import org.apache.sshd.common.future.SshFuture;
import org.apache.sshd.common.io.IoSession;
@@ -453,32 +454,37 @@ public class ClientSessionImpl extends AbstractSession implements ClientSession
@Override
public SshdSocketAddress startLocalPortForwarding(SshdSocketAddress local, SshdSocketAddress remote) throws IOException {
- return getConnectionService().getTcpipForwarder().startLocalPortForwarding(local, remote);
+ return getTcpipForwarder().startLocalPortForwarding(local, remote);
}
@Override
public void stopLocalPortForwarding(SshdSocketAddress local) throws IOException {
- getConnectionService().getTcpipForwarder().stopLocalPortForwarding(local);
+ getTcpipForwarder().stopLocalPortForwarding(local);
}
@Override
public SshdSocketAddress startRemotePortForwarding(SshdSocketAddress remote, SshdSocketAddress local) throws IOException {
- return getConnectionService().getTcpipForwarder().startRemotePortForwarding(remote, local);
+ return getTcpipForwarder().startRemotePortForwarding(remote, local);
}
@Override
public void stopRemotePortForwarding(SshdSocketAddress remote) throws IOException {
- getConnectionService().getTcpipForwarder().stopRemotePortForwarding(remote);
+ getTcpipForwarder().stopRemotePortForwarding(remote);
}
@Override
public SshdSocketAddress startDynamicPortForwarding(SshdSocketAddress local) throws IOException {
- return getConnectionService().getTcpipForwarder().startDynamicPortForwarding(local);
+ return getTcpipForwarder().startDynamicPortForwarding(local);
}
@Override
public void stopDynamicPortForwarding(SshdSocketAddress local) throws IOException {
- getConnectionService().getTcpipForwarder().stopDynamicPortForwarding(local);
+ getTcpipForwarder().stopDynamicPortForwarding(local);
+ }
+
+ protected TcpipForwarder getTcpipForwarder() {
+ ConnectionService service = ValidateUtils.checkNotNull(getConnectionService(), "No connection service");
+ return ValidateUtils.checkNotNull(service.getTcpipForwarder(), "No forwarder");
}
@Override
@@ -490,7 +496,7 @@ public class ClientSessionImpl extends AbstractSession implements ClientSession
@Override
public int waitFor(int mask, long timeout) {
- long t = 0;
+ long t = 0L;
synchronized (lock) {
for (;;) {
int cond = 0;
@@ -504,27 +510,50 @@ public class ClientSessionImpl extends AbstractSession implements ClientSession
cond |= WAIT_AUTH;
}
if ((cond & mask) != 0) {
+ if (log.isTraceEnabled()) {
+ log.trace("WaitFor call returning on session {}, mask=0x{}, cond=0x{}",
+ this, Integer.toHexString(mask), Integer.toHexString(cond));
+ }
return cond;
}
- if (timeout > 0) {
- if (t == 0) {
+ if (timeout > 0L) {
+ if (t == 0L) {
t = System.currentTimeMillis() + timeout;
} else {
timeout = t - System.currentTimeMillis();
- if (timeout <= 0) {
+ if (timeout <= 0L) {
+ if (log.isTraceEnabled()) {
+ log.trace("WaitFor call timeout on session {}, mask=0x{}", this, Integer.toHexString(mask));
+ }
cond |= TIMEOUT;
return cond;
}
}
}
+
+ if (log.isTraceEnabled()) {
+ log.trace("Waiting {} millis for lock on session {}, mask={}, cond={}", timeout, this, mask, cond);
+ }
+
+ long nanoStart = System.nanoTime();
try {
if (timeout > 0) {
lock.wait(timeout);
} else {
lock.wait();
}
+
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ if (log.isTraceEnabled()) {
+ log.trace("Lock notified on session {} after {} nanos", this, nanoDuration);
+ }
} catch (InterruptedException e) {
- // Ignore
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ if (log.isTraceEnabled()) {
+ log.trace("waitFor({}) mask={} - ignoring interrupted exception after {} nanos", this, Integer.toHexString(mask), nanoDuration);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/channel/AbstractChannel.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/AbstractChannel.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/AbstractChannel.java
index 04ed9ad..62bd7c1 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/AbstractChannel.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/AbstractChannel.java
@@ -46,6 +46,7 @@ import org.apache.sshd.common.util.ValidateUtils;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.common.util.buffer.BufferUtils;
import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.common.util.io.IoUtils;
import org.apache.sshd.common.util.threads.ExecutorServiceConfigurer;
/**
@@ -60,7 +61,7 @@ public abstract class AbstractChannel
public static final int DEFAULT_WINDOW_SIZE = 0x200000;
public static final int DEFAULT_PACKET_SIZE = 0x8000;
- public static final long DEFAULT_CHANNEL_CLOSE_TIMEOUT = 5000;
+ public static final long DEFAULT_CHANNEL_CLOSE_TIMEOUT = TimeUnit.SECONDS.toMillis(5L);
/**
* Default growth factor function used to resize response buffers
@@ -73,8 +74,8 @@ public abstract class AbstractChannel
protected ExecutorService executor;
protected boolean shutdownExecutor;
- protected final Window localWindow = new Window(this, null, getClass().getName().contains(".client."), true);
- protected final Window remoteWindow = new Window(this, null, getClass().getName().contains(".client."), false);
+ protected final Window localWindow;
+ protected final Window remoteWindow;
protected ConnectionService service;
protected Session session;
protected int id;
@@ -89,12 +90,14 @@ public abstract class AbstractChannel
protected final Collection<ChannelListener> channelListeners = new CopyOnWriteArraySet<>();
protected final ChannelListener channelListenerProxy;
- protected AbstractChannel() {
- this("");
+ protected AbstractChannel(boolean client) {
+ this("", client);
}
- protected AbstractChannel(String discriminator) {
+ protected AbstractChannel(String discriminator, boolean client) {
super(discriminator);
+ localWindow = new Window(this, null, client, true);
+ remoteWindow = new Window(this, null, client, false);
channelListenerProxy = EventListenerUtils.proxyWrapper(ChannelListener.class, getClass().getClassLoader(), channelListeners);
}
@@ -353,6 +356,22 @@ public abstract class AbstractChannel
// clear the listeners since we are closing the channel (quicker GC)
this.channelListeners.clear();
}
+
+ IOException err = IoUtils.closeQuietly(localWindow, remoteWindow);
+ if (err != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Failed (" + err.getClass().getSimpleName() + ") to close window(s) of " + this + ": " + err.getMessage());
+ }
+
+ if (log.isTraceEnabled()) {
+ Throwable[] suppressed = err.getSuppressed();
+ if (GenericUtils.length(suppressed) > 0) {
+ for (Throwable t : suppressed) {
+ log.trace("Suppressed " + t.getClass().getSimpleName() + ") while close window(s) of " + this + ": " + t.getMessage());
+ }
+ }
+ }
+ }
}
@Override
@@ -360,6 +379,7 @@ public abstract class AbstractChannel
if (service != null) {
service.unregisterChannel(AbstractChannel.this);
}
+
super.doCloseImmediately();
}
@@ -454,7 +474,7 @@ public abstract class AbstractChannel
protected void sendWindowAdjust(int len) throws IOException {
if (log.isDebugEnabled()) {
- log.debug("Send SSH_MSG_CHANNEL_WINDOW_ADJUST on channel {}", Integer.valueOf(id));
+ log.debug("Send SSH_MSG_CHANNEL_WINDOW_ADJUST (len={}) on channel {}", len, this);
}
Session s = getSession();
Buffer buffer = s.createBuffer(SshConstants.SSH_MSG_CHANNEL_WINDOW_ADJUST);
@@ -465,6 +485,6 @@ public abstract class AbstractChannel
@Override
public String toString() {
- return getClass().getSimpleName() + "[id=" + getId() + ", recipient=" + getRecipient() + "]";
+ return getClass().getSimpleName() + "[id=" + getId() + ", recipient=" + getRecipient() + "]" + "-" + getSession();
}
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelAsyncOutputStream.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelAsyncOutputStream.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelAsyncOutputStream.java
index e06cc03..dd25023 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelAsyncOutputStream.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelAsyncOutputStream.java
@@ -27,6 +27,7 @@ import org.apache.sshd.common.future.SshFutureListener;
import org.apache.sshd.common.io.IoOutputStream;
import org.apache.sshd.common.io.IoWriteFuture;
import org.apache.sshd.common.io.WritePendingException;
+import org.apache.sshd.common.session.Session;
import org.apache.sshd.common.util.CloseableUtils;
import org.apache.sshd.common.util.buffer.Buffer;
@@ -66,47 +67,61 @@ public class ChannelAsyncOutputStream extends CloseableUtils.AbstractCloseable i
protected synchronized void doWriteIfPossible(boolean resume) {
final IoWriteFutureImpl future = pendingWrite.get();
- if (future != null) {
- final Buffer buffer = future.getBuffer();
- final int total = buffer.available();
- if (total > 0) {
- final int length = Math.min(Math.min(channel.getRemoteWindow().getSize(), total), channel.getRemoteWindow().getPacketSize());
- if (length > 0) {
- if (resume) {
- log.debug("Resuming write due to more space available in the remote window");
- }
- Buffer buf = channel.getSession().createBuffer(cmd, length + 12);
- buf.putInt(channel.getRecipient());
- if (cmd == SshConstants.SSH_MSG_CHANNEL_EXTENDED_DATA) {
- buf.putInt(1);
+ if (future == null) {
+ if (log.isTraceEnabled()) {
+ log.trace("doWriteIfPossible({})[resume={}] no pending write future", this, resume);
+ }
+ return;
+ }
+
+ final Buffer buffer = future.getBuffer();
+ final int total = buffer.available();
+ if (total > 0) {
+ Window remoteWindow = channel.getRemoteWindow();
+ final int length = Math.min(Math.min(remoteWindow.getSize(), total), remoteWindow.getPacketSize());
+ if (log.isTraceEnabled()) {
+ log.trace("doWriteIfPossible({})[resume={}] attempting to write {} out of {}", this, resume, length, total);
+ }
+
+ if (length > 0) {
+ if (resume) {
+ if (log.isDebugEnabled()) {
+ log.debug("Resuming {} write due to more space ({}) available in the remote window", this, length);
}
- buf.putInt(length);
- buf.putRawBytes(buffer.array(), buffer.rpos(), length);
- buffer.rpos(buffer.rpos() + length);
- channel.getRemoteWindow().consume(length);
- try {
- channel.getSession().writePacket(buf).addListener(new SshFutureListener<org.apache.sshd.common.io.IoWriteFuture>() {
- @SuppressWarnings("synthetic-access")
- @Override
- public void operationComplete(org.apache.sshd.common.io.IoWriteFuture f) {
- if (total > length) {
- doWriteIfPossible(false);
- } else {
- pendingWrite.compareAndSet(future, null);
- future.setValue(Boolean.TRUE);
- }
+ }
+
+ Session s = channel.getSession();
+ Buffer buf = s.createBuffer(cmd, length + 12);
+ buf.putInt(channel.getRecipient());
+ if (cmd == SshConstants.SSH_MSG_CHANNEL_EXTENDED_DATA) {
+ buf.putInt(1);
+ }
+ buf.putInt(length);
+ buf.putRawBytes(buffer.array(), buffer.rpos(), length);
+ buffer.rpos(buffer.rpos() + length);
+ remoteWindow.consume(length);
+ try {
+ s.writePacket(buf).addListener(new SshFutureListener<IoWriteFuture>() {
+ @SuppressWarnings("synthetic-access")
+ @Override
+ public void operationComplete(IoWriteFuture f) {
+ if (total > length) {
+ doWriteIfPossible(false);
+ } else {
+ pendingWrite.compareAndSet(future, null);
+ future.setValue(Boolean.TRUE);
}
- });
- } catch (IOException e) {
- future.setValue(e);
- }
- } else if (!resume) {
- log.debug("Delaying write until space is available in the remote window");
+ }
+ });
+ } catch (IOException e) {
+ future.setValue(e);
}
- } else {
- pendingWrite.compareAndSet(future, null);
- future.setValue(Boolean.TRUE);
+ } else if (!resume) {
+ log.debug("Delaying write to {} until space is available in the remote window", this);
}
+ } else {
+ pendingWrite.compareAndSet(future, null);
+ future.setValue(Boolean.TRUE);
}
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedInputStream.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedInputStream.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedInputStream.java
index 3df0526..a01b955 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedInputStream.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedInputStream.java
@@ -45,7 +45,7 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
private final Window localWindow;
private final Buffer buffer = new ByteArrayBuffer();
private final byte[] b = new byte[1];
- private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final AtomicBoolean open = new AtomicBoolean(true);
private final AtomicBoolean eofSent = new AtomicBoolean(false);
private final Lock lock = new ReentrantLock();
@@ -65,6 +65,11 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
this.timeout = FactoryManagerUtils.getLongProperty(localWindow.getProperties(), FactoryManager.WINDOW_TIMEOUT, DEFAULT_TIMEOUT);
}
+ @Override
+ public boolean isOpen() {
+ return open.get();
+ }
+
public void setTimeout(long timeout) {
this.timeout = timeout;
}
@@ -104,7 +109,9 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
lock.lock();
try {
for (int index = 0;; index++) {
- if ((closed.get() && writerClosed.get() && eofSent.get()) || (closed.get() && (!writerClosed.get()))) {
+ boolean openState = isOpen();
+ boolean writerClosedState = writerClosed.get();
+ if (((!openState) && writerClosedState && eofSent.get()) || ((!openState) && (!writerClosedState))) {
throw new IOException("Pipe closed after " + index + " cycles");
}
if (buffer.available() > 0) {
@@ -116,7 +123,7 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
}
try {
- if (timeout > 0) {
+ if (timeout > 0L) {
long remaining = timeout - (System.currentTimeMillis() - startTime);
if (remaining <= 0) {
throw new SocketException("Timeout (" + timeout + ") exceeded after " + index + " cycles");
@@ -160,7 +167,7 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
try {
dataAvailable.signalAll();
} finally {
- closed.set(true);
+ open.set(false);
lock.unlock();
}
}
@@ -169,7 +176,7 @@ public class ChannelPipedInputStream extends InputStream implements ChannelPiped
public void receive(byte[] bytes, int off, int len) throws IOException {
lock.lock();
try {
- if (writerClosed.get() || closed.get()) {
+ if (writerClosed.get() || (!isOpen())) {
throw new IOException("Pipe closed");
}
buffer.putRawBytes(bytes, off, len);
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedSink.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedSink.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedSink.java
index 8647596..c1b8987 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedSink.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/ChannelPipedSink.java
@@ -24,7 +24,7 @@ import java.io.IOException;
/**
* @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
*/
-public interface ChannelPipedSink {
+public interface ChannelPipedSink extends java.nio.channels.Channel {
/**
* @param bytes Bytes to be sent to the sink
* @param off Offset in buffer
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
index 9f84a88..15e7b8a 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
@@ -19,8 +19,11 @@
package org.apache.sshd.common.channel;
import java.io.IOException;
+import java.io.StreamCorruptedException;
import java.util.Collections;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.sshd.common.FactoryManager;
import org.apache.sshd.common.FactoryManagerUtils;
@@ -37,22 +40,23 @@ import org.apache.sshd.common.util.logging.AbstractLoggingBean;
*
* @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
*/
-public class Window extends AbstractLoggingBean {
+public class Window extends AbstractLoggingBean implements java.nio.channels.Channel {
+ private final AtomicInteger waitingCount = new AtomicInteger(0);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final AtomicBoolean initialized = new AtomicBoolean(false);
private final AbstractChannel channel;
private final Object lock;
- private final String name;
+ private final String suffix;
private int size;
private int maxSize;
private int packetSize;
- private boolean waiting;
- private boolean closed;
private Map<String, ?> props = Collections.<String, Object>emptyMap();
public Window(AbstractChannel channel, Object lock, boolean client, boolean local) {
this.channel = ValidateUtils.checkNotNull(channel, "No channel provided");
this.lock = (lock != null) ? lock : this;
- this.name = String.valueOf(channel) + ": " + (client ? "client" : "server") + " " + (local ? "local " : "remote") + " window";
+ this.suffix = ": " + (client ? "client" : "server") + " " + (local ? "local " : "remote") + " window";
}
public Map<String, ?> getProperties() {
@@ -83,123 +87,226 @@ public class Window extends AbstractLoggingBean {
public void init(Map<String, ?> props) {
init(FactoryManagerUtils.getIntProperty(props, FactoryManager.WINDOW_SIZE, AbstractChannel.DEFAULT_WINDOW_SIZE),
- FactoryManagerUtils.getIntProperty(props, FactoryManager.MAX_PACKET_SIZE, AbstractChannel.DEFAULT_PACKET_SIZE),
- props);
+ FactoryManagerUtils.getIntProperty(props, FactoryManager.MAX_PACKET_SIZE, AbstractChannel.DEFAULT_PACKET_SIZE),
+ props);
}
public void init(int size, int packetSize, Map<String, ?> props) {
+ ValidateUtils.checkTrue(size >= 0, "Illegal initial size: %d", size);
+ ValidateUtils.checkTrue(packetSize > 0, "Illegal packet size: %d", packetSize);
+
synchronized (lock) {
this.size = size;
this.maxSize = size;
this.packetSize = packetSize;
- this.props = props;
+ this.props = (props == null) ? Collections.<String, Object>emptyMap() : props;
lock.notifyAll();
}
+
+ initialized.set(true);
+
+ if (log.isDebugEnabled()) {
+ log.debug("init({}) size={}, max.={}, packet={}", this, getSize(), getMaxSize(), getPacketSize());
+ }
}
public void expand(int window) {
+ ValidateUtils.checkTrue(window >= 0, "Negative window size: %d", window);
+ checkInitialized("expand");
+
+ long expandedSize;
synchronized (lock) {
- size += window;
- if (log.isDebugEnabled()) {
- log.debug("Increase " + name + " by " + window + " up to " + size);
+ /*
+ * See RFC-4254 section 5.2:
+ *
+ * "Implementations MUST correctly handle window sizes
+ * of up to 2^32 - 1 bytes. The window MUST NOT be increased above
+ * 2^32 - 1 bytes.
+ */
+ expandedSize = size + window;
+ if (expandedSize > Integer.MAX_VALUE) {
+ size = Integer.MAX_VALUE;
+ } else {
+ size = (int) expandedSize;
}
lock.notifyAll();
}
+
+ if (expandedSize > Integer.MAX_VALUE) {
+ log.warn("expand({}) window={} - truncated expanded size ({}) to {}", this, window, expandedSize, Integer.MAX_VALUE);
+ } else if (log.isDebugEnabled()) {
+ log.debug("Increase {} by {} up to {}", this, window, expandedSize);
+ }
}
public void consume(int len) {
+ ValidateUtils.checkTrue(len >= 0, "Negative consumption length: %d", len);
+ checkInitialized("consume");
+
+ int remainLen;
synchronized (lock) {
- //assert size > len;
- size -= len;
- if (log.isTraceEnabled()) {
- log.trace("Consume " + name + " by " + len + " down to " + size);
+ remainLen = size - len;
+ if (remainLen >= 0) {
+ size = remainLen;
}
}
+
+ if (remainLen < 0) {
+ throw new IllegalStateException("consume(" + this + ") required length (" + len + ") above available: " + (remainLen + len));
+ }
+
+ if (log.isTraceEnabled()) {
+ log.trace("Consume {} by {} down to {}", this, len, remainLen);
+ }
}
public void consumeAndCheck(int len) throws IOException {
synchronized (lock) {
- //assert size > len;
- size -= len;
- if (log.isTraceEnabled()) {
- log.trace("Consume " + name + " by " + len + " down to " + size);
+ try {
+ consume(len);
+ check(maxSize);
+ } catch (RuntimeException e) {
+ throw new StreamCorruptedException("consumeAndCheck(" + this + ")"
+ + " failed " + e.getClass().getSimpleName() + ")"
+ + " to consume " + len + " bytes"
+ + ": " + e.getMessage());
}
- check(maxSize);
}
}
public void check(int maxFree) throws IOException {
+ ValidateUtils.checkTrue(maxFree >= 0, "Negative check size: %d", maxFree);
+ checkInitialized("check");
+
+ int adjustSize = -1;
synchronized (lock) {
- if (size < maxFree / 2) {
- if (log.isDebugEnabled()) {
- log.debug("Increase " + name + " by " + (maxFree - size) + " up to " + maxFree);
- }
- channel.sendWindowAdjust(maxFree - size);
+ // TODO make the adjust factor configurable via FactoryManager property
+ if (size < (maxFree / 2)) {
+ adjustSize = maxFree - size;
+ channel.sendWindowAdjust(adjustSize);
size = maxFree;
}
}
+
+ if (adjustSize >= 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("Increase {} by {} up to {}", this, adjustSize, maxFree);
+ }
+ }
}
public void waitAndConsume(int len) throws InterruptedException, WindowClosedException {
+ ValidateUtils.checkTrue(len >= 0, "Negative wait consume length: %d", len);
+ checkInitialized("waitAndConsume");
+
synchronized (lock) {
- while (size < len && !closed) {
+ while ((size < len) && isOpen()) {
+ int waiters = waitingCount.incrementAndGet();
if (log.isDebugEnabled()) {
- log.debug("Waiting for {} bytes on {}", Integer.valueOf(len), name);
+ log.debug("waitAndConsume({}) - requested={}, available={}, waiters={}", this, len, size, waiters);
}
- waiting = true;
- lock.wait();
- }
- if (waiting) {
- if (closed) {
- log.debug("Window {} has been closed", name);
- } else {
- log.debug("Space available for {}", name);
+
+ long nanoStart = System.nanoTime();
+ try {
+ lock.wait();
+ } finally {
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ waiters = waitingCount.decrementAndGet();
+ if (log.isTraceEnabled()) {
+ log.debug("waitAndConsume({}) - requested={}, available={}, waiters={} - ended after {} nanos",
+ this, len, size, waiters, nanoDuration);
+ }
}
- waiting = false;
}
- if (closed) {
- throw new WindowClosedException(name);
+
+ if (!isOpen()) {
+ throw new WindowClosedException(toString());
}
- size -= len;
- if (log.isTraceEnabled()) {
- log.trace("Consume " + name + " by " + len + " down to " + size);
+
+ if (log.isDebugEnabled()) {
+ log.debug("waitAndConsume({}) - requested={}, available={}", this, len, size);
}
+
+ consume(len);
}
}
+ /**
+ * Waits (forever) until some data becomes available
+ *
+ * @return Amount of available data - always positive
+ * @throws InterruptedException If interrupted while waiting
+ * @throws WindowClosedException If window closed while waiting
+ */
public int waitForSpace() throws InterruptedException, WindowClosedException {
+ checkInitialized("waitForSpace");
+
synchronized (lock) {
- while (size == 0 && !closed) {
- log.debug("Waiting for some space on {}", name);
- waiting = true;
- lock.wait();
- }
- if (waiting) {
- if (closed) {
- log.debug("Window {} has been closed", name);
- } else {
- log.debug("Space available for {}", name);
+ while ((size == 0) && isOpen()) {
+ int waiters = waitingCount.incrementAndGet();
+ if (log.isDebugEnabled()) {
+ log.debug("waitForSpace({}) - waiters={}", this, waiters);
}
- waiting = false;
+
+ long nanoStart = System.nanoTime();
+ try {
+ lock.wait();
+ } finally {
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ waiters = waitingCount.decrementAndGet();
+ if (log.isTraceEnabled()) {
+ log.debug("waitForSpace({}) - waiters={} - ended after {} nanos", this, waiters, nanoDuration);
+ }
+ }
+ }
+
+ if (!isOpen()) {
+ throw new WindowClosedException(toString());
}
- if (closed) {
- throw new WindowClosedException(name);
+
+ if (log.isDebugEnabled()) {
+ log.debug("waitForSpace({}) available: {}", this, size);
}
return size;
}
}
- public void notifyClosed() {
+ protected void checkInitialized(String location) {
+ if (!initialized.get()) {
+ throw new IllegalStateException(location + " - window not initialized: " + this);
+ }
+ }
+
+ @Override
+ public boolean isOpen() {
+ return !closed.get();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (isOpen()) {
+ closed.set(true);
+ log.debug("Closing {}", this);
+ }
+
+ // just in case someone is still waiting
+ int waiters;
synchronized (lock) {
- closed = true;
- if (waiting) {
+ waiters = waitingCount.get();
+ if (waiters > 0) {
lock.notifyAll();
}
}
+
+ if (log.isDebugEnabled()) {
+ log.debug("close({}) waiters={}", this, waiters);
+ }
}
@Override
public String toString() {
- return name;
+ return String.valueOf(channel) + suffix;
}
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/common/util/io/IoUtils.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/util/io/IoUtils.java b/sshd-core/src/main/java/org/apache/sshd/common/util/io/IoUtils.java
index 6bcb627..df9a0a3 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/util/io/IoUtils.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/util/io/IoUtils.java
@@ -88,16 +88,29 @@ public final class IoUtils {
return nread;
}
- public static void closeQuietly(Closeable... closeables) {
+ /**
+ * Closes a bunch of resources suppressing any {@link IOException}s their
+ * {@link Closeable#close()} method may have thrown
+ *
+ * @param closeables The {@link Closeable}s to close
+ * @return The <U>first</U> {@link IOException} that occurred during closing
+ * of a resource - if more than one exception occurred, they are added as
+ * suppressed exceptions to the first one
+ * @see Throwable#getSuppressed()
+ */
+ public static IOException closeQuietly(Closeable... closeables) {
+ IOException err = null;
for (Closeable c : closeables) {
try {
if (c != null) {
c.close();
}
} catch (IOException e) {
- // Ignore
+ GenericUtils.accumulateException(err, e);
}
}
+
+ return err;
}
/**
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/server/channel/AbstractServerChannel.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/server/channel/AbstractServerChannel.java b/sshd-core/src/main/java/org/apache/sshd/server/channel/AbstractServerChannel.java
index 8161e67..c61cec1 100644
--- a/sshd-core/src/main/java/org/apache/sshd/server/channel/AbstractServerChannel.java
+++ b/sshd-core/src/main/java/org/apache/sshd/server/channel/AbstractServerChannel.java
@@ -22,10 +22,13 @@ import java.io.IOException;
import org.apache.sshd.client.future.DefaultOpenFuture;
import org.apache.sshd.client.future.OpenFuture;
+import org.apache.sshd.common.FactoryManager;
import org.apache.sshd.common.SshConstants;
import org.apache.sshd.common.channel.AbstractChannel;
import org.apache.sshd.common.channel.ChannelListener;
+import org.apache.sshd.common.session.Session;
import org.apache.sshd.common.util.GenericUtils;
+import org.apache.sshd.common.util.ValidateUtils;
import org.apache.sshd.common.util.buffer.Buffer;
/**
@@ -42,13 +45,16 @@ public abstract class AbstractServerChannel extends AbstractChannel implements S
}
protected AbstractServerChannel(String discriminator) {
- super(discriminator);
+ super(discriminator, false);
}
@Override
public OpenFuture open(int recipient, int rwSize, int packetSize, Buffer buffer) {
this.recipient = recipient;
- this.remoteWindow.init(rwSize, packetSize, session.getFactoryManager().getProperties());
+
+ Session s = getSession();
+ FactoryManager manager = ValidateUtils.checkNotNull(s.getFactoryManager(), "No factory manager");
+ this.remoteWindow.init(rwSize, packetSize, manager.getProperties());
configureWindow();
return doInit(buffer);
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/main/java/org/apache/sshd/server/channel/ChannelSession.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/main/java/org/apache/sshd/server/channel/ChannelSession.java b/sshd-core/src/main/java/org/apache/sshd/server/channel/ChannelSession.java
index a4a020d..7041cbc 100644
--- a/sshd-core/src/main/java/org/apache/sshd/server/channel/ChannelSession.java
+++ b/sshd-core/src/main/java/org/apache/sshd/server/channel/ChannelSession.java
@@ -264,8 +264,7 @@ public class ChannelSession extends AbstractServerChannel {
command.destroy();
command = null;
}
- remoteWindow.notifyClosed();
- IoUtils.closeQuietly(out, err, receiver);
+ IoUtils.closeQuietly(remoteWindow, out, err, receiver);
super.doCloseImmediately();
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/WindowTest.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/WindowTest.java b/sshd-core/src/test/java/org/apache/sshd/WindowTest.java
deleted file mode 100644
index 084f4fe..0000000
--- a/sshd-core/src/test/java/org/apache/sshd/WindowTest.java
+++ /dev/null
@@ -1,345 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.sshd;
-
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.sshd.client.SshClient;
-import org.apache.sshd.client.channel.ChannelShell;
-import org.apache.sshd.client.channel.ClientChannel;
-import org.apache.sshd.client.future.OpenFuture;
-import org.apache.sshd.client.session.ClientSession;
-import org.apache.sshd.common.FactoryManager;
-import org.apache.sshd.common.FactoryManagerUtils;
-import org.apache.sshd.common.NamedFactory;
-import org.apache.sshd.common.RuntimeSshException;
-import org.apache.sshd.common.Service;
-import org.apache.sshd.common.channel.Channel;
-import org.apache.sshd.common.channel.Window;
-import org.apache.sshd.common.io.IoReadFuture;
-import org.apache.sshd.common.session.Session;
-import org.apache.sshd.common.util.buffer.Buffer;
-import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
-import org.apache.sshd.server.Command;
-import org.apache.sshd.server.CommandFactory;
-import org.apache.sshd.server.SshServer;
-import org.apache.sshd.server.auth.pubkey.AcceptAllPublickeyAuthenticator;
-import org.apache.sshd.server.channel.ChannelSession;
-import org.apache.sshd.server.channel.ChannelSessionFactory;
-import org.apache.sshd.server.command.UnknownCommand;
-import org.apache.sshd.server.forward.DirectTcpipFactory;
-import org.apache.sshd.server.session.ServerConnectionService;
-import org.apache.sshd.server.session.ServerConnectionServiceFactory;
-import org.apache.sshd.server.session.ServerUserAuthService;
-import org.apache.sshd.server.session.ServerUserAuthServiceFactory;
-import org.apache.sshd.util.AsyncEchoShellFactory;
-import org.apache.sshd.util.BaseTestSupport;
-import org.apache.sshd.util.BogusPasswordAuthenticator;
-import org.apache.sshd.util.EchoShellFactory;
-import org.apache.sshd.util.Utils;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.FixMethodOrder;
-import org.junit.Test;
-import org.junit.runners.MethodSorters;
-
-/**
- * TODO Add javadoc
- *
- * @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
- */
-@FixMethodOrder(MethodSorters.NAME_ASCENDING)
-public class WindowTest extends BaseTestSupport {
-
- private SshServer sshd;
- private SshClient client;
- private int port;
- private CountDownLatch authLatch;
- private CountDownLatch channelLatch;
-
- @Before
- public void setUp() throws Exception {
- authLatch = new CountDownLatch(0);
- channelLatch = new CountDownLatch(0);
-
- sshd = SshServer.setUpDefaultServer();
- sshd.setKeyPairProvider(Utils.createTestHostKeyProvider());
- sshd.setShellFactory(new TestEchoShellFactory());
- sshd.setCommandFactory(new CommandFactory() {
- @Override
- public Command createCommand(String command) {
- return new UnknownCommand(command);
- }
- });
- sshd.setPasswordAuthenticator(BogusPasswordAuthenticator.INSTANCE);
- sshd.setPublickeyAuthenticator(AcceptAllPublickeyAuthenticator.INSTANCE);
- sshd.setServiceFactories(Arrays.asList(
- new ServerUserAuthServiceFactory() {
- @Override
- public Service create(Session session) throws IOException {
- return new ServerUserAuthService(session) {
- @SuppressWarnings("synthetic-access")
- @Override
- public void process(int cmd, Buffer buffer) throws Exception {
- authLatch.await();
- super.process(cmd, buffer);
- }
- };
- }
- },
- ServerConnectionServiceFactory.INSTANCE
- ));
- sshd.setChannelFactories(Arrays.<NamedFactory<Channel>>asList(
- new ChannelSessionFactory() {
- @Override
- public Channel create() {
- return new ChannelSession() {
- @SuppressWarnings("synthetic-access")
- @Override
- public OpenFuture open(int recipient, int rwsize, int rmpsize, Buffer buffer) {
- try {
- channelLatch.await();
- } catch (InterruptedException e) {
- throw new RuntimeSshException(e);
- }
- return super.open(recipient, rwsize, rmpsize, buffer);
- }
-
- @Override
- public String toString() {
- return "ChannelSession" + "[id=" + id + ", recipient=" + recipient + "]";
- }
- };
- }
- },
- DirectTcpipFactory.INSTANCE));
- sshd.start();
- port = sshd.getPort();
-
- client = SshClient.setUpDefaultClient();
- }
-
- @After
- public void tearDown() throws Exception {
- if (sshd != null) {
- sshd.stop(true);
- }
- if (client != null) {
- client.stop();
- }
- }
-
- @Test
- public void testWindowConsumptionWithInvertedStreams() throws Exception {
- sshd.setShellFactory(new AsyncEchoShellFactory());
- FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
- FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
- client.start();
-
- try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
- session.addPasswordIdentity(getCurrentTestName());
- session.auth().verify(5L, TimeUnit.SECONDS);
-
- try (ChannelShell channel = session.createShellChannel()) {
- channel.open().verify(5L, TimeUnit.SECONDS);
-
- try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
- Window clientLocal = channel.getLocalWindow();
- Window clientRemote = channel.getRemoteWindow();
- Window serverLocal = serverChannel.getLocalWindow();
- Window serverRemote = serverChannel.getRemoteWindow();
-
- final String message = "0123456789";
- final int nbMessages = 500;
-
- try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(channel.getInvertedIn()));
- BufferedReader reader = new BufferedReader(new InputStreamReader(channel.getInvertedOut()))) {
-
- for (int i = 0; i < nbMessages; i++) {
- writer.write(message);
- writer.write("\n");
- writer.flush();
-
- waitForWindowNotEquals(clientLocal, serverRemote, "client local", "server remote");
-
- String line = reader.readLine();
- assertEquals("Mismatched message at line #" + i, message, line);
-
- waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote");
- waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local");
- }
- }
- }
- }
- } finally {
- client.stop();
- }
- }
-
- @Test
- public void testWindowConsumptionWithDirectStreams() throws Exception {
- sshd.setShellFactory(new AsyncEchoShellFactory());
- FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
- FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
-
- client.start();
-
- try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
- session.addPasswordIdentity(getCurrentTestName());
- session.auth().verify(5L, TimeUnit.SECONDS);
-
- try (ChannelShell channel = session.createShellChannel();
- PipedInputStream inPis = new PipedInputStream();
- PipedOutputStream inPos = new PipedOutputStream(inPis);
- PipedInputStream outPis = new PipedInputStream();
- PipedOutputStream outPos = new PipedOutputStream(outPis)) {
-
- channel.setIn(inPis);
- channel.setOut(outPos);
- channel.open().verify();
-
- try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
- Window clientLocal = channel.getLocalWindow();
- Window clientRemote = channel.getRemoteWindow();
- Window serverLocal = serverChannel.getLocalWindow();
- Window serverRemote = serverChannel.getRemoteWindow();
-
- final String message = "0123456789";
- final int nbMessages = 500;
-
- try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(inPos));
- BufferedReader reader = new BufferedReader(new InputStreamReader(outPis))) {
- for (int i = 0; i < nbMessages; i++) {
- writer.write(message);
- writer.write("\n");
- writer.flush();
-
- waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote");
-
- String line = reader.readLine();
- assertEquals("Mismatched message at line #" + i, message, line);
-
- waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote");
- waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local");
- }
- }
- }
- }
- } finally {
- client.stop();
- }
- }
-
- @Test
- public void testWindowConsumptionWithAsyncStreams() throws Exception {
- sshd.setShellFactory(new AsyncEchoShellFactory());
- FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
- FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
-
- client.start();
-
- try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
- session.addPasswordIdentity(getCurrentTestName());
- session.auth().verify(5L, TimeUnit.SECONDS);
-
- try (ChannelShell channel = session.createShellChannel()) {
- channel.setStreaming(ClientChannel.Streaming.Async);
- channel.open().verify(5L, TimeUnit.SECONDS);
-
- try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
- Window clientLocal = channel.getLocalWindow();
- Window clientRemote = channel.getRemoteWindow();
- Window serverLocal = serverChannel.getLocalWindow();
- Window serverRemote = serverChannel.getRemoteWindow();
-
- final String message = "0123456789\n";
- final byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
- final int nbMessages = 500;
- for (int i = 0; i < nbMessages; i++) {
- Buffer buffer = new ByteArrayBuffer(bytes);
- channel.getAsyncIn().write(buffer).verify();
-
- waitForWindowNotEquals(clientLocal, serverRemote, "client local", "server remote");
-
- Buffer buf = new ByteArrayBuffer(16);
- IoReadFuture future = channel.getAsyncOut().read(buf);
- future.verify();
- assertEquals("Mismatched available data at line #" + i, message.length(), buf.available());
- assertEquals("Mismatched data at line #" + i, message, new String(buf.array(), buf.rpos(), buf.available()));
-
- waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote");
- waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local");
- }
- }
- }
- } finally {
- client.stop();
- }
- }
-
- protected void waitForWindowNotEquals(Window w1, Window w2, String n1, String n2) throws InterruptedException {
- for (int j = 0; j < 50; j++) {
- if (w1.getSize() != w2.getSize()) {
- break;
- }
- Thread.sleep(1);
- }
- assertNotEquals(n1 + " and " + n2, w1.getSize(), w2.getSize());
- }
-
- protected void waitForWindowEquals(Window w1, Window w2, String n1, String n2) throws InterruptedException {
- for (int j = 0; j < 50; j++) {
- if (w1.getSize() == w2.getSize()) {
- break;
- }
- Thread.sleep(1);
- }
- assertEquals(n1 + " and " + n2, w1.getSize(), w2.getSize());
- }
-
- public static class TestEchoShellFactory extends EchoShellFactory {
- @Override
- public Command create() {
- return new TestEchoShell();
- }
-
- public static class TestEchoShell extends EchoShell {
-
- public static CountDownLatch latch = new CountDownLatch(1);
-
- @Override
- public void destroy() {
- if (latch != null) {
- latch.countDown();
- }
- super.destroy();
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/common/channel/ChannelPipedInputStreamTest.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/common/channel/ChannelPipedInputStreamTest.java b/sshd-core/src/test/java/org/apache/sshd/common/channel/ChannelPipedInputStreamTest.java
index caadf0e..926bb71 100644
--- a/sshd-core/src/test/java/org/apache/sshd/common/channel/ChannelPipedInputStreamTest.java
+++ b/sshd-core/src/test/java/org/apache/sshd/common/channel/ChannelPipedInputStreamTest.java
@@ -21,6 +21,7 @@ package org.apache.sshd.common.channel;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Collections;
import org.apache.sshd.util.BaseTestSupport;
import org.apache.sshd.util.BogusChannel;
@@ -36,8 +37,7 @@ public class ChannelPipedInputStreamTest extends BaseTestSupport {
@Test
public void testAvailable() throws IOException {
- Window window = new Window(new BogusChannel(), null, true, true);
- try (ChannelPipedInputStream stream = new ChannelPipedInputStream(window)) {
+ try (ChannelPipedInputStream stream = createTestStream()) {
byte[] b = getCurrentTestName().getBytes(StandardCharsets.UTF_8);
stream.receive(b, 0, b.length);
assertEquals("Mismatched reported available size after receive", b.length, stream.available());
@@ -54,8 +54,7 @@ public class ChannelPipedInputStreamTest extends BaseTestSupport {
@Test
public void testIdempotentClose() throws IOException {
- Window window = new Window(new BogusChannel(), null, true, true);
- try (ChannelPipedInputStream stream = new ChannelPipedInputStream(window)) {
+ try (ChannelPipedInputStream stream = createTestStream()) {
byte[] b = getCurrentTestName().getBytes(StandardCharsets.UTF_8);
stream.receive(b, 0, b.length);
stream.eof();
@@ -66,11 +65,17 @@ public class ChannelPipedInputStreamTest extends BaseTestSupport {
}
}
- private void assertStreamEquals(byte[] expected, byte[] read) {
+ private static ChannelPipedInputStream createTestStream() {
+ Window window = new Window(new BogusChannel(), null, true, true);
+ window.init(Collections.<String,Object>emptyMap());
+ return new ChannelPipedInputStream(window);
+ }
+
+ private static void assertStreamEquals(byte[] expected, byte[] read) {
if (expected.length > read.length) {
fail("Less bytes than expected: " + Arrays.toString(expected) + " but got: " + Arrays.toString(read));
} else {
- assertArrayEquals(expected, Arrays.copyOf(read, expected.length));
+ assertArrayEquals("Mismatched stream content", expected, Arrays.copyOf(read, expected.length));
for (int i = expected.length; i < read.length; i++) {
assertTrue("Non-zero value at position " + i, read[i] == 0);
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/common/channel/WindowTest.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/common/channel/WindowTest.java b/sshd-core/src/test/java/org/apache/sshd/common/channel/WindowTest.java
new file mode 100644
index 0000000..f8b23fb
--- /dev/null
+++ b/sshd-core/src/test/java/org/apache/sshd/common/channel/WindowTest.java
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sshd.common.channel;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.channel.ChannelShell;
+import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.future.OpenFuture;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.FactoryManager;
+import org.apache.sshd.common.FactoryManagerUtils;
+import org.apache.sshd.common.NamedFactory;
+import org.apache.sshd.common.RuntimeSshException;
+import org.apache.sshd.common.Service;
+import org.apache.sshd.common.io.IoInputStream;
+import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.io.IoReadFuture;
+import org.apache.sshd.common.session.Session;
+import org.apache.sshd.common.util.buffer.Buffer;
+import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.server.Command;
+import org.apache.sshd.server.CommandFactory;
+import org.apache.sshd.server.SshServer;
+import org.apache.sshd.server.auth.pubkey.AcceptAllPublickeyAuthenticator;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.channel.ChannelSessionFactory;
+import org.apache.sshd.server.command.UnknownCommand;
+import org.apache.sshd.server.forward.DirectTcpipFactory;
+import org.apache.sshd.server.session.ServerConnectionService;
+import org.apache.sshd.server.session.ServerConnectionServiceFactory;
+import org.apache.sshd.server.session.ServerUserAuthService;
+import org.apache.sshd.server.session.ServerUserAuthServiceFactory;
+import org.apache.sshd.util.AsyncEchoShellFactory;
+import org.apache.sshd.util.BaseTestSupport;
+import org.apache.sshd.util.BogusPasswordAuthenticator;
+import org.apache.sshd.util.EchoShellFactory;
+import org.apache.sshd.util.Utils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+
+/**
+ * TODO Add javadoc
+ *
+ * @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class WindowTest extends BaseTestSupport {
+
+ private SshServer sshd;
+ private SshClient client;
+ private int port;
+ private CountDownLatch authLatch;
+ private CountDownLatch channelLatch;
+
+ public WindowTest() {
+ super();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ authLatch = new CountDownLatch(0);
+ channelLatch = new CountDownLatch(0);
+
+ sshd = SshServer.setUpDefaultServer();
+ sshd.setKeyPairProvider(Utils.createTestHostKeyProvider());
+ sshd.setShellFactory(new TestEchoShellFactory());
+ sshd.setCommandFactory(new CommandFactory() {
+ @Override
+ public Command createCommand(String command) {
+ return new UnknownCommand(command);
+ }
+ });
+ sshd.setPasswordAuthenticator(BogusPasswordAuthenticator.INSTANCE);
+ sshd.setPublickeyAuthenticator(AcceptAllPublickeyAuthenticator.INSTANCE);
+ sshd.setServiceFactories(Arrays.asList(
+ new ServerUserAuthServiceFactory() {
+ @Override
+ public Service create(Session session) throws IOException {
+ return new ServerUserAuthService(session) {
+ @SuppressWarnings("synthetic-access")
+ @Override
+ public void process(int cmd, Buffer buffer) throws Exception {
+ authLatch.await();
+ super.process(cmd, buffer);
+ }
+ };
+ }
+ },
+ ServerConnectionServiceFactory.INSTANCE
+ ));
+ sshd.setChannelFactories(Arrays.<NamedFactory<Channel>>asList(
+ new ChannelSessionFactory() {
+ @Override
+ public Channel create() {
+ return new ChannelSession() {
+ @SuppressWarnings("synthetic-access")
+ @Override
+ public OpenFuture open(int recipient, int rwsize, int rmpsize, Buffer buffer) {
+ try {
+ channelLatch.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeSshException(e);
+ }
+ return super.open(recipient, rwsize, rmpsize, buffer);
+ }
+
+ @Override
+ public String toString() {
+ return "ChannelSession" + "[id=" + id + ", recipient=" + recipient + "]";
+ }
+ };
+ }
+ },
+ DirectTcpipFactory.INSTANCE));
+ sshd.start();
+ port = sshd.getPort();
+
+ client = SshClient.setUpDefaultClient();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (sshd != null) {
+ sshd.stop(true);
+ }
+ if (client != null) {
+ client.stop();
+ }
+ }
+
+ @Test
+ public void testWindowConsumptionWithInvertedStreams() throws Exception {
+ sshd.setShellFactory(new AsyncEchoShellFactory());
+ FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
+ FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
+ client.start();
+
+ try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
+ session.addPasswordIdentity(getCurrentTestName());
+ session.auth().verify(5L, TimeUnit.SECONDS);
+
+ try (ChannelShell channel = session.createShellChannel()) {
+ channel.open().verify(5L, TimeUnit.SECONDS);
+
+ try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
+ Window clientLocal = channel.getLocalWindow();
+ Window clientRemote = channel.getRemoteWindow();
+ Window serverLocal = serverChannel.getLocalWindow();
+ Window serverRemote = serverChannel.getRemoteWindow();
+
+ final String message = "0123456789";
+ final int nbMessages = 500;
+
+ try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(channel.getInvertedIn()));
+ BufferedReader reader = new BufferedReader(new InputStreamReader(channel.getInvertedOut()))) {
+
+ for (int i = 0; i < nbMessages; i++) {
+ writer.write(message);
+ writer.write("\n");
+ writer.flush();
+
+ waitForWindowNotEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+
+ String line = reader.readLine();
+ assertEquals("Mismatched message at line #" + i, message, line);
+
+ waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+ waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local", TimeUnit.SECONDS.toMillis(3L));
+ }
+ }
+ }
+ }
+ } finally {
+ client.stop();
+ }
+ }
+
+ @Test
+ public void testWindowConsumptionWithDirectStreams() throws Exception {
+ sshd.setShellFactory(new AsyncEchoShellFactory());
+ FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
+ FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
+
+ client.start();
+
+ try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
+ session.addPasswordIdentity(getCurrentTestName());
+ session.auth().verify(5L, TimeUnit.SECONDS);
+
+ try (ChannelShell channel = session.createShellChannel();
+ PipedInputStream inPis = new PipedInputStream();
+ PipedOutputStream inPos = new PipedOutputStream(inPis);
+ PipedInputStream outPis = new PipedInputStream();
+ PipedOutputStream outPos = new PipedOutputStream(outPis)) {
+
+ channel.setIn(inPis);
+ channel.setOut(outPos);
+ channel.open().verify(7L, TimeUnit.SECONDS);
+
+ try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
+ Window clientLocal = channel.getLocalWindow();
+ Window clientRemote = channel.getRemoteWindow();
+ Window serverLocal = serverChannel.getLocalWindow();
+ Window serverRemote = serverChannel.getRemoteWindow();
+
+ final String message = "0123456789";
+ final int nbMessages = 500;
+
+ try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(inPos));
+ BufferedReader reader = new BufferedReader(new InputStreamReader(outPis))) {
+ for (int i = 0; i < nbMessages; i++) {
+ writer.write(message);
+ writer.write('\n');
+ writer.flush();
+
+ waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+
+ String line = reader.readLine();
+ assertEquals("Mismatched message at line #" + i, message, line);
+
+ waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+ waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local", TimeUnit.SECONDS.toMillis(3L));
+ }
+ }
+ }
+ }
+ } finally {
+ client.stop();
+ }
+ }
+
+ @Test
+ public void testWindowConsumptionWithAsyncStreams() throws Exception {
+ sshd.setShellFactory(new AsyncEchoShellFactory());
+ FactoryManagerUtils.updateProperty(sshd, FactoryManager.WINDOW_SIZE, 1024);
+ FactoryManagerUtils.updateProperty(client, FactoryManager.WINDOW_SIZE, 1024);
+
+ client.start();
+
+ try (ClientSession session = client.connect(getCurrentTestName(), "localhost", port).verify(7L, TimeUnit.SECONDS).getSession()) {
+ session.addPasswordIdentity(getCurrentTestName());
+ session.auth().verify(5L, TimeUnit.SECONDS);
+
+ try (ChannelShell channel = session.createShellChannel()) {
+ channel.setStreaming(ClientChannel.Streaming.Async);
+ channel.open().verify(5L, TimeUnit.SECONDS);
+
+ try (Channel serverChannel = sshd.getActiveSessions().iterator().next().getService(ServerConnectionService.class).getChannels().iterator().next()) {
+ Window clientLocal = channel.getLocalWindow();
+ Window clientRemote = channel.getRemoteWindow();
+ Window serverLocal = serverChannel.getLocalWindow();
+ Window serverRemote = serverChannel.getRemoteWindow();
+
+ final String message = "0123456789\n";
+ final byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
+ final int nbMessages = 500;
+ IoOutputStream output = channel.getAsyncIn();
+ IoInputStream input = channel.getAsyncOut();
+ for (int i = 0; i < nbMessages; i++) {
+ Buffer buffer = new ByteArrayBuffer(bytes);
+ output.write(buffer).verify(5L, TimeUnit.SECONDS);
+
+ waitForWindowNotEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+
+ Buffer buf = new ByteArrayBuffer(16);
+ IoReadFuture future = input.read(buf);
+ future.verify(5L, TimeUnit.SECONDS);
+ assertEquals("Mismatched available data at line #" + i, message.length(), buf.available());
+ assertEquals("Mismatched data at line #" + i, message, new String(buf.array(), buf.rpos(), buf.available()));
+
+ waitForWindowEquals(clientLocal, serverRemote, "client local", "server remote", TimeUnit.SECONDS.toMillis(3L));
+ waitForWindowEquals(clientRemote, serverLocal, "client remote", "server local", TimeUnit.SECONDS.toMillis(3L));
+ }
+ }
+ }
+ } finally {
+ client.stop();
+ }
+ }
+
+ private static void waitForWindowNotEquals(Window w1, Window w2, String n1, String n2, long maxWait) throws InterruptedException {
+ for (long waited = 0L, maxWaitNanos = TimeUnit.MILLISECONDS.toNanos(maxWait); waited < maxWaitNanos; ) {
+ if (w1.getSize() != w2.getSize()) {
+ return;
+ }
+
+ long nanoStart = System.nanoTime();
+ Thread.sleep(1L);
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ waited += nanoDuration;
+ }
+
+ // one last chance ...
+ assertNotEquals(n1 + " and " + n2, w1.getSize(), w2.getSize());
+ }
+
+ private static void waitForWindowEquals(Window w1, Window w2, String n1, String n2, long maxWait) throws InterruptedException {
+ for (long waited = 0L, maxWaitNanos = TimeUnit.MILLISECONDS.toNanos(maxWait); waited < maxWaitNanos; ) {
+ if (w1.getSize() == w2.getSize()) {
+ return;
+ }
+
+ long nanoStart = System.nanoTime();
+ Thread.sleep(1L);
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
+ waited += nanoDuration;
+ }
+
+ // one last chance ...
+ assertEquals(n1 + " and " + n2, w1.getSize(), w2.getSize());
+ }
+
+ public static class TestEchoShellFactory extends EchoShellFactory {
+ @Override
+ public Command create() {
+ return new TestEchoShell();
+ }
+
+ public static class TestEchoShell extends EchoShell {
+
+ public static CountDownLatch latch = new CountDownLatch(1);
+
+ @Override
+ public void destroy() {
+ if (latch != null) {
+ latch.countDown();
+ }
+ super.destroy();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/server/ServerTest.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/server/ServerTest.java b/sshd-core/src/test/java/org/apache/sshd/server/ServerTest.java
index 0a3b2a0..4c2c4ba 100644
--- a/sshd-core/src/test/java/org/apache/sshd/server/ServerTest.java
+++ b/sshd-core/src/test/java/org/apache/sshd/server/ServerTest.java
@@ -301,7 +301,8 @@ public class ServerTest extends BaseTestSupport {
for (long totalNanoTime = 0L; channel.getRemoteWindow().getSize() > 0; ) {
long nanoStart = System.nanoTime();
Thread.sleep(1L);
- long nanoEnd = System.nanoTime(), nanoDuration = nanoEnd - nanoStart;
+ long nanoEnd = System.nanoTime();
+ long nanoDuration = nanoEnd - nanoStart;
totalNanoTime += nanoDuration;
assertTrue("Waiting for too long on remote window size to reach zero", totalNanoTime < TimeUnit.MILLISECONDS.toNanos(MAX_TIMEOUT_VALUE));
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/server/channel/ChannelSessionTest.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/server/channel/ChannelSessionTest.java b/sshd-core/src/test/java/org/apache/sshd/server/channel/ChannelSessionTest.java
index ce1ad71..5e13a07 100644
--- a/sshd-core/src/test/java/org/apache/sshd/server/channel/ChannelSessionTest.java
+++ b/sshd-core/src/test/java/org/apache/sshd/server/channel/ChannelSessionTest.java
@@ -20,6 +20,8 @@ package org.apache.sshd.server.channel;
import java.io.IOException;
import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.sshd.common.channel.ChannelAsyncOutputStream;
import org.apache.sshd.common.util.buffer.Buffer;
@@ -34,8 +36,9 @@ import org.junit.runners.MethodSorters;
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class ChannelSessionTest extends BaseTestSupport {
-
- private boolean expanded = false;
+ public ChannelSessionTest() {
+ super();
+ }
/**
* Test whether onWindowExpanded is called from server session
@@ -45,17 +48,21 @@ public class ChannelSessionTest extends BaseTestSupport {
final Buffer buffer = new ByteArrayBuffer();
buffer.putInt(1234);
- try (ChannelSession channelSession = new ChannelSession()) {
+ try (ChannelSession channelSession = new ChannelSession() {
+ {
+ this.remoteWindow.init(Collections.<String,Object>emptyMap());
+ }
+ }) {
+ final AtomicBoolean expanded = new AtomicBoolean(false);
channelSession.asyncOut = new ChannelAsyncOutputStream(new BogusChannel(), (byte) 0) {
- @SuppressWarnings("synthetic-access")
@Override
public void onWindowExpanded() throws IOException {
- expanded = true;
+ expanded.set(true);
super.onWindowExpanded();
}
};
channelSession.handleWindowAdjust(buffer);
- assertTrue(expanded);
+ assertTrue("Expanded ?", expanded.get());
}
}
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/java/org/apache/sshd/util/BogusChannel.java
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/java/org/apache/sshd/util/BogusChannel.java b/sshd-core/src/test/java/org/apache/sshd/util/BogusChannel.java
index 14b0055..028e1d8 100644
--- a/sshd-core/src/test/java/org/apache/sshd/util/BogusChannel.java
+++ b/sshd-core/src/test/java/org/apache/sshd/util/BogusChannel.java
@@ -26,6 +26,9 @@ import org.apache.sshd.common.channel.AbstractChannel;
import org.apache.sshd.common.util.buffer.Buffer;
public class BogusChannel extends AbstractChannel {
+ public BogusChannel() {
+ super(false);
+ }
@Override
protected void doWriteData(byte[] data, int off, int len) throws IOException {
http://git-wip-us.apache.org/repos/asf/mina-sshd/blob/bda6cb09/sshd-core/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/sshd-core/src/test/resources/log4j.properties b/sshd-core/src/test/resources/log4j.properties
index 13cd606..fd960fc 100644
--- a/sshd-core/src/test/resources/log4j.properties
+++ b/sshd-core/src/test/resources/log4j.properties
@@ -23,7 +23,7 @@
#
log4j.rootLogger=INFO, stdout
#log4j.logger.org.apache.sshd=TRACE
-#log4j.logger.org.apache.sshd.common.channel.Window=DEBUG
+log4j.logger.org.apache.sshd.common.channel.Window=DEBUG
# CONSOLE appender
log4j.appender.stdout=org.apache.log4j.ConsoleAppender