如何在 projectreactor 中使用 sshd mina/netty

Question

我想开发一个 java 反应应用程序，它需要通过 ssh 与一些外部服务进行通信。作为反应式框架，我使用项目反应器中的 spring boot webflux 和 ssh 客户端的 sshd mina/netty。基本上，应用程序将打开一个 ssh 会话并在服务器上运行一些命令。该命令的逻辑取决于先前命令的响应。

问题是，如何将sshd mina 集成到spring boot 项目reactor (Mono/Flux) 中？

sshd mina 提供了使用异步响应的可能性，如测试所示：https://github.com/apache/mina-sshd/blob/c876cce935f9278b0d50f02fd554eff1af949060/sshd-core/src/test/java/org/apache/sshd/client/ClientTest.java#L560

但我不知道如何将其与 Mono/Flux 集成。

到目前为止，我能够在发送命令后获得与登录对应的响应，但不能获得以下响应。

这是我的设置代码

测试 ssh 服务器是通过 docker 创建的

docker run -d --rm -e SUDO_ACCESS=false -e PASSWORD_ACCESS=true -e USER_NAME=username -e USER_PASSWORD=password -p 2222:2222 lscr.io/linuxserver/openssh-server:latest

java项目包含以下依赖项

    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
      <groupId>org.apache.sshd</groupId>
      <artifactId>sshd-netty</artifactId>
      <version>2.8.0</version>
    </dependency>

    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-test</artifactId>
      <scope>test</scope>
    </dependency>

我希望它具有反应性的 ssh 客户端代码 使用 mina 文档 (https://github.com/apache/mina-sshd) 以及我能找到的关于 mina 异步客户端用法的唯一示例 (https://github.com/apache/mina-sshd/blob/master/sshd-core/src/test/java/org/apache/sshd/client/ClientTest.java#L518)

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.time.Duration;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.channel.ClientChannel;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.common.channel.StreamingChannel;
import org.apache.sshd.common.future.SshFutureListener;
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.io.IoWriteFuture;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;

public class SshDocker implements AutoCloseable {
  private static final String hostname = "localhost";
  private static final String username = "username";
  private static final String password = "password";
  private static final int port = 2222;
  private static final Duration clientTimeout = Duration.ofSeconds(10);
  private SshClient client;
  private ClientSession session;
  private ClientChannel channel;

  public Mono<String> open() throws IOException {
    client = SshClient.setUpDefaultClient();
    client.start();

    session = client.connect(username, hostname, port).verify(clientTimeout).getSession();
    session.addPasswordIdentity(password);
    session.auth().verify(clientTimeout);

    channel = session.createShellChannel();
    channel.setStreaming(StreamingChannel.Streaming.Async);
    channel.open().verify(clientTimeout);

    final Duration timeout = Duration.ofSeconds(10);
    return readResponse(timeout);
  }

  @Override
  public void close() throws Exception {
    channel.close();
    session.close();
    client.close();
  }

  public Mono<String> execCommand(final String command, final Duration timeout) {
    return runCommand(command, timeout).flatMap(v -> readResponse(timeout));
  }

  private Mono<Void> runCommand(final String command, final Duration timeout) {
    final IoOutputStream requestStream = channel.getAsyncIn();
    return Mono.create(
        voidMonoSink -> {
          final ReactiveMonoRequestListener reactiveMonoRequestListener =
              new ReactiveMonoRequestListener(timeout, voidMonoSink);
          try {
            requestStream
                .writeBuffer(new ByteArrayBuffer(command.getBytes()))
                .addListener(reactiveMonoRequestListener);
          } catch (final IOException e) {
            throw new RuntimeException(e);
          }
        });
  }

  private Mono<String> readResponse(final Duration timeout) {
    final IoInputStream responseStream = channel.getAsyncOut();
    return Mono.create(
        monoSink -> {
          final ReactiveMonoResponseListener reactiveResponseListener =
              new ReactiveMonoResponseListener(responseStream, timeout, monoSink);
          responseStream.read(new ByteArrayBuffer()).addListener(reactiveResponseListener);
        });
  }

  public static class ReactiveMonoResponseListener implements SshFutureListener<IoReadFuture> {

    final IoInputStream responseStream;
    final ByteArrayOutputStream result = new ByteArrayOutputStream();
    private final Duration timeout;
    private final MonoSink<String> handler;

    public ReactiveMonoResponseListener(
        final IoInputStream responseStream,
        final Duration timeout,
        final MonoSink<String> handler) {
      this.responseStream = responseStream;
      this.timeout = timeout;
      this.handler = handler;
    }

    @Override
    public void operationComplete(final IoReadFuture ioReadFuture) {
      System.out.println("Operation Read Complete");
      if (handler != null) {
        try {
          ioReadFuture.verify(timeout);
          final Buffer buffer = ioReadFuture.getBuffer();
          result.write(buffer.array(), buffer.rpos(), buffer.available());
          buffer.rpos(buffer.rpos() + buffer.available());
          buffer.compact();
          if (!result.toString().endsWith("$ ")) { // read response until next prompt
            responseStream.read(buffer).addListener(this);
          } else {
            System.out.println("response >>>>>>>>");
            System.out.println(result);
            System.out.println("<<<<<<<< response");

            handler.success(result.toString());
          }
        } catch (final IOException e) {
          handler.error(e);
        }
      }
    }
  }

  public static class ReactiveMonoRequestListener implements SshFutureListener<IoWriteFuture> {

    private final MonoSink<Void> handler;
    private final Duration timeout;

    public ReactiveMonoRequestListener(final Duration timeout, final MonoSink<Void> handler) {
      this.handler = handler;
      this.timeout = timeout;
    }

    @Override
    public void operationComplete(final IoWriteFuture ioWriteFuture) {
      System.out.println("Operation Write Complete");
      if (handler != null) {
        try {
          ioWriteFuture.verify(timeout);
          handler.success();
        } catch (final IOException e) {
          handler.error(e);
        }
      }
    }
  }
}

用于运行反应式 ssh 客户端的测试

import java.time.Duration;
import org.junit.jupiter.api.Test;

class SshDockerTest {

  @Test
  void run() throws Exception {
    final SshDocker sshClient = new SshDocker();
    sshClient
        .open()
        .flatMap(v -> sshClient.execCommand("ls\n", Duration.ofSeconds(3)))
        .subscribe(System.out::println);
  }
}

运行测试时，在调试日志旁边，我获得：

Operation Read Complete
response >>>>>>>>
Welcome to OpenSSH Server

65d098057769:~$
<<<<<<<< response
Operation Write Complete

但没有迹象表明对ls 命令有响应

如果无法将 sshd mina 转换为反应式，那么替代反应式解决方案是什么？

谢谢

Answer 1

我终于找到了一个可行的解决方案。之前的问题是Mono<Void>没有触发管道中的后续任务。

以下是新变化：

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.time.Duration;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.channel.ClientChannel;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.common.channel.StreamingChannel;
import org.apache.sshd.common.future.SshFutureListener;
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.io.IoWriteFuture;
import org.apache.sshd.common.util.buffer.Buffer;
import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;

// start a docker container
// docker run -d --rm -e SUDO_ACCESS=false -e PASSWORD_ACCESS=true -e USER_NAME=username -e
// USER_PASSWORD=password -p 2222:2222 lscr.io/linuxserver/openssh-server:latest

public class SshDocker {
  private static final String hostname = "localhost";
  private static final String username = "username";
  private static final String password = "password";
  private static final int port = 2222;
  private static final Duration clientTimeout = Duration.ofSeconds(10);
  private IoInputStream responseStream;
  private IoOutputStream requestStream;
  private SshClient client;
  private ClientSession session;
  private ClientChannel channel;

  private Boolean openNotReactive() throws IOException {
    client = SshClient.setUpDefaultClient();
    client.start();

    session = client.connect(username, hostname, port).verify(clientTimeout).getSession();
    session.addPasswordIdentity(password);
    session.auth().verify(clientTimeout);

    channel = session.createShellChannel();
    channel.setStreaming(StreamingChannel.Streaming.Async);
    channel.open().verify(clientTimeout);

    responseStream = channel.getAsyncOut();
    requestStream = channel.getAsyncIn();
    return true;
  }

  public Mono<String> open() {
    final Duration timeout = Duration.ofSeconds(10);
    final Mono<Boolean> open =
        Mono.fromCallable(this::openNotReactive).subscribeOn(Schedulers.boundedElastic());
    return open.flatMap(r -> readResponse(timeout));
  }

  public Mono<Boolean> close() {
    return Mono.fromCallable(
            () -> {
              closeNotReactive();
              return true;
            })
        .subscribeOn(Schedulers.boundedElastic());
  }

  private void closeNotReactive() throws Exception {
    System.out.println("Closing");
    channel.close();
    session.close();
    client.close();
  }

  public Mono<String> execCommand(final String command, final Duration timeout) {
    return runCommand(command, timeout).flatMap(v -> readResponse(timeout)).log();
  }

  private Mono<Boolean> runCommand(final String command, final Duration timeout) {
    final String cmd = String.format("%s
", command.strip());
    return Mono.create(
        monoSink -> {
          final ReactiveMonoRequestListener reactiveMonoRequestListener =
              new ReactiveMonoRequestListener(timeout, monoSink);
          try {
            final IoWriteFuture writeFuture =
                requestStream.writeBuffer(new ByteArrayBuffer(cmd.getBytes()));
            writeFuture.addListener(reactiveMonoRequestListener);
            monoSink.onDispose(() -> writeFuture.removeListener(reactiveMonoRequestListener));
          } catch (final IOException e) {
            throw new RuntimeException(e);
          }
        });
  }

  private Mono<String> readResponse(final Duration timeout) {
    // https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html#create-java.util.function.Consumer-
    return Mono.create(
        monoSink -> {
          final ReactiveMonoResponseListener reactiveResponseListener =
              new ReactiveMonoResponseListener(responseStream, timeout, monoSink);
          final IoReadFuture readFuture = responseStream.read(new ByteArrayBuffer());
          readFuture.addListener(reactiveResponseListener);
          monoSink.onDispose(() -> readFuture.removeListener(reactiveResponseListener));
        });
  }

  public static class ReactiveMonoResponseListener implements SshFutureListener<IoReadFuture> {

    final IoInputStream responseStream;
    final ByteArrayOutputStream result = new ByteArrayOutputStream();
    private final Duration timeout;
    private final MonoSink<String> handler;

    public ReactiveMonoResponseListener(
        final IoInputStream responseStream,
        final Duration timeout,
        final MonoSink<String> handler) {
      this.responseStream = responseStream;
      this.timeout = timeout;
      this.handler = handler;
    }

    @Override
    public void operationComplete(final IoReadFuture ioReadFuture) {
      System.out.println("Operation Read Complete");
      if (handler != null) {
        try {
          ioReadFuture.verify(timeout);
          final Buffer buffer = ioReadFuture.getBuffer();
          result.write(buffer.array(), buffer.rpos(), buffer.available());
          buffer.rpos(buffer.rpos() + buffer.available());
          buffer.compact();
          if (!result.toString().endsWith("$ ")) { // read response until next prompt
            responseStream.read(buffer).addListener(this);
          } else {
            System.out.println("response mono >>>>>>>>");
            System.out.println(result);
            System.out.println("<<<<<<<< response mono");

            handler.success(result.toString());
          }
        } catch (final IOException e) {
          handler.error(e);
        }
      }
    }
  }

  public static class ReactiveMonoRequestListener implements SshFutureListener<IoWriteFuture> {

    private final MonoSink<Boolean> handler;
    private final Duration timeout;

    public ReactiveMonoRequestListener(final Duration timeout, final MonoSink<Boolean> handler) {
      this.handler = handler;
      this.timeout = timeout;
    }

    @Override
    public void operationComplete(final IoWriteFuture ioWriteFuture) {
      System.out.println("Operation Write Complete");
      if (handler != null) {
        try {
          ioWriteFuture.verify(timeout);
          handler.success(true);
        } catch (final IOException e) {
          handler.error(e);
        }
      }
    }
  }
}

和运行应用程序的“测试”：

import java.time.Duration;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

class SshDockerTest {

  @Test
  void run() {
    final SshDocker sshClient = new SshDocker();
    final Mono<Boolean> result = sshClient
        .open()
        .flatMap(r -> sshClient.execCommand("ls", Duration.ofSeconds(3)))
        .flatMap(r -> sshClient.execCommand("pwd", Duration.ofSeconds(3)))
        .flatMap(r -> sshClient.close());
    StepVerifier.create(result).expectNext(true).verifyComplete();
  }
}

是否可以将 AutoClosable 与 reactor 一起使用，这样我就不需要太注意手动调用 close 方法？

如果您看到任何改进，请告诉我。

Answer 2

我使用了下面的代码，我想知道如果命令是 vi，如何读取响应 if (!result.toString().endsWith("$ ")) { // read response until next prompt responseStream.read(buffer).addListener(this); }