More stream tests

src/machine/config.rs

@@ -9,7 +9,6 @@ use crate::Machine;
 use super::{
     bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Arena, Atom,
     Callback, CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream,
-    StreamOptions,
 };
 
 #[derive(Default)]
@@ -77,14 +76,19 @@ impl OutputStreamConfig {
     }
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug)]
 enum InputStreamConfigInner {
-    #[default]
+    String(String),
-    Null,
     Stdin,
     Channel(Receiver<Vec<u8>>),
 }
 
+impl Default for InputStreamConfigInner {
+    fn default() -> Self {
+        Self::String("".into())
+    }
+}
+
 /// Configuration for an input stream;
 #[derive(Debug, Default)]
 pub struct InputStreamConfig {
@@ -92,10 +96,10 @@ pub struct InputStreamConfig {
 }
 
 impl InputStreamConfig {
-    /// Ignores all input.
+    /// Gets input from string.
-    pub fn null() -> Self {
+    pub fn string(s: impl Into<String>) -> Self {
         Self {
-            inner: InputStreamConfigInner::Null,
+            inner: InputStreamConfigInner::String(s.into()),
         }
     }
 
@@ -119,7 +123,7 @@ impl InputStreamConfig {
 
     fn into_stream(self, arena: &mut Arena, add_history: bool) -> Stream {
         match self.inner {
-            InputStreamConfigInner::Null => Stream::Null(StreamOptions::default()),
+            InputStreamConfigInner::String(s) => Stream::from_owned_string(s, arena),
             InputStreamConfigInner::Stdin => Stream::stdin(arena, add_history),
             InputStreamConfigInner::Channel(channel) => Stream::input_channel(channel, arena),
         }
@@ -152,14 +156,12 @@ impl StreamConfig {
         }
     }
 
-    /// Binds the output stream to a memory buffer, and the error stream to stderr.
+    /// Binds the output and error streams to memory buffers and has an empty input.
-    ///
-    /// The input stream is ignored.
     pub fn in_memory() -> Self {
         StreamConfig {
-            stdin: InputStreamConfig::null(),
+            stdin: InputStreamConfig::string(""),
             stdout: OutputStreamConfig::memory(),
-            stderr: OutputStreamConfig::stderr(),
+            stderr: OutputStreamConfig::memory(),
         }
     }
 

src/machine/lib_machine/mod.rs

@@ -6,11 +6,13 @@ use crate::heap_iter::{stackful_post_order_iter, NonListElider};
 use crate::machine::machine_indices::VarKey;
 use crate::machine::mock_wam::CompositeOpDir;
 use crate::machine::{
-    F64Offset, F64Ptr, Fixnum, Number, BREAK_FROM_DISPATCH_LOOP_LOC, LIB_QUERY_SUCCESS,
+    ArenaHeaderTag, F64Offset, F64Ptr, Fixnum, Number, BREAK_FROM_DISPATCH_LOOP_LOC,
+    LIB_QUERY_SUCCESS,
 };
 use crate::parser::ast::{Var, VarPtr};
 use crate::parser::parser::{Parser, Tokens};
 use crate::read::{write_term_to_heap, TermWriteResult};
+use crate::types::UntypedArenaPtr;
 
 use dashu::{Integer, Rational};
 use indexmap::IndexMap;
@@ -280,11 +282,32 @@ impl Term {
                 (HeapCellValueTag::Fixnum, n) => {
                     term_stack.push(Term::Integer(n.into()));
                 }
-                (HeapCellValueTag::Cons) => {
+                (HeapCellValueTag::Cons, ptr) => {
-                    match Number::try_from(addr) {
+                    if let Ok(n) = Number::try_from(addr) {
-                        Ok(Number::Integer(i)) => term_stack.push(Term::Integer((*i).clone())),
+                        match n {
-                        Ok(Number::Rational(r)) => term_stack.push(Term::Rational((*r).clone())),
+                            Number::Integer(i) => term_stack.push(Term::Integer((*i).clone())),
-                        _ => {}
+                            Number::Rational(r) => term_stack.push(Term::Rational((*r).clone())),
+                            _ => { unreachable!() },
+                        }
+                    } else {
+                        match_untyped_arena_ptr!(ptr,
+                            (ArenaHeaderTag::Stream, stream) => {
+                                let stream_term = if let Some(alias) = stream.options().get_alias() {
+                                    Term::atom(alias.as_str().to_string())
+                                } else {
+                                    Term::compound("$stream", [
+                                        Term::integer(stream.as_ptr() as usize)
+                                    ])
+                                };
+                                term_stack.push(stream_term);
+                            }
+                            (ArenaHeaderTag::Dropped, _stream) => {
+                                term_stack.push(Term::atom("$dropped_value"));
+                            }
+                            _ => {
+                                unreachable!();
+                            }
+                        );
                     }
                 }
                 (HeapCellValueTag::CStr, s) => {
@@ -394,6 +417,7 @@ impl Term {
                 }
                 */
                 _ => {
+                    unreachable!();
                 }
             );
         }

src/machine/lib_machine/tests.rs

@@ -1,8 +1,5 @@
-use std::io::Write;
-use std::{cell::RefCell, io::Read, rc::Rc};
-
 use super::*;
-use crate::{MachineBuilder, StreamConfig};
+use crate::MachineBuilder;
 
 #[test]
 #[cfg_attr(miri, ignore = "it takes too long to run")]
@@ -611,34 +608,3 @@ 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");
-}

src/machine/streams.rs

@@ -2097,9 +2097,143 @@ impl MachineState {
 }
 
 #[cfg(test)]
-mod test {
+mod tests {
-    use super::*;
+    use crate::*;
-    use crate::machine::config::*;
+    use std::{cell::RefCell, io::Read, io::Write, rc::Rc};
+
+    fn succeeded(answer: Vec<Result<LeafAnswer, Term>>) -> bool {
+        // Ideally this should be a method in QueryState or LeafAnswer.
+        matches!(
+            answer[0].as_ref(),
+            Ok(LeafAnswer::True) | Ok(LeafAnswer::LeafAnswer { .. })
+        )
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn user_input_string_stream() {
+        let streams = StreamConfig {
+            stdin: InputStreamConfig::string("a(1,2,3)."),
+            ..Default::default()
+        };
+
+        let mut machine = MachineBuilder::default().with_streams(streams).build();
+
+        let complete_answer: Vec<_> = machine
+            .run_query(r#"current_input(_), \+ at_end_of_stream."#)
+            .collect();
+
+        assert!(succeeded(complete_answer));
+
+        let complete_answer: Vec<_> = machine.run_query("read(A).").collect();
+
+        assert_eq!(
+            complete_answer,
+            [Ok(LeafAnswer::from_bindings([(
+                "A",
+                Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3),])
+            )]))]
+        );
+
+        let complete_answer: Vec<_> = machine.run_query(r#"at_end_of_stream."#).collect();
+
+        assert!(succeeded(complete_answer));
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn user_input_channel_stream() {
+        let (mut user_input, channel_stream) = InputStreamConfig::channel();
+        let streams = StreamConfig {
+            stdin: channel_stream,
+            ..Default::default()
+        };
+
+        let mut machine = MachineBuilder::default().with_streams(streams).build();
+
+        let complete_answer: Vec<_> = machine
+            .run_query(r#"current_input(_), \+ at_end_of_stream."#)
+            .collect();
+
+        assert!(succeeded(complete_answer));
+
+        write!(user_input, "a(1,2,3).").unwrap();
+
+        let complete_answer: Vec<_> = machine
+            .run_query(r#"\+ at_end_of_stream, read(A)."#)
+            .collect();
+
+        assert_eq!(
+            complete_answer,
+            [Ok(LeafAnswer::from_bindings([(
+                "A",
+                Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3),])
+            )]))]
+        );
+
+        // End-of-data but not end-of-stream;
+        let complete_answer: Vec<_> = machine
+            .run_query(
+                r#"
+                use_module(library(charsio)),
+                current_input(In), get_n_chars(In, N, C),
+                N == 0, \+ at_end_of_stream.
+            "#,
+            )
+            .collect();
+
+        assert!(succeeded(complete_answer));
+
+        // Dropping the sender closes the input
+        drop(user_input);
+
+        let complete_answer: Vec<_> = machine
+            .run_query(
+                r#"
+                current_input(In), get_n_chars(In, N, _),
+                N == 0, at_end_of_stream.
+            "#,
+            )
+            .collect();
+
+        assert!(succeeded(complete_answer));
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn user_output_callback_stream() {
+        let test_string = Rc::new(RefCell::new(String::new()));
+
+        let streams = StreamConfig {
+            stdout: OutputStreamConfig::callback(Box::new({
+                let test_string = test_string.clone();
+                move |x| {
+                    x.read_to_string(&mut test_string.borrow_mut()).unwrap();
+                }
+            })),
+            ..Default::default()
+        };
+
+        let mut machine = MachineBuilder::default().with_streams(streams).build();
+
+        let complete_answer: Vec<_> = machine
+            .run_query(r#"current_output(Out), \+ at_end_of_stream(Out)."#)
+            .collect();
+
+        assert!(succeeded(complete_answer));
+
+        let complete_answer: Vec<_> = machine
+            .run_query(r#"write(asdf), nl, flush_output."#)
+            .collect();
+
+        assert!(succeeded(complete_answer));
+        assert_eq!(test_string.borrow().as_str(), "asdf\n");
+
+        let complete_answer: Vec<_> = machine.run_query(r#"write(abcd), flush_output."#).collect();
+
+        assert!(succeeded(complete_answer));
+        assert_eq!(test_string.borrow().as_str(), "asdf\nabcd");
+    }
 
     #[test]
     #[cfg_attr(miri, ignore)]