add streams.rs, consume them in place of the old PrologStream

src/prolog/clause_types.rs

@@ -170,6 +170,7 @@ pub enum SystemClauseType {
     CheckCutPoint,
     CopyToLiftedHeap,
     CreatePartialString,
+    CurrentInput,
     DeleteAttribute,
     DeleteHeadAttribute,
     DynamicModuleResolution(usize),
@@ -283,6 +284,7 @@ impl SystemClauseType {
             &SystemClauseType::CodesToNumber => clause_name!("$codes_to_number"),
             &SystemClauseType::CopyTermWithoutAttrVars => clause_name!("$copy_term_without_attr_vars"),
             &SystemClauseType::CreatePartialString => clause_name!("$create_partial_string"),
+            &SystemClauseType::CurrentInput => clause_name!("$current_input"),
             &SystemClauseType::REPL(REPLCodePtr::CompileBatch) => clause_name!("$compile_batch"),
 	    &SystemClauseType::REPL(REPLCodePtr::UseModule) => clause_name!("$use_module"),
 	    &SystemClauseType::REPL(REPLCodePtr::UseQualifiedModule) => {
@@ -433,6 +435,7 @@ impl SystemClauseType {
             ("$check_cp", 1) => Some(SystemClauseType::CheckCutPoint),
             ("$compile_batch", 0) => Some(SystemClauseType::REPL(REPLCodePtr::CompileBatch)),
             ("$copy_to_lh", 2) => Some(SystemClauseType::CopyToLiftedHeap),
+            ("$current_input", 1) => Some(SystemClauseType::CurrentInput),
             ("$del_attr_non_head", 1) => Some(SystemClauseType::DeleteAttribute),
             ("$del_attr_head", 1) => Some(SystemClauseType::DeleteHeadAttribute),
             ("$get_next_db_ref", 2) => Some(SystemClauseType::GetNextDBRef),

src/prolog/heap_iter.rs

@@ -72,7 +72,7 @@ impl<'a> HCPreOrderIterator<'a> {
 
                 Addr::Con(Constant::String(n, s))
             }
-            Addr::Con(_) | Addr::DBRef(_) => {
+            Addr::Con(_) | Addr::DBRef(_) | Addr::Stream(_) => {
                 da
             }
             Addr::Lis(a) => {

src/prolog/lib/builtins.pl

@@ -44,7 +44,7 @@ user:term_expansion((:- op(Pred, Spec, [Op | OtherOps])), OpResults) :-
                      abolish/1, asserta/1, assertz/1, atom_chars/2,
                      atom_codes/2, atom_concat/3, atom_length/2,
                      bagof/3, catch/3, char_code/2, clause/2,
-                     current_op/3, current_predicate/1,
+		     current_input/1, current_op/3, current_predicate/1,
                      current_prolog_flag/2, expand_goal/2,
                      expand_term/2, fail/0, false/0, findall/3,
                      findall/4, get_char/1, halt/0, max_arity/1,
@@ -1012,3 +1012,5 @@ subsumes_term(General, Specific) :-
    ).
 
 unify_with_occurs_check(X, Y) :- '$unify_with_occurs_check'(X, Y).
+
+current_input(S) :- '$current_input'(S).    

src/prolog/machine/compile.rs

@@ -21,7 +21,6 @@ use ref_thread_local::RefThreadLocal;
 use std::cell::Cell;
 use std::collections::VecDeque;
 use std::fs::File;
-use std::io::Read;
 use std::mem;
 use std::path::PathBuf;
 
@@ -60,9 +59,9 @@ fn fix_filename(
     Ok(path)
 }
 
-fn load_module<R: Read>(
+fn load_module(
     wam: &mut Machine,
-    stream: ParsingStream<R>,
+    stream: Stream,
     suppress_warnings: bool,
     listing_src: &ListingSource,
 ) -> Result<ClauseName, SessionError> {
@@ -78,7 +77,11 @@ fn load_module<R: Read>(
         listing_src.clone(),
     );
 
-    let results = compiler.gather_items(wam, stream, &mut indices);
+    let results = compiler.gather_items(
+        wam,
+        &mut parsing_stream(stream),
+        &mut indices,
+    );
 
     let module_name = if let Some(ref module) = &compiler.module {
         module.module_decl.name.clone()
@@ -114,14 +117,14 @@ fn load_module_from_file(
     let mut path_buf = fix_filename(wam.indices.atom_tbl.clone(), path_buf)?;
     let filename = clause_name!(path_buf.to_string_lossy().to_string(), wam.indices.atom_tbl);
 
-    let file_handle = File::open(&path_buf).or_else(|_| {
+    let file_handle = Stream::from(File::open(&path_buf).or_else(|_| {
         Err(SessionError::InvalidFileName(filename.clone()))
-    })?;
+    })?);
 
     path_buf.pop();
 
     let listing_src = ListingSource::from_file_and_path(filename, path_buf);
-    load_module(wam, parsing_stream(file_handle), suppress_warnings, &listing_src)
+    load_module(wam, file_handle, suppress_warnings, &listing_src)
 }
 
 pub type PredicateCompileQueue = (Predicate, VecDeque<TopLevel>);
@@ -320,10 +323,11 @@ fn setup_module_expansions(wam: &mut Machine, module: &Module) {
     );
 }
 
-pub(super) fn compile_into_module<R: Read>(
+pub(super)
+fn compile_into_module(
     wam: &mut Machine,
     module_name: ClauseName,
-    src: ParsingStream<R>,
+    src: Stream,
     name: ClauseName,
 ) -> EvalSession {
     let mut indices = default_index_store!(wam.atom_tbl_of(&name));
@@ -333,7 +337,11 @@ pub(super) fn compile_into_module<R: Read>(
     indices.op_dir   = module.op_dir.clone();
     indices.atom_tbl = module.atom_tbl.clone();
 
-    let mut compiler = ListingCompiler::new(&wam.code_repo, true, module.listing_src.clone());
+    let mut compiler = ListingCompiler::new(
+        &wam.code_repo,
+        true,
+        module.listing_src.clone(),
+    );
 
     match compile_into_module_impl(wam, &mut compiler, module, src, indices) {
         Ok(()) => EvalSession::EntrySuccess,
@@ -348,11 +356,11 @@ pub(super) fn compile_into_module<R: Read>(
     }
 }
 
-fn compile_into_module_impl<R: Read>(
+fn compile_into_module_impl(
     wam: &mut Machine,
     compiler: &mut ListingCompiler,
     module: Module,
-    src: ParsingStream<R>,
+    src: Stream,
     mut indices: IndexStore,
 ) -> Result<(), SessionError> {
     setup_module_expansions(wam, &module);
@@ -363,7 +371,11 @@ fn compile_into_module_impl<R: Read>(
     wam.code_repo.compile_hook(CompileTimeHook::TermExpansion)?;
     wam.code_repo.compile_hook(CompileTimeHook::GoalExpansion)?;
 
-    let mut results = compiler.gather_items(wam, src, &mut indices)?;
+    let mut results = compiler.gather_items(
+        wam,
+        &mut parsing_stream(src),
+        &mut indices,
+    )?;
 
     compiler.adapt_in_situ_code(
         results.worker_results,
@@ -392,7 +404,6 @@ fn compile_into_module_impl<R: Read>(
     );
 
     wam.code_repo.code.extend(results.in_situ_code.into_iter());
-
     clause_code_generator.add_clause_code(wam, results.dynamic_clause_map);
 
     Ok(compiler.drop_expansions(&mut wam.code_repo))
@@ -591,7 +602,7 @@ fn load_library(
 
             load_module(
                 wam,
-                parsing_stream(code.as_bytes()),
+                Stream::from(*code),
                 suppress_warnings,
                 &listing_src,
             )
@@ -1022,10 +1033,10 @@ impl ListingCompiler {
         }
     }
 
-    fn setup_multifile_decl<R: Read>(
+    fn setup_multifile_decl(
         &self,
         indicator: MultiFileIndicator,
-        worker: &mut TopLevelBatchWorker<R>,
+        worker: &mut TopLevelBatchWorker,
     ) -> Result<(), SessionError> {
         match indicator {
             MultiFileIndicator::LocalScoped(name, arity) => {
@@ -1055,10 +1066,10 @@ impl ListingCompiler {
         Ok(())
     }
 
-    fn process_and_commit_decl<R: Read>(
+    fn process_and_commit_decl(
         &mut self,
         decl: Declaration,
-        worker: &mut TopLevelBatchWorker<R>,
+        worker: &mut TopLevelBatchWorker,
         indices: &mut IndexStore,
         flags: MachineFlags,
     ) -> Result<(), SessionError> {
@@ -1105,15 +1116,15 @@ impl ListingCompiler {
         result
     }
 
-    pub(crate) fn gather_items<R: Read>(
+    pub(crate) fn gather_items(
         &mut self,
         wam: &mut Machine,
-        mut src: ParsingStream<R>,
+        src: &mut ParsingStream<Stream>,
         indices: &mut IndexStore,
     ) -> Result<GatherResult, SessionError> {
         let flags = wam.machine_flags();
         let atom_tbl = indices.atom_tbl.clone();
-        let mut worker = TopLevelBatchWorker::new(&mut src, atom_tbl.clone(), flags, wam);
+        let mut worker = TopLevelBatchWorker::new(src, atom_tbl.clone(), flags, wam);
 
         let mut toplevel_results = vec![];
         let mut toplevel_indices = default_index_store!(atom_tbl.clone());
@@ -1306,28 +1317,32 @@ fn compile_work_impl(
     Ok(())
 }
 
-fn compile_work<R: Read>(
+fn compile_work(
     compiler: &mut ListingCompiler,
     wam: &mut Machine,
-    src: ParsingStream<R>,
+    src: Stream,
     mut indices: IndexStore,
 ) -> EvalSession {
+    let src = &mut parsing_stream(src);
     let results = try_eval_session!(compiler.gather_items(wam, src, &mut indices));
+
     try_eval_session!(compile_work_impl(compiler, wam, indices, results));
+
     EvalSession::EntrySuccess
 }
 
 /* This is a truncated version of compile_user_module, used for
 compiling code composing special forms, ie. the code that calls
 M:verify_attributes on attributed variables. */
-pub fn compile_special_form<R: Read>(
+pub fn compile_special_form(
     wam: &mut Machine,
-    src: ParsingStream<R>,
+    src: Stream,
     listing_src: ListingSource,
 ) -> Result<usize, SessionError> {
     let mut indices = default_index_store!(wam.indices.atom_tbl.clone());
     setup_indices(wam, clause_name!("builtins"), &mut indices)?;
 
+    let src = &mut parsing_stream(src);
     let mut compiler = ListingCompiler::new(&wam.code_repo, true, listing_src);
     let mut results = compiler.gather_items(wam, src, &mut indices)?;
 
@@ -1352,9 +1367,9 @@ pub fn compile_special_form<R: Read>(
 }
 
 #[inline]
-pub fn compile_listing<R: Read>(
+pub fn compile_listing(
     wam: &mut Machine,
-    src: ParsingStream<R>,
+    src: Stream,
     indices: IndexStore,
     suppress_warnings: bool,
     listing_src: ListingSource,
@@ -1388,9 +1403,9 @@ pub(super) fn setup_indices(
     }
 }
 
-pub fn compile_user_module<R: Read>(
+pub fn compile_user_module(
     wam: &mut Machine,
-    src: ParsingStream<R>,
+    src: Stream,
     suppress_warnings: bool,
     listing_src: ListingSource,
 ) -> EvalSession {

src/prolog/machine/copier.rs

@@ -330,7 +330,7 @@ impl<T: CopierTarget> CopyTermState<T> {
                         Addr::PStrLocation(addr, n) => {
                             self.copy_partial_string_from(addr, n);
                         }
-                        Addr::Con(_) | Addr::DBRef(_) => {
+                        Addr::Con(_) | Addr::DBRef(_) | Addr::Stream(_) => {
                             self.scan += 1;
                         }
                     }

src/prolog/machine/dynamic_database.rs

@@ -1,11 +1,10 @@
 use prolog_parser::ast::*;
 
 use crate::prolog::heap_print::*;
+use crate::prolog::machine::*;
 use crate::prolog::machine::compile::*;
 use crate::prolog::machine::machine_errors::*;
-use crate::prolog::machine::*;
+use crate::prolog::machine::streams::*;
-
-use std::io::Read;
 
 impl Machine {
     pub(super) fn atom_tbl_of(&self, name: &ClauseName) -> TabledData<Atom> {
@@ -15,9 +14,9 @@ impl Machine {
         }
     }
 
-    fn compile_into_machine<R: Read>(
+    fn compile_into_machine(
         &mut self,
-        src: ParsingStream<R>,
+        src: Stream,
         name: ClauseName,
         arity: usize,
     ) -> EvalSession {
@@ -130,7 +129,12 @@ impl Machine {
     ) {
         let machine_st = mem::replace(&mut self.machine_st, MachineState::new());
 
-        let result = self.compile_into_machine(parsing_stream(pred_str.as_bytes()), name, arity);
+        let result = self.compile_into_machine(
+            Stream::from(pred_str),
+            name,
+            arity,
+        );
+        
         self.machine_st = machine_st;
 
         if let EvalSession::Error(err) = result {

src/prolog/machine/machine_errors.rs

@@ -333,12 +333,14 @@ impl ValidType {
 #[derive(Clone, Copy)]
 pub enum DomainError {
     NotLessThanZero,
+    Stream,
 }
 
 impl DomainError {
     pub fn as_str(self) -> &'static str {
         match self {
             DomainError::NotLessThanZero => "not_less_than_zero",
+            DomainError::Stream => "stream",
         }
     }
 }

src/prolog/machine/machine_indices.rs

@@ -9,6 +9,7 @@ use crate::prolog::machine::Ball;
 use crate::prolog::machine::heap::*;
 use crate::prolog::machine::partial_string::*;
 use crate::prolog::machine::raw_block::RawBlockTraits;
+use crate::prolog::machine::streams::Stream;
 use crate::prolog::instructions::*;
 use crate::prolog::rug::Integer;
 
@@ -49,6 +50,7 @@ pub enum Addr {
     Str(usize),
     PStrLocation(usize, usize), // location of pstr in heap, offset into string in bytes.
     PStrTail(usize, usize), // location of pstr in heap, offset into string in bytes.
+    Stream(Stream),
 }
 
 #[derive(Clone, Copy, Hash, Eq, PartialEq)]

src/prolog/machine/machine_state.rs

@@ -9,7 +9,7 @@ use crate::prolog::machine::machine_errors::*;
 use crate::prolog::machine::machine_indices::*;
 use crate::prolog::machine::modules::*;
 use crate::prolog::machine::stack::*;
-use crate::prolog::read::PrologStream;
+use crate::prolog::machine::streams::*;
 use crate::prolog::rug::Integer;
 
 use downcast::Any;
@@ -721,7 +721,7 @@ pub(crate) trait CallPolicy: Any {
         machine_st: &mut MachineState,
         ct: &BuiltInClauseType,
         indices: &mut IndexStore,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> CallResult {
         match ct {
             &BuiltInClauseType::AcyclicTerm => {
@@ -768,7 +768,11 @@ pub(crate) trait CallPolicy: Any {
                 return_from_clause!(machine_st.last_call, machine_st)
             }
             &BuiltInClauseType::Read => {
-                match machine_st.read(parsing_stream, indices.atom_tbl.clone(), &indices.op_dir) {
+                match machine_st.read(
+                    &mut parsing_stream(current_input_stream.clone()),
+                    indices.atom_tbl.clone(),
+                    &indices.op_dir,
+                ) {
                     Ok(offset) => {
                         let addr = machine_st[temp_v!(1)].clone();
                         machine_st.unify(addr, Addr::HeapCell(offset.heap_loc));
@@ -891,13 +895,13 @@ pub(crate) trait CallPolicy: Any {
         machine_st: &mut MachineState,
         arity: usize,
         indices: &mut IndexStore,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> CallResult {
         if let Some((name, arity)) = machine_st.setup_call_n(arity) {
             match ClauseType::from(name.clone(), arity, None) {
                 ClauseType::BuiltIn(built_in) => {
                     machine_st.setup_built_in_call(built_in.clone());
-                    self.call_builtin(machine_st, &built_in, indices, parsing_stream)?;
+                    self.call_builtin(machine_st, &built_in, indices, current_input_stream)?;
                 }
                 ClauseType::CallN => {
                     machine_st.handle_internal_call_n(arity);
@@ -977,10 +981,10 @@ impl CallPolicy for CWILCallPolicy {
         machine_st: &mut MachineState,
         ct: &BuiltInClauseType,
         indices: &mut IndexStore,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> CallResult {
         self.prev_policy
-            .call_builtin(machine_st, ct, indices, parsing_stream)?;
+            .call_builtin(machine_st, ct, indices, current_input_stream)?;
         self.increment(machine_st)
     }
 
@@ -989,10 +993,10 @@ impl CallPolicy for CWILCallPolicy {
         machine_st: &mut MachineState,
         arity: usize,
         indices: &mut IndexStore,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> CallResult {
         self.prev_policy
-            .call_n(machine_st, arity, indices, parsing_stream)?;
+            .call_n(machine_st, arity, indices, current_input_stream)?;
         self.increment(machine_st)
     }
 }

src/prolog/machine/machine_state_impl.rs

@@ -15,8 +15,8 @@ use crate::prolog::machine::machine_errors::*;
 use crate::prolog::machine::machine_indices::*;
 use crate::prolog::machine::machine_state::*;
 use crate::prolog::machine::stack::*;
+use crate::prolog::machine::streams::*;
 use crate::prolog::ordered_float::*;
-use crate::prolog::read::PrologStream;
 use crate::prolog::rug::{Integer, Rational};
 
 use indexmap::{IndexMap, IndexSet};
@@ -1768,7 +1768,8 @@ impl MachineState {
 
                 let offset = match addr {
                     Addr::HeapCell(_) | Addr::StackCell(..)
-                  | Addr::AttrVar(..) | Addr::PStrTail(..) => {
+                  | Addr::AttrVar(..) | Addr::PStrTail(..) 
+                  | Addr::Stream(_) => {
                         v
                     }
                     Addr::Con(Constant::String(n, ref s)) => {
@@ -1790,7 +1791,7 @@ impl MachineState {
                     }
                     Addr::Str(_) => {
                         s
-                    }
+                    }                    
                     Addr::DBRef(_) => {
                         self.fail = true;
                         return;
@@ -2788,7 +2789,7 @@ impl MachineState {
         let a1 = self.store(self.deref(self[temp_v!(1)].clone()));
 
         match a1.clone() {
-            Addr::DBRef(_) => self.fail = true,
+            Addr::Stream(_) | Addr::DBRef(_) => self.fail = true,
             Addr::Con(Constant::String(n, ref s))
                 if !self.flags.double_quotes.is_atom() && !s[n ..].is_empty() =>
             {
@@ -3217,7 +3218,7 @@ impl MachineState {
         code_repo: &CodeRepo,
         call_policy: &mut Box<dyn CallPolicy>,
         cut_policy: &mut Box<dyn CutPolicy>,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
         ct: &ClauseType,
         arity: usize,
         lco: bool,
@@ -3244,11 +3245,11 @@ impl MachineState {
         match ct {
             &ClauseType::BuiltIn(ref ct) => try_or_fail!(
                 self,
-                call_policy.call_builtin(self, ct, indices, parsing_stream)
+                call_policy.call_builtin(self, ct, indices, current_input_stream)
             ),
             &ClauseType::CallN => try_or_fail!(
                 self,
-                call_policy.call_n(self, arity, indices, parsing_stream)
+                call_policy.call_n(self, arity, indices, current_input_stream)
             ),
             &ClauseType::Hook(ref hook) => try_or_fail!(self, call_policy.compile_hook(self, hook)),
             &ClauseType::Inlined(ref ct) => {
@@ -3272,7 +3273,7 @@ impl MachineState {
                     indices,
                     call_policy,
                     cut_policy,
-                    parsing_stream
+                    current_input_stream,
                 )
             ),
         };
@@ -3286,18 +3287,20 @@ impl MachineState {
         code_repo: &CodeRepo,
         call_policy: &mut Box<dyn CallPolicy>,
         cut_policy: &mut Box<dyn CutPolicy>,
-        parsing_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
         instr: &ControlInstruction,
     ) {
         match instr {
-            &ControlInstruction::Allocate(num_cells) => self.allocate(num_cells),
+            &ControlInstruction::Allocate(num_cells) => {
+                self.allocate(num_cells);
+            }
             &ControlInstruction::CallClause(ref ct, arity, _, lco, use_default_cp) => self
                 .handle_call_clause(
                     indices,
                     code_repo,
                     call_policy,
                     cut_policy,
-                    parsing_stream,
+                    current_input_stream,
                     ct,
                     arity,
                     lco,
@@ -3313,7 +3316,9 @@ impl MachineState {
                 self.b0 = self.b;
                 self.p += offset;
             }
-            &ControlInstruction::Proceed => self.p = CodePtr::Local(self.cp.clone())
+            &ControlInstruction::Proceed => {
+                self.p = CodePtr::Local(self.cp.clone());
+            }
         };
     }
 

src/prolog/machine/mod.rs

@@ -22,6 +22,7 @@ pub mod modules;
 mod partial_string;
 mod raw_block;
 mod stack;
+pub(super) mod streams;
 pub(super) mod term_expansion;
 pub mod toplevel;
 
@@ -36,14 +37,13 @@ use crate::prolog::machine::machine_errors::*;
 use crate::prolog::machine::machine_indices::*;
 use crate::prolog::machine::machine_state::*;
 use crate::prolog::machine::modules::*;
+use crate::prolog::machine::streams::*;
 use crate::prolog::machine::toplevel::*;
-use crate::prolog::read::PrologStream;
 
 use indexmap::IndexMap;
 
 use std::collections::VecDeque;
 use std::fs::File;
-use std::io::Read;
 use std::mem;
 use std::ops::Index;
 use std::path::PathBuf;
@@ -76,7 +76,7 @@ pub struct Machine {
     pub(super) indices: IndexStore,
     pub(super) code_repo: CodeRepo,
     pub(super) toplevel_idx: usize,
-    pub(super) prolog_stream: ParsingStream<Box<dyn Read>>,
+    pub(super) current_input_stream: Stream,
 }
 
 impl Index<LocalCodePtr> for CodeRepo {
@@ -216,7 +216,11 @@ impl Machine {
             current_dir.clone(),
         );
 
-        match compile_special_form(self, parsing_stream(VERIFY_ATTRS.as_bytes()), verify_attrs_src)
+        match compile_special_form(
+            self,
+            Stream::from(VERIFY_ATTRS),
+            verify_attrs_src,
+        )
         {
             Ok(p) => {
                 self.machine_st.attr_var_init.verify_attrs_loc = p;
@@ -230,7 +234,11 @@ impl Machine {
             current_dir,
         );
 
-        match compile_special_form(self, parsing_stream(PROJECT_ATTRS.as_bytes()), project_attrs_src)
+        match compile_special_form(
+            self,
+            Stream::from(PROJECT_ATTRS),
+            project_attrs_src,
+        )
         {
             Ok(p) => {
                 self.machine_st.attr_var_init.project_attrs_loc = p;
@@ -254,7 +262,7 @@ impl Machine {
 
         compile_user_module(
             self,
-            parsing_stream(TOPLEVEL.as_bytes()),
+            Stream::from(TOPLEVEL),
             true,
             top_lvl_src,
         );
@@ -282,7 +290,7 @@ impl Machine {
 
         if path.is_file() {
             let file_src = match File::open(&path) {
-                Ok(file_handle) => parsing_stream(file_handle),
+                Ok(file_handle) => Stream::from(file_handle),
                 Err(_) => return,
             };
 
@@ -297,7 +305,7 @@ impl Machine {
 
     #[cfg(test)]
     pub fn reset(&mut self) {
-        self.prolog_stream = readline::input_stream();
+        self.current_input_stream = readline::input_stream();
         self.policies.cut_policy = Box::new(DefaultCutPolicy {});
         self.machine_st.reset();
     }
@@ -336,7 +344,7 @@ impl Machine {
         self.run_query();
     }
 
-    pub fn new(prolog_stream: PrologStream) -> Self
+    pub fn new(current_input_stream: Stream) -> Self
     {
         let mut wam = Machine {
             machine_st: MachineState::new(),
@@ -345,7 +353,7 @@ impl Machine {
             indices: IndexStore::new(),
             code_repo: CodeRepo::new(),
             toplevel_idx: 0,
-            prolog_stream,
+            current_input_stream,
         };
 
         let atom_tbl = wam.indices.atom_tbl.clone();
@@ -358,7 +366,7 @@ impl Machine {
 
         compile_listing(
             &mut wam,
-            parsing_stream(BUILTINS.as_bytes()),
+            Stream::from(BUILTINS),
             default_index_store!(atom_tbl.clone()),
             true,
             ListingSource::from_file_and_path(
@@ -369,25 +377,36 @@ impl Machine {
 
         wam.compile_special_forms();
 
-        compile_user_module(&mut wam, parsing_stream(ERROR.as_bytes()), true,
+        compile_user_module(&mut wam,
+                            Stream::from(ERROR),
+                            true,
                             ListingSource::from_file_and_path(
                                 clause_name!("error"),
                                 lib_path.clone(),
                             )
         );
-        compile_user_module(&mut wam, parsing_stream(LISTS.as_bytes()), true,
+        
+        compile_user_module(&mut wam,
+                            Stream::from(LISTS),
+                            true,
                             ListingSource::from_file_and_path(
                                 clause_name!("lists"),
                                 lib_path.clone(),
                             ),
         );
-        compile_user_module(&mut wam, parsing_stream(ISO_EXT.as_bytes()), true,
+        
+        compile_user_module(&mut wam,
+                            Stream::from(ISO_EXT),
+                            true,
                             ListingSource::from_file_and_path(
                                 clause_name!("iso_ext"),
                                 lib_path.clone(),
                             )
         );
-        compile_user_module(&mut wam, parsing_stream(SI.as_bytes()), true,
+        
+        compile_user_module(&mut wam,
+                            Stream::from(SI),
+                            true,
                             ListingSource::from_file_and_path(
                                 clause_name!("si"),
                                 lib_path.clone(),
@@ -755,12 +774,15 @@ impl Machine {
                 &mut self.indices,
                 &mut self.policies,
                 &mut self.code_repo,
-                &mut self.prolog_stream,
+                &mut self.current_input_stream,
             );
 
             match self.machine_st.p {
-                CodePtr::Local(LocalCodePtr::TopLevel(_, p)) if p > 0 => {}
+                CodePtr::Local(LocalCodePtr::TopLevel(_, p)) if p > 0 => {
-                CodePtr::REPL(code_ptr, p) => self.handle_toplevel_command(code_ptr, p),
+                }
+                CodePtr::REPL(code_ptr, p) => {
+                    self.handle_toplevel_command(code_ptr, p)
+                }
                 CodePtr::DynamicTransaction(trans_type, p) => {
                     // self.code_repo.cached_query is about to be overwritten by the term expander,
                     // so hold onto it locally and restore it after the compiler has finished.
@@ -788,7 +810,7 @@ impl MachineState {
         indices: &mut IndexStore,
         policies: &mut MachinePolicies,
         code_repo: &CodeRepo,
-        prolog_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) {
         match instr {
             &Line::Arithmetic(ref arith_instr) => self.execute_arith_instr(arith_instr),
@@ -803,7 +825,7 @@ impl MachineState {
                 code_repo,
                 &mut policies.call_policy,
                 &mut policies.cut_policy,
-                prolog_stream,
+                current_input_stream,
                 control_instr,
             ),
             &Line::Fact(ref fact_instr) => {
@@ -826,14 +848,20 @@ impl MachineState {
         indices: &mut IndexStore,
         policies: &mut MachinePolicies,
         code_repo: &CodeRepo,
-        prolog_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) {
         let instr = match code_repo.lookup_instr(self.last_call, &self.p) {
             Some(instr) => instr,
             None => return,
         };
 
-        self.dispatch_instr(instr.as_ref(), indices, policies, code_repo, prolog_stream);
+        self.dispatch_instr(
+            instr.as_ref(),
+            indices,
+            policies,
+            code_repo,
+            current_input_stream,
+        );
     }
 
     fn backtrack(&mut self) {
@@ -884,7 +912,7 @@ impl MachineState {
         indices: &mut IndexStore,
         policies: &mut MachinePolicies,
         code_repo: &mut CodeRepo,
-        prolog_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> bool {
         loop {
             let instr = match code_repo.lookup_instr(self.last_call, &self.p) {
@@ -900,7 +928,13 @@ impl MachineState {
                 None => return false,
             };
 
-            self.dispatch_instr(instr.as_ref(), indices, policies, code_repo, prolog_stream);
+            self.dispatch_instr(
+                instr.as_ref(),
+                indices,
+                policies,
+                code_repo,
+                current_input_stream
+            );
 
             if self.fail {
                 self.backtrack();
@@ -924,10 +958,10 @@ impl MachineState {
         indices: &mut IndexStore,
         policies: &mut MachinePolicies,
         code_repo: &mut CodeRepo,
-        prolog_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) {
         loop {
-            self.execute_instr(indices, policies, code_repo, prolog_stream);
+            self.execute_instr(indices, policies, code_repo, current_input_stream);
 
             if self.fail {
                 self.backtrack();
@@ -943,7 +977,7 @@ impl MachineState {
                     if !instigating_instr.as_ref().is_head_instr() {
                         let cp = self.p.local();
                         self.run_verify_attr_interrupt(cp);
-                    } else if !self.verify_attr_stepper(indices, policies, code_repo, prolog_stream) {
+                    } else if !self.verify_attr_stepper(indices, policies, code_repo, current_input_stream) {
                         if self.fail {
                             break;
                         }

src/prolog/machine/streams.rs

@@ -0,0 +1,245 @@
+use crate::prolog_parser::ast::*;
+
+use crate::prolog::read::readline::*;
+
+use std::cell::RefCell;
+use std::error::Error;
+use std::fmt;
+use std::fs::File;
+use std::io::{Cursor, ErrorKind, Read, Write};
+use std::hash::{Hash, Hasher};
+use std::net::TcpStream;
+use std::rc::Rc;
+
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
+pub enum StreamType {
+    Binary,
+    Text,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Hash)]
+pub enum EOFAction {
+    EOFCode,
+    Error,
+    Reset,
+}
+
+/* all these streams are closed automatically when the instance is
+ * dropped. */
+pub enum StreamInstance {
+    Bytes(Cursor<Vec<u8>>),
+    DynReadSource(Box<dyn Read>),
+    File(File),    
+    ReadlineStream(ReadlineStream),
+    TcpStream(TcpStream),
+}
+
+#[derive(Clone)]
+struct WrappedStreamInstance(Rc<RefCell<StreamInstance>>);
+
+impl WrappedStreamInstance {
+    #[inline]
+    fn new(stream_inst: StreamInstance) -> Self {
+        WrappedStreamInstance(Rc::new(RefCell::new(stream_inst)))
+    }
+}
+
+impl PartialEq for WrappedStreamInstance {
+    #[inline]
+    fn eq(&self, other: &Self) -> bool {
+        Rc::ptr_eq(&self.0, &other.0)
+    }
+}
+
+impl Eq for WrappedStreamInstance {}
+
+impl Hash for WrappedStreamInstance {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        let rc = &self.0;
+        let ptr = Rc::into_raw(rc.clone());
+
+        state.write_usize(ptr as usize);
+
+        unsafe {
+            // necessary to avoid memory leak.
+            let _ = Rc::from_raw(ptr);
+        };
+    }
+}
+
+#[derive(Debug)]
+enum StreamError {
+    ReadFromOutputStream,
+    WriteToInputStream,
+    FlushToInputStream,
+}
+
+impl fmt::Display for StreamError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            StreamError::ReadFromOutputStream => {
+                write!(f, "attempted to read from a write-only stream")
+            }
+            StreamError::WriteToInputStream => {
+                write!(f, "attempted to write to a read-only stream")
+            }
+            StreamError::FlushToInputStream => {
+                write!(f, "attempted to flush a read-only stream")
+            }
+        }
+    }
+}
+
+impl Error for StreamError {}
+
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct StreamOptions {
+    pub stream_type: StreamType,
+    pub reposition: bool,
+    pub alias: Option<ClauseName>,
+    pub eof_action: EOFAction,
+}
+
+impl Default for StreamOptions {
+    #[inline]
+    fn default() -> Self {
+        StreamOptions {
+            stream_type: StreamType::Text,
+            reposition: false,
+            alias: None,
+            eof_action: EOFAction::EOFCode,
+        }
+    }
+}
+
+#[derive(Clone, PartialEq, Eq, Hash)]
+pub struct Stream {
+    pub options: StreamOptions,
+    stream_inst: WrappedStreamInstance,
+}
+
+impl From<String> for Stream {
+    fn from(string: String) -> Self {
+        Stream {
+            options: StreamOptions::default(),
+            stream_inst: WrappedStreamInstance::new(
+                StreamInstance::Bytes(Cursor::new(string.into_bytes()))
+            )
+        }
+    }
+}
+
+impl From<ReadlineStream> for Stream {
+    fn from(rl_stream: ReadlineStream) -> Self {
+        Stream {
+            options: StreamOptions::default(),
+            stream_inst: WrappedStreamInstance::new(
+                StreamInstance::ReadlineStream(rl_stream)
+            ),
+        }
+    }
+}
+
+impl From<&'static str> for Stream {
+    fn from(src: &'static str) -> Stream {
+        Stream {
+            options: StreamOptions::default(),
+            stream_inst: WrappedStreamInstance::new(
+                StreamInstance::DynReadSource(Box::new(src.as_bytes()))
+            ),
+        }
+    }
+}
+
+impl From<File> for Stream {
+    fn from(file: File) -> Stream {
+        Stream {
+            options: StreamOptions::default(),
+            stream_inst: WrappedStreamInstance::new(
+                StreamInstance::File(file)
+            ),
+        }
+    }
+}
+
+impl Stream {
+    #[inline]
+    pub(crate)
+    fn as_ptr(&self) -> *const RefCell<StreamInstance> {
+        let rc = self.stream_inst.0.clone();
+        let ptr = Rc::into_raw(rc);
+
+        unsafe {
+            // must be done to avoid memory leak.
+            let _ = Rc::from_raw(ptr);
+        }
+
+        ptr
+    }
+}
+
+impl Read for Stream {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        match *self.stream_inst.0.borrow_mut() {
+            StreamInstance::File(ref mut file) => {
+                file.read(buf)
+            }
+            StreamInstance::TcpStream(ref mut tcp_stream) => {
+                tcp_stream.read(buf)
+            }
+            StreamInstance::ReadlineStream(ref mut rl_stream) => {
+                rl_stream.read(buf)
+            }
+            StreamInstance::DynReadSource(ref mut src) => {
+                src.read(buf)
+            }
+            StreamInstance::Bytes(ref mut cursor) => {
+                cursor.read(buf)
+            }
+        }
+    }
+}
+
+impl Write for Stream {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        match *self.stream_inst.0.borrow_mut() {
+            StreamInstance::File(ref mut file) => {
+                file.write(buf)
+            }
+            StreamInstance::TcpStream(ref mut tcp_stream) => {
+                tcp_stream.write(buf)
+            }
+            StreamInstance::Bytes(ref mut cursor) => {
+                cursor.write(buf)
+            }
+            _ => {
+                Err(std::io::Error::new(
+                    ErrorKind::PermissionDenied,
+                    StreamError::WriteToInputStream,
+                ))
+            }
+        }
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        match *self.stream_inst.0.borrow_mut() {
+            StreamInstance::File(ref mut file) => {
+                file.flush()
+            }
+            StreamInstance::TcpStream(ref mut tcp_stream) => {
+                tcp_stream.flush()
+            }
+            StreamInstance::Bytes(ref mut cursor) => {
+                cursor.flush()
+            }
+            _ => {
+                Err(std::io::Error::new(
+                    ErrorKind::PermissionDenied,
+                    StreamError::FlushToInputStream,
+                ))
+            }
+        }
+    }
+}
+
+//TODO: write a Seek instance.

src/prolog/machine/system_calls.rs

@@ -12,9 +12,10 @@ use crate::prolog::machine::code_walker::*;
 use crate::prolog::machine::machine_errors::*;
 use crate::prolog::machine::machine_indices::*;
 use crate::prolog::machine::machine_state::*;
+use crate::prolog::machine::streams::*;
 use crate::prolog::machine::toplevel::to_op_decl;
 use crate::prolog::ordered_float::OrderedFloat;
-use crate::prolog::read::{readline, PrologStream};
+use crate::prolog::read::readline;
 use crate::prolog::rug::Integer;
 
 use crate::ref_thread_local::RefThreadLocal;
@@ -354,12 +355,12 @@ impl MachineState {
     }
 
     fn read_term(&mut self,
-                 current_input_stream: &mut PrologStream,
+                 current_input_stream: &mut Stream,
                  indices: &mut IndexStore)
                  -> CallResult
     {
         match self.read(
-            current_input_stream,
+            &mut parsing_stream(current_input_stream.clone()),
             indices.atom_tbl.clone(),
             &indices.op_dir,
         ) {
@@ -671,7 +672,7 @@ impl MachineState {
         indices: &mut IndexStore,
         call_policy: &mut Box<dyn CallPolicy>,
         cut_policy: &mut Box<dyn CutPolicy>,
-        current_input_stream: &mut PrologStream,
+        current_input_stream: &mut Stream,
     ) -> CallResult {
         match ct {
             &SystemClauseType::AbolishClause => {
@@ -721,6 +722,32 @@ impl MachineState {
                 self.p = CodePtr::DynamicTransaction(trans_type, p);
                 return Ok(());
             }
+            &SystemClauseType::CurrentInput => {
+                let addr = self.store(self.deref(self[temp_v!(1)].clone()));
+                let stream = current_input_stream.clone();
+
+                match addr {
+                    addr if addr.is_ref() => {                        
+                        self.unify(Addr::Stream(stream), addr);
+                    }
+                    Addr::Stream(other_stream) => {
+                        self.fail = stream != other_stream;
+                    }
+                    addr => {
+                        let stub = MachineError::functor_stub(
+                            clause_name!("current_input"),
+                            1,
+                        );
+
+                        let err = MachineError::domain_error(
+                            DomainError::Stream,
+                            addr,
+                        );
+
+                        return Err(self.error_form(err, stub));
+                    }
+                }                
+            }
             &SystemClauseType::AtEndOfExpansion => {
                 if self.cp == LocalCodePtr::TopLevel(0, 0) {
                     self.at_end_of_expansion = true;
@@ -1177,7 +1204,9 @@ impl MachineState {
                 };
             }
             &SystemClauseType::GetChar => {
-                let result = current_input_stream.next();
+                let mut iter = parsing_stream(current_input_stream.clone());
+                let result = iter.next();
+                
                 let a1 = self[temp_v!(1)].clone();
 
                 match result {
@@ -2315,7 +2344,7 @@ impl MachineState {
                         // get the call site so that the number of active permanent variables can be read
                         // from it later.
                         let cp = (self.stack.index_and_frame(e).prelude.cp - 1).unwrap();
-
+                        
                         let p = cp.as_functor(&mut self.heap);
                         let e = self.stack.index_and_frame(e).prelude.e;
 

src/prolog/machine/term_expansion.rs

@@ -8,7 +8,6 @@ use crate::prolog::rug::Integer;
 
 use std::cell::Cell;
 use std::collections::VecDeque;
-use std::io::Read;
 use std::iter::Rev;
 use std::vec::IntoIter;
 
@@ -28,7 +27,11 @@ where
     term
 }
 
-fn extract_from_list(head: Box<Term>, tail: Box<Term>) -> Result<Rev<IntoIter<Term>>, ParserError> {
+fn extract_from_list(
+    head: Box<Term>,
+    tail: Box<Term>,
+) -> Result<Rev<IntoIter<Term>>, ParserError>
+{
     let mut terms = vec![*head];
     let mut tail = *tail;
 
@@ -44,10 +47,10 @@ fn extract_from_list(head: Box<Term>, tail: Box<Term>) -> Result<Rev<IntoIter<Te
     }
 }
 
-pub struct TermStream<'a, R: Read> {
+pub struct TermStream<'a> {
     stack: Vec<Term>,
     pub(crate) wam: &'a mut Machine,
-    parser: Parser<'a, R>,
+    parser: Parser<'a, Stream>,
     pub(crate) flags: MachineFlags,
     term_expansion_lens: (usize, usize),
     goal_expansion_lens: (usize, usize),
@@ -75,7 +78,7 @@ impl ExpansionAdditionResult {
     }
 }
 
-impl<'a, R: Read> Drop for TermStream<'a, R> {
+impl<'a> Drop for TermStream<'a> {
     fn drop(&mut self) {
         self.wam.indices.in_situ_code_dir.clear();
         self.wam.indices.in_situ_module_dir.clear();
@@ -85,9 +88,9 @@ impl<'a, R: Read> Drop for TermStream<'a, R> {
     }
 }
 
-impl<'a, R: Read> TermStream<'a, R> {
+impl<'a> TermStream<'a> {
     pub fn new(
-        src: &'a mut ParsingStream<R>,
+        src: &'a mut ParsingStream<Stream>,
         atom_tbl: TabledData<Atom>,
         flags: MachineFlags,
         wam: &'a mut Machine,

src/prolog/machine/toplevel.rs

@@ -13,7 +13,6 @@ use indexmap::{IndexMap, IndexSet};
 use std::borrow::BorrowMut;
 use std::cell::Cell;
 use std::collections::VecDeque;
-use std::io::Read;
 use std::mem;
 use std::ops::DerefMut;
 use std::rc::Rc;
@@ -30,15 +29,15 @@ fn op_dir<'a, 'b: 'a>(from: &'b IndexSource<'a, IndexStore>) -> RefOrOwned<'a, O
     }
 }
 
-struct CompositeIndices<'a, 'b, 'c, R: Read> {
+struct CompositeIndices<'a, 'b, 'c> {
-    term_stream: &'b mut TermStream<'a, R>,
+    term_stream: &'b mut TermStream<'a>,
     index_src: IndexSource<'c, IndexStore>,
     static_code_dir: Option<IndexSource<'c, CodeDir>>
 }
 
-impl<'a, 'b, 'c, R: Read> CompositeIndices<'a, 'b, 'c, R> {
+impl<'a, 'b, 'c> CompositeIndices<'a, 'b, 'c> {
     fn new(
-        term_stream: &'b mut TermStream<'a, R>,
+        term_stream: &'b mut TermStream<'a>,
         index_src: IndexSource<'c, IndexStore>,
         static_code_dir: Option<IndexSource<'c, CodeDir>>,
     ) -> Self {
@@ -577,8 +576,8 @@ fn draw_from_term_dir_impl(
     }
 }
 
-fn draw_from_term_dir<R: Read>(
+fn draw_from_term_dir(
-    indices: &CompositeIndices<R>,
+    indices: &CompositeIndices,
     intra_module_term_dirs: &mut IndexMap<ClauseName, TermDirQuantum>,
     top_level_term_dirs: &mut TermDirQuantum,
     key: &PredicateKey,
@@ -615,8 +614,8 @@ fn draw_from_term_dir<R: Read>(
     );
 }
 
-fn setup_declaration<'a, 'b, 'c, R: Read>(
+fn setup_declaration<'a, 'b, 'c>(
-    indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+    indices: &mut CompositeIndices<'a, 'b, 'c>,
     flags: MachineFlags,
     mut terms: Vec<Box<Term>>,
     line_num: usize,
@@ -796,9 +795,9 @@ impl RelationWorker {
         self.fabricate_rule(fold_by_str(prec_seq.into_iter(), body_term, comma_sym))
     }
 
-    fn to_query_term<'a, 'b, 'c, R: Read>(
+    fn to_query_term<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         term: Term,
     ) -> Result<QueryTerm, ParserError> {
         match term {
@@ -872,9 +871,9 @@ impl RelationWorker {
         }
     }
 
-    fn pre_query_term<'a, 'b, 'c, R: Read>(
+    fn pre_query_term<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         term: Term,
     ) -> Result<QueryTerm, ParserError> {
         match term {
@@ -893,9 +892,9 @@ impl RelationWorker {
         }
     }
 
-    fn setup_query<'a, 'b, 'c, R: Read>(
+    fn setup_query<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         terms: Vec<Box<Term>>,
         blocks_cuts: bool,
     ) -> Result<Vec<QueryTerm>, ParserError> {
@@ -946,10 +945,10 @@ impl RelationWorker {
         Ok(query_terms)
     }
 
-    fn setup_hook<'a, 'b, 'c, R: Read>(
+    fn setup_hook<'a, 'b, 'c>(
         &mut self,
         hook: CompileTimeHook,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         term: Term,
     ) -> Result<CompileTimeHookCompileInfo, ParserError> {
         match flatten_hook(term) {
@@ -970,9 +969,9 @@ impl RelationWorker {
         }
     }
 
-    fn setup_rule<'a, 'b, 'c, R: Read>(
+    fn setup_rule<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         mut terms: Vec<Box<Term>>,
         blocks_cuts: bool,
         assume_dyn: bool,
@@ -1003,9 +1002,9 @@ impl RelationWorker {
         }
     }
 
-    fn try_term_to_query<'a, 'b, 'c, R: Read>(
+    fn try_term_to_query<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         terms: Vec<Box<Term>>,
         blocks_cuts: bool,
     ) -> Result<TopLevel, ParserError> {
@@ -1016,10 +1015,10 @@ impl RelationWorker {
         )?))
     }
 
-    fn compact_module_scoped_head<'a, 'b, 'c, R: Read>(
+    fn compact_module_scoped_head<'a, 'b, 'c>(
         &self,
         term: &mut Term,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
     ) {
         let inner_term = match term {
             Term::Clause(_, ref name, ref mut inner_terms, _)
@@ -1044,9 +1043,9 @@ impl RelationWorker {
         *term = inner_term;
     }
 
-    fn try_term_to_tl<'a, 'b, 'c, R: Read>(
+    fn try_term_to_tl<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         term: Term,
         blocks_cuts: bool,
     ) -> Result<TopLevel, ParserError> {
@@ -1085,14 +1084,14 @@ impl RelationWorker {
         }
     }
 
-    fn try_terms_to_tls<'a, 'b, 'c, I, R>(
+    fn try_terms_to_tls<'a, 'b, 'c, I>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
         terms: I,
         blocks_cuts: bool,
     ) -> Result<VecDeque<TopLevel>, ParserError>
     where
-        I: IntoIterator<Item = Term>, R: Read
+        I: IntoIterator<Item = Term>
     {
         let mut results = VecDeque::new();
 
@@ -1103,9 +1102,9 @@ impl RelationWorker {
         Ok(results)
     }
 
-    fn parse_queue<'a, 'b, 'c, R: Read>(
+    fn parse_queue<'a, 'b, 'c>(
         &mut self,
-        indices: &mut CompositeIndices<'a, 'b, 'c, R>,
+        indices: &mut CompositeIndices<'a, 'b, 'c>,
     ) -> Result<VecDeque<TopLevel>, ParserError> {
         let mut queue = VecDeque::new();
 
@@ -1128,8 +1127,8 @@ pub type DynamicClause = Vec<(Term, Term)>;
 
 pub type DynamicClauseMap = IndexMap<(ClauseName, usize), DynamicClause>;
 
-pub struct TopLevelBatchWorker<'a, R: Read> {
+pub struct TopLevelBatchWorker<'a> {
-    pub(crate) term_stream: TermStream<'a, R>,
+    pub(crate) term_stream: TermStream<'a>,
     rel_worker: RelationWorker,
     pub(crate) results: Vec<(Predicate, VecDeque<TopLevel>)>,
     pub(crate) dynamic_clause_map: DynamicClauseMap,
@@ -1139,14 +1138,14 @@ pub struct TopLevelBatchWorker<'a, R: Read> {
     pub(crate) non_counted_bt_preds: IndexSet<PredicateKey>,
 }
 
-impl<'a, R: Read> TopLevelBatchWorker<'a, R> {
+impl<'a> TopLevelBatchWorker<'a> {
     pub fn new(
-        inner: &'a mut ParsingStream<R>,
+        stream: &'a mut ParsingStream<Stream>,
         atom_tbl: TabledData<Atom>,
         flags: MachineFlags,
         wam: &'a mut Machine,
     ) -> Self {
-        let term_stream = TermStream::new(inner, atom_tbl, flags, wam);
+        let term_stream = TermStream::new(stream, atom_tbl, flags, wam);
 
         let line_num = term_stream.line_num();
         let col_num  = term_stream.col_num();

src/prolog/read.rs

@@ -6,9 +6,9 @@ use crate::prolog::forms::*;
 use crate::prolog::iterators::*;
 use crate::prolog::machine::machine_indices::*;
 use crate::prolog::machine::machine_state::MachineState;
+use crate::prolog::machine::streams::Stream;
 
 use std::collections::VecDeque;
-use std::io::Read;
 
 type SubtermDeque = VecDeque<(usize, usize)>;
 
@@ -23,10 +23,10 @@ impl<'a> TermRef<'a> {
     }
 }
 
-pub type PrologStream = ParsingStream<Box<dyn Read>>;
+pub type PrologStream = ParsingStream<Stream>;
 
 pub mod readline {
-    use prolog_parser::ast::*;
+    use crate::prolog::machine::streams::Stream;
     use crate::prolog::rustyline::error::ReadlineError;
     use crate::prolog::rustyline::{Cmd, Editor, KeyPress};
     use std::io::{Cursor, Read};
@@ -52,10 +52,10 @@ pub mod readline {
     }
 
     impl ReadlineStream {
-        fn input_stream(pending_input: String) -> Self {
+        pub fn input_stream(pending_input: String) -> Stream {
             let mut rl = Editor::<()>::new();
             rl.bind_sequence(KeyPress::Tab, Cmd::Insert(1, "\t".to_string()));
-            ReadlineStream { rl, pending_input: Cursor::new(pending_input) }
+            Stream::from(ReadlineStream { rl, pending_input: Cursor::new(pending_input) })
         }
 
         fn call_readline(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
@@ -99,9 +99,9 @@ pub mod readline {
     }
 
     #[inline]
-    pub fn input_stream() -> crate::PrologStream {
+    pub fn input_stream() -> Stream {
-        let reader: Box<dyn Read> = Box::new(ReadlineStream::input_stream(String::from("")));
+        let input_stream = ReadlineStream::input_stream(String::from(""));
-        parsing_stream(reader)
+        Stream::from(input_stream)
     }
 }
 

src/prolog/write.rs

@@ -184,6 +184,7 @@ impl fmt::Display for Addr {
             &Addr::Str(s) => write!(f, "Addr::Str({})", s),
             &Addr::PStrLocation(h, n) => write!(f, "Addr::PStrLocation({}, {})", h, n),
             &Addr::PStrTail(h, n) => write!(f, "Addr::PStrTail({}, {})", h, n),
+            &Addr::Stream(ref stream) => write!(f, "Addr::Stream({})", stream.as_ptr() as usize),
         }
     }
 }