Callback streams for use as library

src/arena.rs

@@ -181,6 +181,8 @@ pub enum ArenaHeaderTag {
     ReadlineStream = 0b110000,
     StaticStringStream = 0b110100,
     ByteStream = 0b111000,
+    CallbackStream = 0b111001,
+    InputChannelStream = 0b111010,
     StandardOutputStream = 0b1100,
     StandardErrorStream = 0b11000,
     NullStream = 0b111100,
@@ -841,6 +843,12 @@ unsafe fn drop_slab_in_place(value: NonNull<AllocSlab>, tag: ArenaHeaderTag) {
         ArenaHeaderTag::ByteStream => {
             drop_typed_slab_in_place!(ByteStream, value);
         }
+        ArenaHeaderTag::CallbackStream => {
+            drop_typed_slab_in_place!(CallbackStream, value);
+        }
+        ArenaHeaderTag::InputChannelStream => {
+            drop_typed_slab_in_place!(InputChannelStream, value);
+        }
         ArenaHeaderTag::LiveLoadState | ArenaHeaderTag::InactiveLoadState => {
             drop_typed_slab_in_place!(LiveLoadState, value);
         }

src/machine/config.rs

@@ -1,12 +1,16 @@
-use std::borrow::Cow;
+use std::cell::RefCell;
+use std::io::{Seek, SeekFrom, Write};
+use std::rc::Rc;
+use std::{borrow::Cow, io::Cursor};
 
 use rand::{rngs::StdRng, SeedableRng};
 
 use crate::Machine;
 
 use super::{
     bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Atom,
-    CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream, StreamOptions,
+    Callback, CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream,
+    StreamOptions,
 };
 
 /// Describes how the streams of a [`Machine`](crate::Machine) will be handled.
@@ -31,13 +35,60 @@ impl StreamConfig {
             inner: StreamConfigInner::Memory,
         }
     }
+
+    /// Calls the given callbacks when the respective streams are written to.
+    ///
+    /// This also returns a handler to the stdin do the [`Machine`](crate::Machine).
+    pub fn with_callbacks(stdout: Option<Callback>, stderr: Option<Callback>) -> (UserInput, Self) {
+        let stdin = Rc::new(RefCell::new(Cursor::new(Vec::new())));
+        (
+            UserInput {
+                inner: stdin.clone(),
+            },
+            StreamConfig {
+                inner: StreamConfigInner::Callbacks {
+                    stdin,
+                    stdout,
+                    stderr,
+                },
+            },
+        )
+    }
+}
+
+/// A handler for the stdin of the [`Machine`](crate::Machine).
+#[derive(Debug)]
+pub struct UserInput {
+    inner: Rc<RefCell<Cursor<Vec<u8>>>>,
+}
+
+impl Write for UserInput {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let mut inner = self.inner.borrow_mut();
+        let pos = inner.position();
+
+        inner.seek(SeekFrom::End(0))?;
+        let result = inner.write(buf);
+        inner.seek(SeekFrom::Start(pos))?;
+
+        result
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.inner.borrow_mut().flush()
+    }
 }
 
 #[derive(Default)]
 enum StreamConfigInner {
     Stdio,
     #[default]
     Memory,
+    Callbacks {
+        stdin: Rc<RefCell<Cursor<Vec<u8>>>>,
+        stdout: Option<Callback>,
+        stderr: Option<Callback>,
+    },
 }
 
 /// Describes how a [`Machine`](crate::Machine) will be configured.
@@ -90,6 +141,21 @@ impl MachineBuilder {
                 Stream::from_owned_string("".to_owned(), &mut machine_st.arena),
                 Stream::stderr(&mut machine_st.arena),
             ),
+            StreamConfigInner::Callbacks {
+                stdin,
+                stdout,
+                stderr,
+            } => (
+                Stream::input_channel(stdin, &mut machine_st.arena),
+                stdout.map_or_else(
+                    || Stream::Null(StreamOptions::default()),
+                    |x| Stream::from_callback(x, &mut machine_st.arena),
+                ),
+                stderr.map_or_else(
+                    || Stream::Null(StreamOptions::default()),
+                    |x| Stream::from_callback(x, &mut machine_st.arena),
+                ),
+            ),
         };
 
         let mut wam = Machine {

src/machine/lib_machine/tests.rs

@@ -1,5 +1,8 @@
+use std::io::Write;
+use std::{cell::RefCell, io::Read, rc::Rc};
+
 use super::*;
-use crate::MachineBuilder;
+use crate::{MachineBuilder, StreamConfig};
 
 #[test]
 #[cfg_attr(miri, ignore = "it takes too long to run")]
@@ -608,3 +611,34 @@ fn errors_and_exceptions() {
         [Ok(LeafAnswer::Exception(Term::atom("a")))]
     );
 }
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn callback_streams() {
+    let test_string = Rc::new(RefCell::new(String::new()));
+    let test_string2 = test_string.clone();
+
+    let (mut user_input, streams) = StreamConfig::with_callbacks(
+        Some(Box::new(move |x| {
+            x.read_to_string(&mut *test_string2.borrow_mut()).unwrap();
+        })),
+        None,
+    );
+    let mut machine = MachineBuilder::default().with_streams(streams).build();
+
+    write!(&mut user_input, "a(1,2,3).").unwrap();
+
+    let complete_answer: Vec<_> = machine
+        .run_query("read(A), write('asdf'), nl, flush_output.")
+        .collect();
+
+    assert_eq!(
+        complete_answer,
+        [Ok(LeafAnswer::from_bindings([(
+            "A",
+            Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3)])
+        ),]))]
+    );
+
+    assert_eq!(*test_string.borrow(), "asdf\n");
+}