added input output stream support

Author: Achal1607

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/CodeEval.java

@@ -32,6 +32,7 @@
 import org.netbeans.modules.java.lsp.server.notebook.CellExecutionResult;
 import org.netbeans.modules.java.lsp.server.notebook.NotebookCellExecutionProgressResultParams;
 import org.netbeans.modules.java.lsp.server.notebook.NotebookCellExecutionProgressResultParams.EXECUTION_STATUS;
+import org.netbeans.modules.java.lsp.server.protocol.NbCodeLanguageClient;
 import org.openide.util.RequestProcessor;
 
 /**
@@ -86,10 +87,10 @@ private String interruptCodeExecution(String notebookId) {
         try {
             JShell jshell = NotebookSessionManager.getInstance().getSession(notebookId);
             String cellId = activeCellExecutionMapping.get(notebookId);
-            if(cellId != null){
+            if (cellId != null) {
                 sendNotification(notebookId, cellId, EXECUTION_STATUS.INTERRUPTED);
             }
-            
+            flushStreams(notebookId);
             List<CompletableFuture<Boolean>> tasks = pendingTasks.get(notebookId);
             if (tasks != null) {
                 tasks.forEach(task -> {
@@ -158,7 +159,7 @@ private void codeEvalTaskRunnable(CompletableFuture<Boolean> future, JShell jshe
             }
 
             runCode(jshell, sourceCode, notebookId);
-
+            flushStreams(notebookId);
             sendNotification(notebookId, EXECUTION_STATUS.SUCCESS);
 
             future.complete(true);
@@ -240,6 +241,13 @@ private List<String> getSnippetValue(SnippetEvent event) {
         return snippetValues;
     }
 
+    private void flushStreams(String notebookId) {
+        JshellStreamsHandler streamHandler = NotebookSessionManager.getInstance().getJshellStreamsHandler(notebookId);
+        if (streamHandler != null) {
+            streamHandler.flushOutputStreams();
+        }
+    }
+
     // This method is directly taken from JShell tool implementation in jdk with some minor modifications
     private List<String> displayableDiagnostic(String source, Diag diag) {
         List<String> toDisplay = new ArrayList<>();
@@ -363,10 +371,12 @@ private void sendNotification(String notebookId, String cellId, byte[] msg, List
                 }
             }
 
-            LanguageClientInstance.getInstance().getClient().notifyNotebookCellExecutionProgress(params);
+            NbCodeLanguageClient client = LanguageClientInstance.getInstance().getClient();
+            if (client != null) {
+                client.notifyNotebookCellExecutionProgress(params);
+            }
         } catch (Exception ex) {
             LOG.log(Level.SEVERE, "Some error ocurred while sending code eval notification to the client {0}", ex.getMessage());
         }
     }
-
 }

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/CustomInputStream.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2025, Oracle and/or its affiliates.
+ *
+ * Licensed 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
+ *
+ *    https://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.netbeans.modules.nbcode.java.notebook;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.ref.WeakReference;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.netbeans.modules.java.lsp.server.input.ShowInputBoxParams;
+import org.netbeans.modules.java.lsp.server.protocol.NbCodeLanguageClient;
+
+/**
+ *
+ * @author atalati
+ */
+public class CustomInputStream extends InputStream {
+
+    private static final Logger LOG = Logger.getLogger(CustomInputStream.class.getName());
+    private ByteArrayInputStream currentStream;
+    WeakReference<NbCodeLanguageClient> client;
+    private static final String USER_PROMPT_REQUEST = "Please provide scanner input here";
+
+    public CustomInputStream(NbCodeLanguageClient client) {
+        this.client = new WeakReference<>(client);
+    }
+
+    @Override
+    public synchronized int read(byte[] b, int off, int len) throws IOException {
+        try {
+            if (client == null || client.get() == null) {
+                LOG.log(Level.WARNING, "client is null");
+                return -1;
+            }
+
+            if (currentStream == null || currentStream.available() == 0) {
+                CompletableFuture<String> future = client.get().showInputBox(new ShowInputBoxParams(USER_PROMPT_REQUEST, "", true));
+                String userInput = future.get();
+
+                if (userInput == null) {
+                    LOG.log(Level.WARNING, "User input is null");
+                    return -1;
+                }
+
+                byte[] inputBytes = (userInput + System.lineSeparator()).getBytes(StandardCharsets.UTF_8);
+                currentStream = new ByteArrayInputStream(inputBytes);
+            }
+
+            return currentStream.read(b, off, len);
+        } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
+            throw new IOException("Interrupted while waiting for user input", ex);
+        } catch (ExecutionException ex) {
+            throw new IOException("Failed to get user input", ex.getCause());
+        }
+    }
+
+    @Override
+    public int read() throws IOException {
+        byte[] oneByte = new byte[1];
+        int n = read(oneByte, 0, 1);
+        return (n == -1) ? -1 : oneByte[0] & 0xFF;
+    }
+}

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/JshellStreamsHandler.java

@@ -17,6 +17,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.PrintStream;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -35,6 +36,7 @@ public class JshellStreamsHandler implements AutoCloseable {
     private final StreamingOutputStream errStream;
     private final PrintStream printOutStream;
     private final PrintStream printErrStream;
+    private final InputStream inputStream;
 
     public JshellStreamsHandler(String notebookId, BiConsumer<String, byte[]> streamCallback) {
         this(notebookId, streamCallback, streamCallback);
@@ -52,6 +54,7 @@ public JshellStreamsHandler(String notebookId,
         this.errStream = new StreamingOutputStream(createCallback(errStreamCallback));
         this.printOutStream = new PrintStream(outStream);
         this.printErrStream = new PrintStream(errStream);
+        this.inputStream = new CustomInputStream(LanguageClientInstance.getInstance().getClient());
     }
 
     private Consumer<byte[]> createCallback(BiConsumer<String, byte[]> callback) {
@@ -66,14 +69,6 @@ public void setErrStreamCallback(BiConsumer<String, byte[]> callback) {
         errStream.setCallback(createCallback(callback));
     }
 
-    public ByteArrayOutputStream getOutStream() {
-        return outStream.getOutputStream();
-    }
-    
-    public ByteArrayOutputStream getErrStream() {
-        return errStream.getOutputStream();
-    }
-    
     public PrintStream getPrintOutStream() {
         return printOutStream;
     }
@@ -82,17 +77,31 @@ public PrintStream getPrintErrStream() {
         return printErrStream;
     }
 
+    public InputStream getInputStream() {
+        return inputStream;
+    }
+    
     public String getNotebookId() {
         return notebookId;
     }
 
+    public void flushOutputStreams() {
+        try {
+            outStream.flush();
+            errStream.flush();
+        } catch (IOException ignored) {
+            // nothing can be done
+        }
+    }
+    
     @Override
     public void close() {
         try {
             printOutStream.close();
             printErrStream.close();
             outStream.close();
             errStream.close();
+            inputStream.close();
         } catch (IOException ex) {
             LOG.log(Level.WARNING, "IOException occurred while closing the streams {0}", ex.getMessage());
         }

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/LanguageClientInstance.java

@@ -15,14 +15,15 @@
  */
 package org.netbeans.modules.nbcode.java.notebook;
 
+import java.lang.ref.WeakReference;
 import org.netbeans.modules.java.lsp.server.protocol.NbCodeLanguageClient;
 
 /**
  *
  * @author atalati
  */
 public class LanguageClientInstance {
-    private NbCodeLanguageClient client = null;
+    private WeakReference<NbCodeLanguageClient> client = null;
 
     private LanguageClientInstance() {
     }
@@ -37,10 +38,10 @@ private static class Singleton {
     }
 
     public NbCodeLanguageClient getClient(){
-        return this.client;
+        return this.client == null ? null: this.client.get();
     }
 
     public void setClient(NbCodeLanguageClient client){
-        this.client = client;
+        this.client = new WeakReference<>(client);
     }
 }

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/NotebookDocumentServiceHandlerImpl.java

@@ -85,6 +85,9 @@ public class NotebookDocumentServiceHandlerImpl implements NotebookDocumentServi
     public void didOpen(DidOpenNotebookDocumentParams params) {
         try {
             NbCodeLanguageClient client = LanguageClientInstance.getInstance().getClient();
+            if(client == null){
+                return;
+            }
             client.showStatusBarMessage(new ShowStatusMessageParams(MessageType.Info,"Intializing Java kernel for notebook."));
             NotebookSessionManager.getInstance().createSession(params.getNotebookDocument()).whenComplete((JShell jshell,Throwable t) -> {
                 if (t == null) {

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/NotebookSessionManager.java

@@ -71,13 +71,15 @@ private CompletableFuture<JShell> jshellBuilder(JshellStreamsHandler streamsHand
                 return JShell.builder()
                         .out(streamsHandler.getPrintOutStream())
                         .err(streamsHandler.getPrintErrStream())
+                        .in(streamsHandler.getInputStream())
                         .compilerOptions()
                         .remoteVMOptions()
                         .build();
             } else {
                 return JShell.builder()
                         .out(streamsHandler.getPrintOutStream())
                         .err(streamsHandler.getPrintErrStream())
+                        .in(streamsHandler.getInputStream())
                         .compilerOptions(compilerOptions.toArray(new String[0]))
                         .remoteVMOptions(remoteOptions.toArray(new String[0]))
                         .build();

nbcode/notebooks/src/org/netbeans/modules/nbcode/java/notebook/StreamingOutputStream.java

@@ -18,83 +18,97 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import org.openide.util.RequestProcessor;
+import org.openide.util.RequestProcessor.Task;
 
 /**
  *
  * @author atalati
  */
 public class StreamingOutputStream extends OutputStream {
+
     private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
     private Consumer<byte[]> callback;
-    // What if line endings are not there then how to flush Systen.out.print()?
-    private static final char NEW_LINE_ENDING = '\n';
-    private static final char NEW_LINE_ENDING_DOS = '\r';
+    private static final int MAX_BUFFER_SIZE = 1024;
+    private final AtomicBoolean isPeriodicFlushOutputStream;
+
+    static RequestProcessor getRequestProcessor() {
+        return RPSingleton.instance;
+    }
+
+    private static int getRequestPeriodicTime() {
+        return RPSingleton.PERIODIC_TIME;
+    }
+
+    private static final class RPSingleton {
+
+        private static final RequestProcessor instance = new RequestProcessor(StreamingOutputStream.class.getName(), 1, true, false);
+        private static final int PERIODIC_TIME = 100;
+    }
 
     public StreamingOutputStream(Consumer<byte[]> callback) {
         this.callback = callback;
+        this.isPeriodicFlushOutputStream = new AtomicBoolean(true);
+        createAndScheduleTask();
     }
 
     @Override
-    public void write(int b) throws IOException {
+    public synchronized void write(int b) throws IOException {
         buffer.write(b);
-        checkNewlineEndings((byte) b);
+        ifBufferOverflowFlush();
     }
 
     @Override
-    public void write(byte[] b, int off, int len) throws IOException {
+    public synchronized void write(byte[] b, int off, int len) throws IOException {
         buffer.write(b, off, len);
-        checkForNewlineAndFlush(b, off, len);
+        ifBufferOverflowFlush();
     }
 
     @Override
-    public void flush() throws IOException {
-        super.flush();
+    public synchronized void flush() throws IOException {
         flushToCallback();
     }
 
     @Override
-    public void write(byte[] b) throws IOException {
+    public synchronized void write(byte[] b) throws IOException {
         buffer.write(b);
-        checkForNewlineAndFlush(b, 0, b.length);
+        ifBufferOverflowFlush();
     }
-    
+
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
         flushToCallback();
+        isPeriodicFlushOutputStream.set(false);
         super.close();
     }
-    
-    public void setCallback(Consumer<byte[]> cb){
-        this.callback = cb;
-    }
-    
-    public ByteArrayOutputStream getOutputStream(){
-        return buffer;
-    }
 
-    private void checkForNewlineAndFlush(byte[] b, int offset, int length) {
-        for (int i = offset; i < offset + length; i++) {
-            boolean isNew = checkNewlineEndings(b[i]);
-            if (isNew) {
-                break;
-            }
-        }
+    public void setCallback(Consumer<byte[]> cb) {
+        this.callback = cb;
     }
 
-    private boolean checkNewlineEndings(byte b) {
-        if (b == NEW_LINE_ENDING || b == NEW_LINE_ENDING_DOS) {
+    private void ifBufferOverflowFlush() {
+        if (buffer.size() > MAX_BUFFER_SIZE) {
             flushToCallback();
-            return true;
         }
-        return false;
     }
 
-    private void flushToCallback() {
+    private synchronized void flushToCallback() {
         if (buffer.size() > 0) {
             byte[] output = buffer.toByteArray();
             buffer.reset();
             callback.accept(output);
         }
     }
+
+    private void createAndScheduleTask() {
+        if (isPeriodicFlushOutputStream.get()) {
+            Task task = getRequestProcessor().create(() -> {
+                flushToCallback();
+                createAndScheduleTask();
+            });
+            task.schedule(getRequestPeriodicTime());
+        }
+    }
 }