This is an automated email from the ASF dual-hosted git repository.
tbonelee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 2275a5d8c3 [ZEPPELIN-6705] Refactor shell terminal session management
2275a5d8c3 is described below
commit 2275a5d8c367c534980a2ff73cc834a1d9f46508
Author: SeungYoung Oh <[email protected]>
AuthorDate: Sun Sep 13 15:49:55 2026 +0900
[ZEPPELIN-6705] Refactor shell terminal session management
### What is this PR for?
This PR refactors `TerminalService` into `TerminalSession` to clarify its
role.
The name `~Service` suggests a long-lived component providing shared
functionality. However, each instance owns the state and resources of a single
terminal connection and must be closed when that connection closes. `~Session`
better reflects this ownership and lifecycle. This mismatch may have
contributed to the missing cleanup in #5447.
This PR also includes minor code cleanup.
### What type of PR is it?
Refactoring
### Todos
### What is the Jira issue?
- [ZEPPELIN-6705]
### How should this be tested?
- Run the shell module tests: `./mvnw test -pl shell.`
### Screenshots (if appropriate)
### Questions:
* Does the license files need to update? N
* Is there breaking changes for older versions? N
* Does this needs documentation? N
Closes #5473 from seung-00/ZEPPELIN-6705.
Signed-off-by: ChanHo Lee <[email protected]>
---
.../zeppelin/shell/terminal/TerminalManager.java | 32 ++--
.../TerminalService.java => TerminalSession.java} | 174 ++++++++++-----------
.../shell/terminal/websocket/TerminalSocket.java | 42 ++---
.../zeppelin/shell/TerminalInterpreterTest.java | 15 +-
.../shell/terminal/TerminalSessionTest.java | 128 +++++++++++++++
5 files changed, 251 insertions(+), 140 deletions(-)
diff --git
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java
index 71d9ff1279..92cbb30a85 100644
---
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java
+++
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java
@@ -18,13 +18,13 @@
package org.apache.zeppelin.shell.terminal;
import org.apache.zeppelin.interpreter.InterpreterContext;
-import org.apache.zeppelin.shell.terminal.service.TerminalService;
import org.apache.zeppelin.shell.terminal.websocket.TerminalSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import static
org.apache.zeppelin.shell.TerminalInterpreter.TERMINAL_SOCKET_CLOSE;
import static
org.apache.zeppelin.shell.TerminalInterpreter.TERMINAL_SOCKET_CONNECT;
@@ -35,8 +35,7 @@ import static
org.apache.zeppelin.shell.TerminalInterpreter.TERMINAL_SOCKET_STAT
public class TerminalManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(TerminalManager.class);
- // TerminalSocket hashCode -> TerminalService
- private HashMap<Integer, TerminalService> terminalSocket2Service;
+ private final Map<TerminalSocket, TerminalSession> terminalSocket2Session;
// NoteId@ParagraphId -> InterpreterContext
private HashMap<String, InterpreterContext> noteParagraphId2IntpContext;
@@ -51,27 +50,18 @@ public class TerminalManager {
}
private TerminalManager() {
- terminalSocket2Service = new HashMap<>();
+ terminalSocket2Session = new ConcurrentHashMap<>();
noteParagraphId2IntpContext = new HashMap<>();
}
- public TerminalService addTerminalService(TerminalSocket terminalSocket) {
- Integer terminalSocketHashcode = terminalSocket.hashCode();
- if (terminalSocket2Service.containsKey(terminalSocketHashcode)) {
- return terminalSocket2Service.get(terminalSocketHashcode);
- } else {
- TerminalService terminalService = new TerminalService();
- terminalSocket2Service.put(terminalSocketHashcode, terminalService);
- return terminalService;
- }
+ public void addTerminalSession(TerminalSocket terminalSocket,
TerminalSession terminalSession) {
+ terminalSocket2Session.put(terminalSocket, terminalSession);
}
- public void removeTerminalService(TerminalSocket terminalSocket) {
- Integer terminalSocketHashcode = terminalSocket.hashCode();
- if (terminalSocket2Service.containsKey(terminalSocketHashcode)) {
- terminalSocket2Service.remove(terminalSocketHashcode).close();
- } else {
- LOGGER.error("Can't find TerminalSocket: {}", terminalSocketHashcode);
+ public void removeTerminalSession(TerminalSocket terminalSocket) {
+ TerminalSession session = terminalSocket2Session.remove(terminalSocket);
+ if (session != null) {
+ session.close();
}
}
@@ -87,7 +77,7 @@ public class TerminalManager {
}
public void runCommand(String command) {
- for (Map.Entry<Integer, TerminalService> entry :
terminalSocket2Service.entrySet()) {
+ for (Map.Entry<TerminalSocket, TerminalSession> entry :
terminalSocket2Session.entrySet()) {
entry.getValue().onCommand(command + "\r");
}
}
@@ -122,7 +112,7 @@ public class TerminalManager {
LOGGER.error("Can't find InterpreterContext from : {}", id);
}
- removeTerminalService(terminalSocket);
+ removeTerminalSession(terminalSocket);
}
// Socket error will trigger socket close
diff --git
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/service/TerminalService.java
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalSession.java
similarity index 52%
rename from
shell/src/main/java/org/apache/zeppelin/shell/terminal/service/TerminalService.java
rename to
shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalSession.java
index 609dbe91a6..7605fbeb55 100644
---
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/service/TerminalService.java
+++
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalSession.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.zeppelin.shell.terminal.service;
+package org.apache.zeppelin.shell.terminal;
import com.google.gson.Gson;
import com.pty4j.PtyProcess;
@@ -28,76 +28,107 @@ import org.slf4j.LoggerFactory;
import jakarta.websocket.Session;
import java.io.BufferedReader;
import java.io.BufferedWriter;
+import java.io.Closeable;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
-public class TerminalService implements AutoCloseable {
- private static final Logger LOGGER =
LoggerFactory.getLogger(TerminalService.class);
+public class TerminalSession implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TerminalSession.class);
+ private static final Gson gson = new Gson();
+
+ private static final int DEFAULT_COLUMNS = 20;
+ private static final int DEFAULT_ROWS = 10;
+ private final AtomicBoolean closed = new AtomicBoolean();
- private String[] termCommand;
private PtyProcess process;
- private Integer columns = 20;
- private Integer rows = 10;
private BufferedReader inputReader;
private BufferedReader errorReader;
private BufferedWriter outputWriter;
private Session webSocketSession;
- private LinkedBlockingQueue<String> commandQueue = new
LinkedBlockingQueue<>();
+ private final ExecutorService commandExecutor =
Executors.newSingleThreadExecutor();
+ private final ExecutorService readerExecutor =
Executors.newFixedThreadPool(2);
- public void onTerminalReady() {
- TerminalService.startThread(() -> {
- try {
- initializeProcess();
- } catch (Exception e) {
- LOGGER.error(e.getMessage(), e);
- }
- });
+ public TerminalSession(Session webSocketSession) throws IOException {
+ this(webSocketSession, startProcess());
}
- private void initializeProcess() throws Exception {
- LOGGER.info("initialize TerminalService Process");
+ TerminalSession(Session webSocketSession, PtyProcess process) {
+ this.process = process;
+ this.webSocketSession = webSocketSession;
+ try {
+ webSocketSession.setMaxIdleTimeout(60 * 60 * 1000);
+ inputReader = new BufferedReader(new
InputStreamReader(process.getInputStream()));
+ errorReader = new BufferedReader(new
InputStreamReader(process.getErrorStream()));
+ outputWriter = new BufferedWriter(new
OutputStreamWriter(process.getOutputStream()));
+ readerExecutor.execute(() -> printReader(inputReader));
+ readerExecutor.execute(() -> printReader(errorReader));
+ } catch (RuntimeException e) {
+ close();
+ throw e;
+ }
+ }
+ private static PtyProcess startProcess() throws IOException {
boolean isWindows = System.getProperty("os.name").startsWith("Windows");
+ String[] termCommand;
if (isWindows) {
- this.termCommand = "cmd.exe".split("\\s+");
+ termCommand = "cmd.exe".split("\\s+");
} else {
- this.termCommand = "/bin/bash -i".split("\\s+");
+ termCommand = "/bin/bash -i".split("\\s+");
}
Map<String, String> envs = new HashMap<>(System.getenv());
envs.put("TERM", "xterm");
- this.process = new
PtyProcessBuilder().setCommand(termCommand).setEnvironment(envs).start();
-
- process.setWinSize(new WinSize(columns, rows));
- this.inputReader = new BufferedReader(new
InputStreamReader(process.getInputStream()));
- this.errorReader = new BufferedReader(new
InputStreamReader(process.getErrorStream()));
- this.outputWriter = new BufferedWriter(new
OutputStreamWriter(process.getOutputStream()));
-
- TerminalService.startThread(() -> {
- printReader(inputReader);
- });
+ return new PtyProcessBuilder()
+ .setCommand(termCommand)
+ .setEnvironment(envs)
+ .setInitialColumns(DEFAULT_COLUMNS)
+ .setInitialRows(DEFAULT_ROWS)
+ .start();
+ }
- TerminalService.startThread(() -> {
- printReader(errorReader);
- });
+ public void onCommand(String command) {
+ if (StringUtils.isEmpty(command)) {
+ return;
+ }
- process.waitFor();
+ try {
+ commandExecutor.execute(() -> {
+ try {
+ outputWriter.write(command);
+ outputWriter.flush();
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ if (!commandExecutor.isShutdown()) {
+ throw e;
+ }
+ }
}
- private void print(String text) throws IOException {
- Map<String, String> map = new HashMap<>();
- map.put("type", "TERMINAL_PRINT");
- map.put("text", text);
+ public void onTerminalResize(String columns, String rows) {
+ if (Objects.nonNull(columns) && Objects.nonNull(rows)) {
+ if (!closed.get()) {
+ process.setWinSize(new WinSize(Integer.parseInt(columns),
Integer.parseInt(rows)));
+ }
+ }
+ }
- Gson gson = new Gson();
+ private synchronized void print(String text) throws IOException {
+ Map<String, String> map = Map.of("type", "TERMINAL_PRINT", "text", text);
String message = gson.toJson(map);
webSocketSession.getBasicRemote().sendText(message);
}
@@ -117,56 +148,23 @@ public class TerminalService implements AutoCloseable {
}
}
- public void onCommand(String command) {
- if (null == command || StringUtils.isEmpty(command)) {
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
return;
}
-
- try {
- commandQueue.put(command);
- } catch (InterruptedException e) {
- LOGGER.error(e.getMessage(), e);
- }
-
- TerminalService.startThread(() -> {
+ commandExecutor.shutdownNow();
+ readerExecutor.shutdownNow();
+ destroyProcess(process);
+ for (Closeable stream : new Closeable[] {outputWriter, inputReader,
errorReader}) {
+ if (stream == null) {
+ continue;
+ }
try {
- outputWriter.write(commandQueue.poll());
- outputWriter.flush();
+ stream.close();
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
}
- });
- }
-
- public void onTerminalResize(String columns, String rows) {
- if (Objects.nonNull(columns) && Objects.nonNull(rows)) {
- this.columns = Integer.valueOf(columns);
- this.rows = Integer.valueOf(rows);
-
- if (Objects.nonNull(process)) {
- process.setWinSize(new WinSize(this.columns, this.rows));
- }
- }
- }
-
- @Override
- public void close() {
- try {
- if (outputWriter != null) {
- outputWriter.close();
- }
- if (inputReader != null) {
- inputReader.close();
- }
- if (errorReader != null) {
- errorReader.close();
- }
- } catch (IOException e) {
- LOGGER.error(e.getMessage(), e);
- }
-
- if (process != null) {
- destroyProcess(process);
}
}
@@ -181,14 +179,4 @@ public class TerminalService implements AutoCloseable {
process.destroyForcibly();
}
}
-
- public void onWebSocketConnect(Session webSocketSession) {
- this.webSocketSession = webSocketSession;
- webSocketSession.setMaxIdleTimeout(60 * 60 * 1000);
- }
-
- public static void startThread(Runnable runnable) {
- Thread thread = new Thread(runnable);
- thread.start();
- }
}
diff --git
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java
index 9a9a7f497e..476c6e632b 100644
---
a/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java
+++
b/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java
@@ -20,11 +20,10 @@ package org.apache.zeppelin.shell.terminal.websocket;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.zeppelin.shell.terminal.TerminalManager;
-import org.apache.zeppelin.shell.terminal.service.TerminalService;
+import org.apache.zeppelin.shell.terminal.TerminalSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import jakarta.websocket.ClientEndpoint;
import jakarta.websocket.CloseReason;
import jakarta.websocket.EndpointConfig;
import jakarta.websocket.OnClose;
@@ -39,7 +38,6 @@ import java.security.MessageDigest;
import java.util.List;
import java.util.Map;
-@ClientEndpoint
@ServerEndpoint(value = "/")
public class TerminalSocket {
private static final Logger LOGGER =
LoggerFactory.getLogger(TerminalSocket.class);
@@ -47,17 +45,16 @@ public class TerminalSocket {
// Key under which TerminalThread publishes the per-server auth token
public static final String AUTH_TOKEN_PROPERTY =
"zeppelin.terminal.auth.token";
- private TerminalService terminalService;
+ private static final Gson gson = new Gson();
+
+ private TerminalSession terminalSession;
private TerminalManager terminalManager = TerminalManager.getInstance();
+ private Session webSocketSession;
private String noteId;
private String paragraphId;
private volatile boolean authorized = false;
- public TerminalSocket() {
- terminalService = terminalManager.addTerminalService(this);
- }
-
@OnOpen
public void onWebSocketConnect(Session sess, EndpointConfig config) {
// This endpoint hands out an OS shell: require the per-server secret that
@@ -77,7 +74,7 @@ public class TerminalSocket {
}
authorized = true;
LOGGER.info("Socket Connected: {}", sess.getId());
- terminalService.onWebSocketConnect(sess);
+ this.webSocketSession = sess;
}
@OnMessage
@@ -96,16 +93,28 @@ public class TerminalSocket {
String type = messageMap.get("type");
switch (type) {
case "TERMINAL_READY":
- terminalService.onTerminalReady();
this.noteId = messageMap.get("noteId");
this.paragraphId = messageMap.get("paragraphId");
+ if (terminalSession == null) {
+ try {
+ terminalSession = new TerminalSession(webSocketSession);
+ terminalManager.addTerminalSession(this, terminalSession);
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ break;
+ }
+ }
terminalManager.onWebSocketConnect(noteId, paragraphId);
break;
case "TERMINAL_COMMAND":
- terminalService.onCommand(messageMap.get("command"));
+ if (terminalSession != null) {
+ terminalSession.onCommand(messageMap.get("command"));
+ }
break;
case "TERMINAL_RESIZE":
- terminalService.onTerminalResize(messageMap.get("columns"),
messageMap.get("rows"));
+ if (terminalSession != null) {
+ terminalSession.onTerminalResize(messageMap.get("columns"),
messageMap.get("rows"));
+ }
break;
default:
LOGGER.error("Unrecognized action: {}", message);
@@ -116,10 +125,8 @@ public class TerminalSocket {
@OnClose
public void onWebSocketClose(CloseReason reason) {
LOGGER.info("Socket Closed: {}", reason);
- if (authorized && noteId != null && paragraphId != null) {
+ if (authorized && terminalSession != null) {
terminalManager.onWebSocketClose(this, noteId, paragraphId);
- } else {
- terminalManager.removeTerminalService(this);
}
}
@@ -142,9 +149,6 @@ public class TerminalSocket {
}
private Map<String, String> getMessageMap(String message) {
- Gson gson = new Gson();
- Map<String, String> map = gson.fromJson(message,
- new TypeToken<Map<String, String>>(){}.getType());
- return map;
+ return gson.fromJson(message, new TypeToken<Map<String,
String>>(){}.getType());
}
}
diff --git
a/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java
b/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java
index 646029a522..14ba94e5cc 100644
--- a/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java
+++ b/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java
@@ -47,7 +47,8 @@ import static org.junit.jupiter.api.Assertions.fail;
import java.io.IOException;
import java.net.URI;
import java.util.Properties;
-import java.util.regex.Pattern;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
class TerminalInterpreterTest extends BaseInterpreterTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(TerminalInterpreterTest.class);
@@ -113,9 +114,9 @@ class TerminalInterpreterTest extends BaseInterpreterTest {
LOGGER.info(msg);
// {"text":"\u001b[?1034hbash-3.2$ \r\u001b[Kbash-3.2$
// \r\u001b[Kbash-3.2$ ","type":"TERMINAL_PRINT"}
- String pattern = "\\{\"text\":\".*\"type\":\"TERMINAL_PRINT\"}";
- boolean isMatch = Pattern.matches(pattern, msg);
- assertTrue(isMatch);
+ JsonObject message = JsonParser.parseString(msg).getAsJsonObject();
+ assertEquals("TERMINAL_PRINT", message.get("type").getAsString());
+ assertTrue(message.get("text").isJsonPrimitive());
// Send invalid_command message
String echoHelloWorldCmd =
String.format("{\"type\":\"TERMINAL_COMMAND\"," +
@@ -199,9 +200,9 @@ class TerminalInterpreterTest extends BaseInterpreterTest {
LOGGER.info(msg);
// {"text":"\u001b[?1034hbash-3.2$ \r\u001b[Kbash-3.2$
// \r\u001b[Kbash-3.2$ ","type":"TERMINAL_PRINT"}
- String pattern = "\\{\"text\":\".*\"type\":\"TERMINAL_PRINT\"}";
- boolean isMatch = Pattern.matches(pattern, msg);
- assertTrue(isMatch);
+ JsonObject message = JsonParser.parseString(msg).getAsJsonObject();
+ assertEquals("TERMINAL_PRINT", message.get("type").getAsString());
+ assertTrue(message.get("text").isJsonPrimitive());
// Send echo 'hello world!' message
String echoHelloWorldCmd =
String.format("{\"type\":\"TERMINAL_COMMAND\"," +
diff --git
a/shell/src/test/java/org/apache/zeppelin/shell/terminal/TerminalSessionTest.java
b/shell/src/test/java/org/apache/zeppelin/shell/terminal/TerminalSessionTest.java
new file mode 100644
index 0000000000..49cc5c6a74
--- /dev/null
+++
b/shell/src/test/java/org/apache/zeppelin/shell/terminal/TerminalSessionTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.zeppelin.shell.terminal;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.never;
+
+import com.pty4j.PtyProcess;
+import jakarta.websocket.Session;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class TerminalSessionTest {
+ @Test
+ void commandsUseOneWorkerInOrderAndAllWorkersStopOnClose() throws Exception {
+ CountDownLatch reads = new CountDownLatch(2);
+ Set<Thread> readerWorkers = ConcurrentHashMap.newKeySet();
+ InputStream input = mock(InputStream.class);
+ when(input.read(any(byte[].class), anyInt(),
anyInt())).thenAnswer(invocation -> {
+ readerWorkers.add(Thread.currentThread());
+ reads.countDown();
+ return -1;
+ });
+ CountDownLatch writes = new CountDownLatch(100);
+ Set<Thread> workers = new HashSet<>();
+ ByteArrayOutputStream output = new ByteArrayOutputStream() {
+ @Override
+ public synchronized void write(byte[] bytes, int offset, int length) {
+ workers.add(Thread.currentThread());
+ super.write(bytes, offset, length);
+ writes.countDown();
+ }
+ };
+ PtyProcess process = mock(PtyProcess.class);
+ when(process.getInputStream()).thenReturn(input);
+ when(process.getErrorStream()).thenReturn(input);
+ when(process.getOutputStream()).thenReturn(output);
+ when(process.waitFor(5L, TimeUnit.SECONDS)).thenReturn(true);
+
+ StringBuilder expected = new StringBuilder();
+ TerminalSession terminal = new TerminalSession(mock(Session.class),
process);
+ try {
+ assertTrue(reads.await(5, TimeUnit.SECONDS));
+ assertEquals(2, readerWorkers.size());
+ terminal.onCommand(null);
+ terminal.onCommand("");
+ for (int i = 0; i < 100; i++) {
+ String command = "command-" + i + "\r";
+ expected.append(command);
+ terminal.onCommand(command);
+ }
+ assertTrue(writes.await(5, TimeUnit.SECONDS));
+ assertEquals(expected.toString(), output.toString());
+ assertEquals(1, workers.size());
+ } finally {
+ terminal.close();
+ }
+
+ assertDoesNotThrow(() -> terminal.onCommand("after close"));
+ terminal.close();
+ verify(process).destroy();
+ verify(process, never()).destroyForcibly();
+ workers.addAll(readerWorkers);
+ for (Thread worker : workers) {
+ worker.join(5000);
+ assertFalse(worker.isAlive());
+ }
+ assertEquals(expected.toString(), output.toString());
+ }
+
+ @Test
+ void closeForciblyDestroysProcessThatDoesNotExit() throws Exception {
+ PtyProcess process = mock(PtyProcess.class);
+ when(process.getInputStream()).thenReturn(InputStream.nullInputStream());
+ when(process.getErrorStream()).thenReturn(InputStream.nullInputStream());
+ when(process.getOutputStream()).thenReturn(new ByteArrayOutputStream());
+ TerminalSession terminal = new TerminalSession(mock(Session.class),
process);
+
+ terminal.close();
+ terminal.close();
+
+ verify(process).destroy();
+ verify(process).destroyForcibly();
+ }
+
+ @Test
+ void initializationFailureDestroysProcess() throws Exception {
+ PtyProcess process = mock(PtyProcess.class);
+ when(process.getInputStream()).thenThrow(new IllegalStateException("stream
unavailable"));
+ when(process.waitFor(5L, TimeUnit.SECONDS)).thenReturn(true);
+
+ assertThrows(IllegalStateException.class,
+ () -> new TerminalSession(mock(Session.class), process));
+
+ verify(process).destroy();
+ }
+
+}
