}
}
+impl From<Stream> for HeapCellValue {
+ #[inline(always)]
+ fn from(stream: Stream) -> Self {
+ if stream.is_null_stream() {
+ let res = atom!("null_stream");
+ atom_as_cell!(res)
+ } else {
+ let res = stream.as_ptr();
+ debug_assert!(!res.is_null());
+ raw_ptr_as_cell!(res)
+ }
+ }
+}
+
impl Stream {
#[inline]
pub(crate) fn position(&mut self) -> Option<(u64, usize)> {
match_untyped_arena_ptr!(ptr,
(ArenaHeaderTag::Stream, stream) => {
return if stream.is_null_stream() {
- Err(self.open_permission_error(stream_as_cell!(stream), caller, arity))
+ Err(self.open_permission_error(HeapCellValue::from(stream), caller, arity))
} else {
Ok(stream)
};
if let Some(alias) = stream.options().get_alias() {
atom_as_cell!(alias)
} else {
- stream_as_cell!(stream)
+ stream.into()
},
);
}
}
}
+
+#[cfg(test)]
+mod test {
+ use super::*;
+ use crate::machine::config::*;
+
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn current_input_null_stream() {
+ let mut machine = MachineBuilder::new()
+ .with_streams(StreamConfig::in_memory())
+ .build();
+
+ let results = machine.run_query("current_input(S).").collect::<Vec<_>>();
+
+ assert_eq!(results.len(), 1);
+ assert!(results[0].is_ok());
+ }
+
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn read_null_stream() {
+ let mut machine = MachineBuilder::new()
+ .with_streams(StreamConfig::in_memory())
+ .build();
+
+ let results = machine.run_query("get_code(C).").collect::<Vec<_>>();
+
+ assert_eq!(results.len(), 1);
+ assert!(results[0].is_err());
+ }
+
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn current_output_null_stream() {
+ // TODO: switch to a proper solution for configuring the machine with null streams
+ // once `StreamConfig` supports it.
+ let mut machine = MachineBuilder::new().build();
+ machine.user_output = Stream::Null(StreamOptions::default());
+ machine.configure_streams();
+
+ let results = machine.run_query("current_output(S).").collect::<Vec<_>>();
+
+ assert_eq!(results.len(), 1);
+ assert!(results[0].is_ok());
+ }
+
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn write_null_stream() {
+ // TODO: switch to a proper solution for configuring the machine with null streams
+ // once `StreamConfig` supports it.
+ let mut machine = MachineBuilder::new().build();
+ machine.user_output = Stream::Null(StreamOptions::default());
+ machine.configure_streams();
+
+ let results = machine.run_query("write(hello).").collect::<Vec<_>>();
+
+ assert_eq!(results.len(), 1);
+ assert!(results[0].is_err());
+ }
+
+ /// A variant of the [`write_null_stream`] that tries to write to a (null) input stream.
+ #[test]
+ #[cfg_attr(miri, ignore)]
+ fn write_null_input_stream() {
+ let mut machine = MachineBuilder::new()
+ .with_streams(StreamConfig::in_memory())
+ .build();
+
+ let results = machine
+ .run_query("current_input(Stream), write(Stream, hello).")
+ .collect::<Vec<_>>();
+
+ assert_eq!(results.len(), 1);
+ assert!(results[0].is_err());
+ }
+}
let stream = self.user_input;
if let Some(var) = addr.as_var() {
- self.machine_st.bind(var, stream_as_cell!(stream));
+ self.machine_st.bind(var, stream.into());
return Ok(());
}
let stream = self.user_output;
if let Some(var) = addr.as_var() {
- self.machine_st.bind(var, stream_as_cell!(stream));
+ self.machine_st.bind(var, stream.into());
return Ok(());
}
match stream.write_all(&bytes) {
Ok(_) => {}
_ => {
- let addr = stream_as_cell!(stream);
+ let addr = stream.into();
let err = self
.machine_st
.existence_error(ExistenceError::Stream(addr));
_ => {
let err = self
.machine_st
- .existence_error(ExistenceError::Stream(stream_as_cell!(stream)));
+ .existence_error(ExistenceError::Stream(stream.into()));
return Err(self.machine_st.error_form(err, stub_gen()));
}
return Ok(());
}
_ => {
- let err = self.machine_st.existence_error(ExistenceError::Stream(
- stream_as_cell!(stream),
- ));
+ let err = self
+ .machine_st
+ .existence_error(ExistenceError::Stream(stream.into()));
return Err(self.machine_st.error_form(err, stub_gen()));
}
self.indices.streams = self.indices.streams.sub(&null_streams);
if let Some(first_stream) = first_stream {
- let stream = stream_as_cell!(first_stream);
+ let stream = first_stream.into();
let var = self.deref_register(1).as_var().unwrap();
if let Some(next_stream) = next_stream {
let var = self.deref_register(2).as_var().unwrap();
- let next_stream = stream_as_cell!(next_stream);
- self.machine_st.bind(var, next_stream);
+ self.machine_st.bind(var, next_stream.into());
} else {
self.machine_st.fail = true;
}
if !stream.is_output_stream() {
let stub = functor_stub(atom!("flush_output"), 1);
- let addr = stream_as_cell!(stream);
+ let addr = HeapCellValue::from(stream);
let err =
self.machine_st
if close_result.is_err() {
let stub = functor_stub(atom!("close"), 1);
- let addr = stream_as_cell!(stream);
+ let addr = stream.into();
let err = self
.machine_st
.existence_error(ExistenceError::Stream(addr));
self.indices.streams.insert(stream);
- let stream = stream_as_cell!(stream);
-
let stream_addr = self.deref_register(2);
- self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
+ self.machine_st
+ .bind(stream_addr.as_var().unwrap(), stream.into());
}
Err(_) => {
self.machine_st.fail = true;
*stream.options_mut() = StreamOptions::default();
stream.options_mut().set_stream_type(StreamType::Binary);
self.indices.streams.insert(stream);
- let stream = stream_as_cell!(stream);
+ let stream: HeapCellValue = stream.into();
let handle: TypedArenaPtr<HttpResponse> = arena_alloc!(request.response, &mut self.machine_st.arena);
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
- match_untyped_arena_ptr!(cons_ptr,
- (ArenaHeaderTag::HttpResponse, http_response) => {
- let mut stream = Stream::from_http_sender(
- http_response,
- status_code,
- headers,
- &mut self.machine_st.arena
+ match_untyped_arena_ptr!(cons_ptr,
+ (ArenaHeaderTag::HttpResponse, http_response) => {
+ let mut stream = Stream::from_http_sender(
+ http_response,
+ status_code,
+ headers,
+ &mut self.machine_st.arena
+ );
+ *stream.options_mut() = StreamOptions::default();
+ stream.options_mut().set_stream_type(StreamType::Binary);
+ self.indices.streams.insert(stream);
+ self.machine_st.bind(stream_addr.as_var().unwrap(), stream.into());
+ }
+ _ => {
+ unreachable!();
+ }
);
- *stream.options_mut() = StreamOptions::default();
- stream.options_mut().set_stream_type(StreamType::Binary);
- self.indices.streams.insert(stream);
- let stream = stream_as_cell!(stream);
- self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
- }
- _ => {
- unreachable!();
- }
- );
}
_ => {
- unreachable!();
+ unreachable!();
}
);
let stream_var = self.deref_register(3);
self.machine_st
- .bind(stream_var.as_var().unwrap(), stream_as_cell!(stream));
+ .bind(stream_var.as_var().unwrap(), stream.into());
} else {
let err = self
.machine_st
self.indices.streams.insert(stream);
- stream_as_cell!(stream)
+ HeapCellValue::from(stream)
}
Err(ErrorKind::PermissionDenied) => {
return Err(self.machine_st.open_permission_error(
self.indices.streams.insert(tcp_stream);
- let tcp_stream = stream_as_cell!(tcp_stream);
let client = atom_as_cell!(client);
let client_addr = self.deref_register(2);
let stream_addr = self.deref_register(3);
self.machine_st.bind(client_addr.as_var().unwrap(), client);
- self.machine_st.bind(stream_addr.as_var().unwrap(), tcp_stream);
+ self.machine_st.bind(stream_addr.as_var().unwrap(), tcp_stream.into());
}
None => {
self.machine_st.fail = true;
let stream = Stream::from_tls_stream(addr, stream, &mut self.machine_st.arena);
self.indices.streams.insert(stream);
- self.machine_st.heap.push(stream_as_cell!(stream));
+ // FIXME: why are we pushing a random, unreferenced cell on the heap?
+ self.machine_st.heap.push(stream.into());
let stream_addr = self.deref_register(3);
self.machine_st
- .bind(stream_addr.as_var().unwrap(), stream_as_cell!(stream));
+ .bind(stream_addr.as_var().unwrap(), stream.into());
Ok(())
} else {
let stream_addr = self.deref_register(4);
self.machine_st
- .bind(stream_addr.as_var().unwrap(), stream_as_cell!(stream));
+ .bind(stream_addr.as_var().unwrap(), stream.into());
} else {
unreachable!();
}
let err = self.machine_st.permission_error(
Permission::Reposition,
atom!("stream"),
- stream_as_cell!(stream),
+ HeapCellValue::from(stream),
);
return Err(self.machine_st.error_form(err, stub));
let lib_stream = Stream::from_static_string(library, &mut self.machine_st.arena);
unify!(
self.machine_st,
- stream_as_cell!(lib_stream),
+ HeapCellValue::from(lib_stream),
self.machine_st.registers[2]
);