add close/{1,2}, better EOF action handling in read_term

src/prolog/machine/streams.rs

@@ -12,7 +12,7 @@ use std::fmt;
 use std::fs::File;
 use std::io::{stdout, Cursor, ErrorKind, Read, Seek, SeekFrom, Write};
 use std::hash::{Hash, Hasher};
-use std::net::TcpStream;
+use std::net::{Shutdown, TcpStream};
 use std::rc::Rc;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -42,6 +42,18 @@ pub enum StreamInstance {
     TcpStream(TcpStream),
 }
 
+impl Drop for StreamInstance {
+    fn drop(&mut self) {
+        match self {
+            StreamInstance::TcpStream(ref mut tcp_stream) => {
+                tcp_stream.shutdown(Shutdown::Both).unwrap();
+            }
+            _ => {
+            }
+        }
+    }
+}
+
 impl fmt::Debug for StreamInstance {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         match self {
@@ -140,12 +152,21 @@ impl Default for StreamOptions {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, Hash)]
 pub struct Stream {
     pub options: StreamOptions,
     stream_inst: WrappedStreamInstance,
 }
 
+impl PartialEq for Stream {
+    #[inline]
+    fn eq(&self, other: &Self) -> bool {
+        self.stream_inst == other.stream_inst
+    }
+}
+
+impl Eq for Stream {}
+
 impl From<TcpStream> for Stream {
     fn from(tcp_stream: TcpStream) -> Self {
         tcp_stream.set_read_timeout(None).unwrap();
@@ -292,6 +313,12 @@ impl Stream {
         }
     }
 
+    #[inline]
+    pub(crate)
+    fn close(&mut self) {
+        *self.stream_inst.0.borrow_mut() = StreamInstance::Null;
+    }
+
     #[inline]
     pub(crate)
     fn is_input_stream(&self) -> bool {

src/prolog/machine/system_calls.rs

@@ -1743,6 +1743,26 @@ impl MachineState {
                     }
                 };
             }
+            &SystemClauseType::Close => {
+                let mut stream =
+                    self.get_stream_or_alias(self[temp_v!(1)], indices, "close", 2)?;
+
+                if stream.is_output_stream() {
+                    stream.flush().unwrap(); // 8.11.6.1b)
+                }
+
+                if stream == *current_input_stream {
+                    *current_input_stream = readline::input_stream();
+                } else if stream == *current_output_stream {
+                    *current_output_stream = Stream::stdout();
+                }
+
+                stream.close();
+
+                if let Some(alias) = stream.options.alias {
+                    indices.stream_aliases.remove(&alias);
+                }
+            }
             &SystemClauseType::CopyToLiftedHeap => {
                 match self.store(self.deref(self[temp_v!(1)])) {
                     Addr::Usize(lh_offset) => {
@@ -2298,6 +2318,10 @@ impl MachineState {
 
                 stream.options = options;
 
+                if let Some(ref alias) = &stream.options.alias {
+                    indices.stream_aliases.insert(alias.clone(), stream.clone());
+                }
+
                 let stream = self.heap.to_unifiable(HeapCellValue::Stream(stream));
                 let stream_var = self.store(self.deref(self[temp_v!(3)]));
 
@@ -3464,12 +3488,12 @@ impl MachineState {
                     match TcpStream::connect(socket_addr).map_err(|e| e.kind()) {
                         Ok(tcp_stream) => {
                             let mut stream = Stream::from(tcp_stream);
+                            stream.options = options;
 
-                            if let Some(ref alias) = &options.alias {
+                            if let Some(ref alias) = &stream.options.alias {
                                 indices.stream_aliases.insert(alias.clone(), stream.clone());
                             }
 
-                            stream.options = options;
                             self.heap.to_unifiable(HeapCellValue::Stream(stream))
                         }
                         Err(ErrorKind::PermissionDenied) => {