implemented OutputWriter to avoid direct writes to stdout and stderr streams

Author: Juraj Veverka

ssh-server/ssh-examples/src/main/java/itx/examples/sshd/commands/SshClientCommandProcessor.java

@@ -12,26 +12,25 @@
 import itx.examples.sshd.sessions.SshClientSessionListenerImpl;
 import itx.ssh.server.commands.CommandProcessor;
 import itx.ssh.server.commands.CommandResult;
+import itx.ssh.server.commands.OutputWriter;
 import itx.ssh.server.commands.subsystem.SshClientSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.io.OutputStream;
 import java.nio.charset.Charset;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
 public class SshClientCommandProcessor implements CommandProcessor, AutoCloseable {
 
     final private static Logger LOG = LoggerFactory.getLogger(SshClientCommandProcessor.class);
-    final private static int ENTER = 13;
 
     private final ObjectMapper mapper;
 
     private String state;
     private long sessionId;
+    private OutputWriter outputWriter;
     private SshClientSessionListenerImpl sshClientSessionListener;
     private ExecutorService executorService;
 
@@ -43,12 +42,13 @@ public SshClientCommandProcessor(SshClientSessionListenerImpl sshClientSessionLi
     }
 
     @Override
-    public void updateSessionId(long sessionId) {
+    public void onSessionStart(long sessionId, OutputWriter outputWriter) {
         this.sessionId = sessionId;
+        this.outputWriter = outputWriter;
     }
 
     @Override
-    public CommandResult processCommand(byte[] command, OutputStream stdout, OutputStream stderr) throws IOException {
+    public CommandResult processCommand(byte[] command) throws IOException {
         LOG.info("processing command: {} ", new String(command, Charset.forName("UTF-8")));
         JsonNode jsonNode = mapper.readTree(command);
         String type = jsonNode.get("type").asText();
@@ -86,8 +86,7 @@ public void run() {
             ErrorData errorData = new ErrorData("unsupported command");
             response = mapper.writeValueAsBytes(errorData);
         }
-        stdout.write(response);
-        stdout.write(ENTER);
+        outputWriter.writeMessage(response);
         return CommandResult.ok();
     }
 

ssh-server/ssh-examples/src/main/java/itx/examples/sshd/commands/StringCommandProcessorImpl.java

@@ -2,11 +2,11 @@
 
 import itx.ssh.server.commands.CommandProcessor;
 import itx.ssh.server.commands.CommandResult;
+import itx.ssh.server.commands.OutputWriter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.io.OutputStream;
 import java.nio.charset.Charset;
 
 public class StringCommandProcessorImpl implements CommandProcessor, AutoCloseable {
@@ -18,18 +18,20 @@ public class StringCommandProcessorImpl implements CommandProcessor, AutoCloseab
 
     private String state;
     private long sessionId;
+    private OutputWriter outputWriter;
 
     public StringCommandProcessorImpl() {
         state = "";
     }
 
     @Override
-    public void updateSessionId(long sessionId) {
+    public void onSessionStart(long sessionId, OutputWriter outputWriter) {
         this.sessionId = sessionId;
+        this.outputWriter = outputWriter;
     }
 
     @Override
-    public CommandResult processCommand(byte[] command, OutputStream stdout, OutputStream stderr) throws IOException {
+    public CommandResult processCommand(byte[] command) throws IOException {
         String cmd = new String(command, Charset.forName("UTF-8"));
         String[] cmdElements = cmd.split(" ");
         if (CMD_SET.equals(cmdElements[0].trim())) {
@@ -42,11 +44,12 @@ public CommandResult processCommand(byte[] command, OutputStream stdout, OutputS
             return CommandResult.ok();
         } else if (CMD_GET.equals(cmdElements[0].trim())) {
             LOG.info("command: {}", cmd);
-            stdout.write("state: ".getBytes());
-            stdout.write(state.getBytes());
-            stdout.write('\n');
-            stdout.write('\r');
-            stdout.flush();
+            StringBuilder sb = new StringBuilder();
+            sb.append("state: ");
+            sb.append(state.getBytes());
+            sb.append('\n');
+            sb.append('\r');
+            outputWriter.writeStdOutAndFlush(sb.toString().getBytes(Charset.forName("UTF-8")));
             return CommandResult.ok();
         } else if (CMD_EXIT.equals(cmdElements[0].trim())) {
             LOG.info("exit");

ssh-server/ssh-server/src/main/java/itx/ssh/server/commands/CommandProcessor.java

@@ -1,7 +1,6 @@
 package itx.ssh.server.commands;
 
 import java.io.IOException;
-import java.io.OutputStream;
 
 /**
  * Implementation of this interface encapsulates command processing logic on ssh-server.
@@ -10,18 +9,18 @@ public interface CommandProcessor {
 
     /**
      * Set client session ID for this command processor.
+     * Commands are processed after the session has started.
+     * @param outputWriter interface for writing output data
      * @param sessionId
      */
-    void updateSessionId(long sessionId);
+    void onSessionStart(long sessionId, OutputWriter outputWriter);
 
     /**
-     * This method is called when new command is available.
+     * This method is called when new command is available. It is called after session has started.
      * @param command command data to be processed.
-     * @param stdout standard output stream to write processing data in.
-     * @param stderr standard error output stream to write processing data in.
      * @return {@link CommandResult} for ssh-server and return code of processed command.
      * @throws IOException
      */
-    CommandResult processCommand(byte[] command, OutputStream stdout, OutputStream stderr) throws IOException;
+    CommandResult processCommand(byte[] command) throws IOException;
 
 }

ssh-server/ssh-server/src/main/java/itx/ssh/server/commands/OutputWriter.java

@@ -0,0 +1,45 @@
+package itx.ssh.server.commands;
+
+import java.io.IOException;
+
+/**
+ * Proxy for writing data into stdout and stderr output streams.
+ */
+public interface OutputWriter {
+
+    /**
+     * Write byte into stdout and flush.
+     * @param b data to write.
+     * @throws IOException
+     */
+    void writeStdOutAndFlush(byte b) throws IOException;
+
+    /**
+     * Write bytes into stdout and flush.
+     * @param bytes data to write.
+     * @throws IOException
+     */
+    void writeStdOutAndFlush(byte[] bytes) throws IOException;
+
+    /**
+     * Write byte into stdout and flush.
+     * @param b data to write.
+     * @throws IOException
+     */
+    void writeStdErrAndFlush(byte b) throws IOException;
+
+    /**
+     * Write bytes into stdout and flush.
+     * @param bytes data to write.
+     * @throws IOException
+     */
+    void writeStdErrAndFlush(byte[] bytes) throws IOException;
+
+    /**
+     * Write message bytes into stdout appends character 13 (ENTER) and flush.
+     * @param bytes message payload to write.
+     * @throws IOException
+     */
+    void writeMessage(byte[] bytes) throws IOException;
+
+}

ssh-server/ssh-server/src/main/java/itx/ssh/server/commands/keymaps/DefaultKeyMap.java

@@ -4,93 +4,93 @@
 
 public class DefaultKeyMap implements KeyMap {
 
-    final private static int ENTER = 13;
-    final private static int BACKSPACE = 127;
-    final private static int SEQUENCE_PREFIX = 27;
-    final private static int ARROW_PREFIX = 91;
-    final private static int[] ARROW_LEFT_SEQUENCE = { SEQUENCE_PREFIX, ARROW_PREFIX, 68 };
-    final private static int[] ARROW_RIGHT_SEQUENCE = { SEQUENCE_PREFIX, ARROW_PREFIX, 67 };
-    final private static int[] DELETE_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 51 };
-    final private static int[] HOME_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 72 };
-    final private static int[] END_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 70 };
+    final private static byte ENTER = 13;
+    final private static byte BACKSPACE = 127;
+    final private static byte SEQUENCE_PREFIX = 27;
+    final private static byte ARROW_PREFIX = 91;
+    final private static byte[] ARROW_LEFT_SEQUENCE = { SEQUENCE_PREFIX, ARROW_PREFIX, 68 };
+    final private static byte[] ARROW_RIGHT_SEQUENCE = { SEQUENCE_PREFIX, ARROW_PREFIX, 67 };
+    final private static byte[] DELETE_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 51 };
+    final private static byte[] HOME_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 72 };
+    final private static byte[] END_SEQUENCE = {SEQUENCE_PREFIX, ARROW_PREFIX, 70 };
 
     @Override
-    public int getEnterKeyCode() {
+    public byte getEnterKeyCode() {
         return ENTER;
     }
 
     @Override
-    public int getBackSpaceKeyCode() {
+    public byte getBackSpaceKeyCode() {
         return BACKSPACE;
     }
 
     @Override
-    public int getSequencePrefix() {
+    public byte getSequencePrefix() {
         return SEQUENCE_PREFIX;
     }
 
     @Override
-    public boolean isKeyLeftSequence(int... sequence) {
+    public boolean isKeyLeftSequence(byte... sequence) {
         return sequence != null && sequence.length >= 3 &&
                 sequence[0] == ARROW_LEFT_SEQUENCE[0] &&
                 sequence[1] == ARROW_LEFT_SEQUENCE[1] &&
                 sequence[2] == ARROW_LEFT_SEQUENCE[2];
     }
 
     @Override
-    public boolean isKeyRightSequence(int... sequence) {
+    public boolean isKeyRightSequence(byte... sequence) {
         return sequence != null && sequence.length >= 3 &&
                 sequence[0] == ARROW_RIGHT_SEQUENCE[0] &&
                 sequence[1] == ARROW_RIGHT_SEQUENCE[1] &&
                 sequence[2] == ARROW_RIGHT_SEQUENCE[2];
     }
 
     @Override
-    public boolean isKeyDeleteSequence(int... sequence) {
+    public boolean isKeyDeleteSequence(byte... sequence) {
         return sequence != null && sequence.length >= 3 &&
                 sequence[0] == DELETE_SEQUENCE[0] &&
                 sequence[1] == DELETE_SEQUENCE[1] &&
                 sequence[2] == DELETE_SEQUENCE[2];
     }
 
     @Override
-    public boolean isKeyHomeSequence(int... sequence) {
+    public boolean isKeyHomeSequence(byte... sequence) {
         return sequence != null && sequence.length >= 3 &&
                 sequence[0] == HOME_SEQUENCE[0] &&
                 sequence[1] == HOME_SEQUENCE[1] &&
                 sequence[2] == HOME_SEQUENCE[2];
     }
 
     @Override
-    public boolean isKeyEndSequence(int... sequence) {
+    public boolean isKeyEndSequence(byte... sequence) {
         return sequence != null && sequence.length >= 3 &&
                 sequence[0] == END_SEQUENCE[0] &&
                 sequence[1] == END_SEQUENCE[1] &&
                 sequence[2] == END_SEQUENCE[2];
     }
 
     @Override
-    public int[] getKeyLeftSequence() {
+    public byte[] getKeyLeftSequence() {
         return Arrays.copyOf(ARROW_LEFT_SEQUENCE, ARROW_LEFT_SEQUENCE.length);
     }
 
     @Override
-    public int[] getKeyRightSequence() {
+    public byte[] getKeyRightSequence() {
         return Arrays.copyOf(ARROW_RIGHT_SEQUENCE, ARROW_RIGHT_SEQUENCE.length);
     }
 
     @Override
-    public int[] getKeyDeleteSequence() {
+    public byte[] getKeyDeleteSequence() {
         return Arrays.copyOf(DELETE_SEQUENCE, DELETE_SEQUENCE.length);
     }
 
     @Override
-    public int[] getKeyHomeSequence() {
+    public byte[] getKeyHomeSequence() {
         return Arrays.copyOf(HOME_SEQUENCE, HOME_SEQUENCE.length);
     }
 
     @Override
-    public int[] getKeyEndSequence() {
+    public byte[] getKeyEndSequence() {
         return Arrays.copyOf(END_SEQUENCE, END_SEQUENCE.length);
     }
 

ssh-server/ssh-server/src/main/java/itx/ssh/server/commands/keymaps/KeyMap.java

@@ -2,30 +2,30 @@
 
 public interface KeyMap {
 
-    int getEnterKeyCode();
+    byte getEnterKeyCode();
 
-    int getBackSpaceKeyCode();
+    byte getBackSpaceKeyCode();
 
-    int getSequencePrefix();
+    byte getSequencePrefix();
 
-    boolean isKeyLeftSequence(int... sequence);
+    boolean isKeyLeftSequence(byte... sequence);
 
-    boolean isKeyRightSequence(int... sequence);
+    boolean isKeyRightSequence(byte... sequence);
 
-    boolean isKeyDeleteSequence(int... sequence);
+    boolean isKeyDeleteSequence(byte... sequence);
 
-    boolean isKeyHomeSequence(int... sequence);
+    boolean isKeyHomeSequence(byte... sequence);
 
-    boolean isKeyEndSequence(int... sequence);
+    boolean isKeyEndSequence(byte... sequence);
 
-    int[] getKeyLeftSequence();
+    byte[] getKeyLeftSequence();
 
-    int[] getKeyRightSequence();
+    byte[] getKeyRightSequence();
 
-    int[] getKeyDeleteSequence();
+    byte[] getKeyDeleteSequence();
 
-    int[] getKeyHomeSequence();
+    byte[] getKeyHomeSequence();
 
-    int[] getKeyEndSequence();
+    byte[] getKeyEndSequence();
 
 }

ssh-server/ssh-server/src/main/java/itx/ssh/server/commands/repl/REPLCommand.java

@@ -1,8 +1,10 @@
 package itx.ssh.server.commands.repl;
 
 import itx.ssh.server.commands.CommandProcessor;
+import itx.ssh.server.commands.OutputWriter;
 import itx.ssh.server.commands.keymaps.KeyMap;
 import itx.ssh.server.commands.subsystem.SshClientSessionCounter;
+import itx.ssh.server.utils.OutputWriterImpl;
 import org.apache.sshd.server.Environment;
 import org.apache.sshd.server.ExitCallback;
 import org.apache.sshd.server.command.Command;
@@ -64,10 +66,11 @@ public void setExitCallback(ExitCallback callback) {
     @Override
     public void start(Environment env) throws IOException {
         long sessionId = sshClientSessionCounter.getNewSessionId();
-        commandProcessor.updateSessionId(sessionId);
+        OutputWriterImpl outputWriter = new OutputWriterImpl(stdout, stderr);
+        commandProcessor.onSessionStart(sessionId, outputWriter);
         LOG.info("start REPL command processor with sessionId: {}", sessionId);
         REPLCommandProcessor replCommandProcessor = new REPLCommandProcessor(prompt, keyMap, commandProcessor,
-                stdin, stdout, stderr, exitCallback);
+                stdin, outputWriter, exitCallback);
         executorService.submit(replCommandProcessor);
     }