1742 lines
57 KiB
Rust
1742 lines
57 KiB
Rust
use crate::arena::*;
|
|
use crate::atom_table::*;
|
|
use crate::parser::ast::*;
|
|
use crate::parser::char_reader::*;
|
|
use crate::read::*;
|
|
|
|
use crate::machine::heap::*;
|
|
use crate::machine::machine_errors::*;
|
|
use crate::machine::machine_indices::*;
|
|
use crate::machine::machine_state::*;
|
|
use crate::types::*;
|
|
use crate::http::HttpResponse;
|
|
|
|
pub use modular_bitfield::prelude::*;
|
|
|
|
use std::cmp::Ordering;
|
|
use std::error::Error;
|
|
use std::fmt;
|
|
use std::fmt::Debug;
|
|
use std::fs::{File, OpenOptions};
|
|
use std::hash::{Hash};
|
|
use std::io;
|
|
use std::io::{BufRead, Cursor, ErrorKind, Read, Seek, SeekFrom, Write};
|
|
use std::mem;
|
|
use std::net::{TcpStream, Shutdown};
|
|
use std::ops::{Deref, DerefMut};
|
|
use std::ptr;
|
|
|
|
use native_tls::TlsStream;
|
|
|
|
#[derive(Debug, BitfieldSpecifier, Clone, Copy, PartialEq, Eq, Hash)]
|
|
#[bits = 1]
|
|
pub enum StreamType {
|
|
Binary,
|
|
Text,
|
|
}
|
|
|
|
impl StreamType {
|
|
#[inline]
|
|
pub(crate) fn as_atom(&self) -> Atom {
|
|
match self {
|
|
StreamType::Binary => atom!("binary_stream"),
|
|
StreamType::Text => atom!("text_stream"),
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn as_property_atom(&self) -> Atom {
|
|
match self {
|
|
StreamType::Binary => atom!("binary"),
|
|
StreamType::Text => atom!("text"),
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn other(self) -> StreamType {
|
|
match self {
|
|
StreamType::Binary => StreamType::Text,
|
|
StreamType::Text => StreamType::Binary,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, BitfieldSpecifier, Clone, Copy, PartialEq, Eq, Hash)]
|
|
#[bits = 2]
|
|
pub enum EOFAction {
|
|
EOFCode,
|
|
Error,
|
|
Reset,
|
|
}
|
|
|
|
#[derive(Debug, BitfieldSpecifier, Copy, Clone, PartialEq)]
|
|
#[bits = 2]
|
|
pub(crate) enum AtEndOfStream {
|
|
Not,
|
|
At,
|
|
Past,
|
|
}
|
|
|
|
impl AtEndOfStream {
|
|
#[inline]
|
|
pub(crate) fn as_atom(&self) -> Atom {
|
|
match self {
|
|
AtEndOfStream::Not => atom!("not"),
|
|
AtEndOfStream::Past => atom!("past"),
|
|
AtEndOfStream::At => atom!("at"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl EOFAction {
|
|
#[inline]
|
|
pub(crate) fn as_atom(&self) -> Atom {
|
|
match self {
|
|
EOFAction::EOFCode => atom!("eof_code"),
|
|
EOFAction::Error => atom!("error"),
|
|
EOFAction::Reset => atom!("reset"),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct ByteStream(Cursor<Vec<u8>>);
|
|
|
|
impl ByteStream {
|
|
#[inline(always)]
|
|
pub fn from_string(string: String) -> Self {
|
|
ByteStream(Cursor::new(string.into()))
|
|
}
|
|
}
|
|
|
|
impl Read for ByteStream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.0.read(buf)
|
|
}
|
|
}
|
|
|
|
impl Write for ByteStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
let pos = self.0.position();
|
|
|
|
self.0.seek(SeekFrom::End(0))?;
|
|
let result = self.0.write(buf);
|
|
self.0.seek(SeekFrom::Start(pos))?;
|
|
|
|
result
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
self.0.flush()
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct InputFileStream {
|
|
file_name: Atom,
|
|
file: File,
|
|
}
|
|
|
|
impl Read for InputFileStream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.file.read(buf)
|
|
}
|
|
}
|
|
|
|
impl StreamLayout<CharReader<InputFileStream>> {
|
|
#[inline]
|
|
fn position(&mut self) -> Option<u64> {
|
|
// stream is the internal CharReader. subtract
|
|
// its pending buffer length from position.
|
|
self.get_mut().file.seek(SeekFrom::Current(0))
|
|
.map(|pos| pos - self.stream.rem_buf_len() as u64)
|
|
.ok()
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct OutputFileStream {
|
|
file_name: Atom,
|
|
file: File,
|
|
is_append: bool,
|
|
}
|
|
|
|
impl Write for OutputFileStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
self.file.write(buf)
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
self.file.flush()
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct StaticStringStream {
|
|
stream: Cursor<&'static str>,
|
|
}
|
|
|
|
impl Read for StaticStringStream {
|
|
#[inline(always)]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.stream.read(buf)
|
|
}
|
|
}
|
|
|
|
impl CharRead for StaticStringStream {
|
|
#[inline(always)]
|
|
fn peek_char(&mut self) -> Option<std::io::Result<char>> {
|
|
let pos = self.stream.position() as usize;
|
|
self.stream.get_ref()[pos ..].chars().next().map(Ok)
|
|
}
|
|
|
|
#[inline(always)]
|
|
fn consume(&mut self, nread: usize) {
|
|
self.stream.seek(SeekFrom::Current(nread as i64)).unwrap();
|
|
}
|
|
|
|
#[inline(always)]
|
|
fn put_back_char(&mut self, c: char) {
|
|
self.stream.seek(SeekFrom::Current(- (c.len_utf8() as i64))).unwrap();
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct NamedTcpStream {
|
|
address: Atom,
|
|
tcp_stream: TcpStream,
|
|
}
|
|
|
|
impl Read for NamedTcpStream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.tcp_stream.read(buf)
|
|
}
|
|
}
|
|
|
|
impl Write for NamedTcpStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
self.tcp_stream.write(buf)
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
self.tcp_stream.flush()
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct NamedTlsStream {
|
|
address: Atom,
|
|
tls_stream: TlsStream<Stream>,
|
|
}
|
|
|
|
impl Read for NamedTlsStream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.tls_stream.read(buf)
|
|
}
|
|
}
|
|
|
|
impl Write for NamedTlsStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
self.tls_stream.write(buf)
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
self.tls_stream.flush()
|
|
}
|
|
}
|
|
|
|
pub struct HttpReadStream {
|
|
url: Atom,
|
|
body_reader: Box<dyn BufRead>,
|
|
}
|
|
|
|
impl Debug for HttpReadStream {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
write!(f, "Http Read Stream [{}]", self.url.as_str())
|
|
}
|
|
}
|
|
|
|
impl Read for HttpReadStream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
self.body_reader.read(buf)
|
|
}
|
|
}
|
|
|
|
pub struct HttpWriteStream {
|
|
status_code: u16,
|
|
headers: hyper::HeaderMap,
|
|
response: TypedArenaPtr<HttpResponse>,
|
|
buffer: Vec<u8>,
|
|
}
|
|
|
|
impl Debug for HttpWriteStream {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
write!(f, "Http Write Stream")
|
|
}
|
|
}
|
|
|
|
impl Write for HttpWriteStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
self.buffer.extend_from_slice(buf);
|
|
Ok(buf.len())
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
let (ready, response, cvar) = &**self.response;
|
|
|
|
let mut ready = ready.lock().unwrap();
|
|
{
|
|
let mut response = response.lock().unwrap();
|
|
|
|
let bytes = bytes::Bytes::copy_from_slice(&self.buffer);
|
|
let mut response_ = hyper::Response::builder()
|
|
.status(self.status_code);
|
|
*response_.headers_mut().unwrap() = self.headers.clone();
|
|
*response = Some(response_.body(http_body_util::Full::new(bytes)).unwrap());
|
|
}
|
|
*ready = true;
|
|
cvar.notify_one();
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct StandardOutputStream {}
|
|
|
|
impl Write for StandardOutputStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
io::stdout().write(buf)
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
io::stdout().flush()
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct StandardErrorStream {}
|
|
|
|
impl Write for StandardErrorStream {
|
|
#[inline]
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
io::stderr().write(buf)
|
|
}
|
|
|
|
#[inline]
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
io::stderr().flush()
|
|
}
|
|
}
|
|
|
|
#[bitfield]
|
|
#[repr(u64)]
|
|
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
|
|
pub struct StreamOptions {
|
|
pub stream_type: StreamType,
|
|
pub reposition: bool,
|
|
pub eof_action: EOFAction,
|
|
pub has_alias: bool,
|
|
pub alias: B59,
|
|
}
|
|
|
|
impl StreamOptions {
|
|
#[inline]
|
|
pub fn get_alias(self) -> Option<Atom> {
|
|
if self.has_alias() {
|
|
Some(Atom::from((self.alias() << 3) as usize))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub fn set_alias_to_atom_opt(&mut self, alias: Option<Atom>) {
|
|
self.set_has_alias(alias.is_some());
|
|
|
|
if let Some(alias) = alias {
|
|
self.set_alias(alias.flat_index());
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Default for StreamOptions {
|
|
#[inline]
|
|
fn default() -> Self {
|
|
StreamOptions::new()
|
|
.with_stream_type(StreamType::Text)
|
|
.with_reposition(false)
|
|
.with_eof_action(EOFAction::EOFCode)
|
|
.with_has_alias(false)
|
|
.with_alias(0)
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Copy, Clone)]
|
|
pub struct StreamLayout<T> {
|
|
pub options: StreamOptions,
|
|
pub lines_read: usize,
|
|
past_end_of_stream: bool,
|
|
stream: T,
|
|
}
|
|
|
|
impl<T> StreamLayout<T> {
|
|
#[inline]
|
|
pub fn new(stream: T) -> Self {
|
|
Self {
|
|
options: StreamOptions::default(),
|
|
lines_read: 0,
|
|
past_end_of_stream: false,
|
|
stream,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> Deref for StreamLayout<T> {
|
|
type Target = T;
|
|
|
|
fn deref(&self) -> &Self::Target {
|
|
&self.stream
|
|
}
|
|
}
|
|
|
|
impl<T> DerefMut for StreamLayout<T> {
|
|
fn deref_mut(&mut self) -> &mut Self::Target {
|
|
&mut self.stream
|
|
}
|
|
}
|
|
|
|
macro_rules! arena_allocated_impl_for_stream {
|
|
($stream_type:ty, $stream_tag:ident) => {
|
|
impl ArenaAllocated for StreamLayout<$stream_type> {
|
|
type PtrToAllocated = TypedArenaPtr<StreamLayout<$stream_type>>;
|
|
|
|
#[inline]
|
|
fn tag() -> ArenaHeaderTag {
|
|
ArenaHeaderTag::$stream_tag
|
|
}
|
|
|
|
#[inline]
|
|
fn size(&self) -> usize {
|
|
mem::size_of::<StreamLayout<$stream_type>>()
|
|
}
|
|
|
|
#[inline]
|
|
fn copy_to_arena(self, dst: *mut Self) -> Self::PtrToAllocated {
|
|
unsafe {
|
|
ptr::write(dst, self);
|
|
TypedArenaPtr::new(dst as *mut Self)
|
|
}
|
|
}
|
|
}
|
|
};
|
|
}
|
|
|
|
arena_allocated_impl_for_stream!(CharReader<ByteStream>, ByteStream);
|
|
arena_allocated_impl_for_stream!(CharReader<InputFileStream>, InputFileStream);
|
|
arena_allocated_impl_for_stream!(OutputFileStream, OutputFileStream);
|
|
arena_allocated_impl_for_stream!(CharReader<NamedTcpStream>, NamedTcpStream);
|
|
arena_allocated_impl_for_stream!(CharReader<NamedTlsStream>, NamedTlsStream);
|
|
arena_allocated_impl_for_stream!(CharReader<HttpReadStream>, HttpReadStream);
|
|
arena_allocated_impl_for_stream!(CharReader<HttpWriteStream>, HttpWriteStream);
|
|
arena_allocated_impl_for_stream!(ReadlineStream, ReadlineStream);
|
|
arena_allocated_impl_for_stream!(StaticStringStream, StaticStringStream);
|
|
arena_allocated_impl_for_stream!(StandardOutputStream, StandardOutputStream);
|
|
arena_allocated_impl_for_stream!(StandardErrorStream, StandardErrorStream);
|
|
|
|
#[derive(Debug, Copy, Clone)]
|
|
pub enum Stream {
|
|
Byte(TypedArenaPtr<StreamLayout<CharReader<ByteStream>>>),
|
|
InputFile(TypedArenaPtr<StreamLayout<CharReader<InputFileStream>>>),
|
|
OutputFile(TypedArenaPtr<StreamLayout<OutputFileStream>>),
|
|
StaticString(TypedArenaPtr<StreamLayout<StaticStringStream>>),
|
|
NamedTcp(TypedArenaPtr<StreamLayout<CharReader<NamedTcpStream>>>),
|
|
NamedTls(TypedArenaPtr<StreamLayout<CharReader<NamedTlsStream>>>),
|
|
HttpRead(TypedArenaPtr<StreamLayout<CharReader<HttpReadStream>>>),
|
|
HttpWrite(TypedArenaPtr<StreamLayout<CharReader<HttpWriteStream>>>),
|
|
Null(StreamOptions),
|
|
Readline(TypedArenaPtr<StreamLayout<ReadlineStream>>),
|
|
StandardOutput(TypedArenaPtr<StreamLayout<StandardOutputStream>>),
|
|
StandardError(TypedArenaPtr<StreamLayout<StandardErrorStream>>),
|
|
}
|
|
|
|
impl From<TypedArenaPtr<StreamLayout<ReadlineStream>>> for Stream {
|
|
#[inline]
|
|
fn from(stream: TypedArenaPtr<StreamLayout<ReadlineStream>>) -> Stream {
|
|
Stream::Readline(stream)
|
|
}
|
|
}
|
|
|
|
impl Stream {
|
|
#[inline]
|
|
pub fn from_readline_stream(stream: ReadlineStream, arena: &mut Arena) -> Stream {
|
|
Stream::Readline(arena_alloc!(StreamLayout::new(stream), arena))
|
|
}
|
|
|
|
#[inline]
|
|
pub fn from_owned_string(string: String, arena: &mut Arena) -> Stream {
|
|
Stream::Byte(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(ByteStream(Cursor::new(string.into_bytes())))),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub fn from_static_string(src: &'static str, arena: &mut Arena) -> Stream {
|
|
Stream::StaticString(arena_alloc!(
|
|
StreamLayout::new(StaticStringStream {
|
|
stream: Cursor::new(src)
|
|
}),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub fn stdin(arena: &mut Arena, add_history: bool) -> Stream {
|
|
Stream::Readline(arena_alloc!(
|
|
StreamLayout::new(ReadlineStream::new("", add_history)),
|
|
arena
|
|
))
|
|
}
|
|
|
|
pub fn from_tag(tag: ArenaHeaderTag, ptr: *const u8) -> Self {
|
|
match tag {
|
|
ArenaHeaderTag::ByteStream => Stream::Byte(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::InputFileStream => Stream::InputFile(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::OutputFileStream => {
|
|
Stream::OutputFile(TypedArenaPtr::new(ptr as *mut _))
|
|
}
|
|
ArenaHeaderTag::NamedTcpStream => Stream::NamedTcp(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::NamedTlsStream => Stream::NamedTls(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::HttpReadStream => Stream::HttpRead(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::HttpWriteStream => Stream::HttpWrite(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::ReadlineStream => Stream::Readline(TypedArenaPtr::new(ptr as *mut _)),
|
|
ArenaHeaderTag::StaticStringStream => {
|
|
Stream::StaticString(TypedArenaPtr::new(ptr as *mut _))
|
|
}
|
|
ArenaHeaderTag::StandardOutputStream => {
|
|
Stream::StandardOutput(TypedArenaPtr::new(ptr as *mut _))
|
|
}
|
|
ArenaHeaderTag::StandardErrorStream => {
|
|
Stream::StandardError(TypedArenaPtr::new(ptr as *mut _))
|
|
}
|
|
ArenaHeaderTag::Dropped | ArenaHeaderTag::NullStream => {
|
|
Stream::Null(StreamOptions::default())
|
|
}
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub fn is_stderr(&self) -> bool {
|
|
if let Stream::StandardError(_) = self {
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub fn is_stdout(&self) -> bool {
|
|
if let Stream::StandardOutput(_) = self {
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub fn is_stdin(&self) -> bool {
|
|
if let Stream::Readline(_) = self {
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
pub fn as_ptr(&self) -> *const ArenaHeader {
|
|
match self {
|
|
Stream::Byte(ptr) => ptr.header_ptr(),
|
|
Stream::InputFile(ptr) => ptr.header_ptr(),
|
|
Stream::OutputFile(ptr) => ptr.header_ptr(),
|
|
Stream::StaticString(ptr) => ptr.header_ptr(),
|
|
Stream::NamedTcp(ptr) => ptr.header_ptr(),
|
|
Stream::NamedTls(ptr) => ptr.header_ptr(),
|
|
Stream::HttpRead(ptr) => ptr.header_ptr(),
|
|
Stream::HttpWrite(ptr) => ptr.header_ptr(),
|
|
Stream::Null(_) => ptr::null(),
|
|
Stream::Readline(ptr) => ptr.header_ptr(),
|
|
Stream::StandardOutput(ptr) => ptr.header_ptr(),
|
|
Stream::StandardError(ptr) => ptr.header_ptr(),
|
|
}
|
|
}
|
|
|
|
pub fn options(&self) -> &StreamOptions {
|
|
match self {
|
|
Stream::Byte(ref ptr) => &ptr.options,
|
|
Stream::InputFile(ref ptr) => &ptr.options,
|
|
Stream::OutputFile(ref ptr) => &ptr.options,
|
|
Stream::StaticString(ref ptr) => &ptr.options,
|
|
Stream::NamedTcp(ref ptr) => &ptr.options,
|
|
Stream::NamedTls(ref ptr) => &ptr.options,
|
|
Stream::HttpRead(ref ptr) => &ptr.options,
|
|
Stream::HttpWrite(ref ptr) => &ptr.options,
|
|
Stream::Null(ref options) => options,
|
|
Stream::Readline(ref ptr) => &ptr.options,
|
|
Stream::StandardOutput(ref ptr) => &ptr.options,
|
|
Stream::StandardError(ref ptr) => &ptr.options,
|
|
}
|
|
}
|
|
|
|
pub fn options_mut(&mut self) -> &mut StreamOptions {
|
|
match self {
|
|
Stream::Byte(ref mut ptr) => &mut ptr.options,
|
|
Stream::InputFile(ref mut ptr) => &mut ptr.options,
|
|
Stream::OutputFile(ref mut ptr) => &mut ptr.options,
|
|
Stream::StaticString(ref mut ptr) => &mut ptr.options,
|
|
Stream::NamedTcp(ref mut ptr) => &mut ptr.options,
|
|
Stream::NamedTls(ref mut ptr) => &mut ptr.options,
|
|
Stream::HttpRead(ref mut ptr) => &mut ptr.options,
|
|
Stream::HttpWrite(ref mut ptr) => &mut ptr.options,
|
|
Stream::Null(ref mut options) => options,
|
|
Stream::Readline(ref mut ptr) => &mut ptr.options,
|
|
Stream::StandardOutput(ref mut ptr) => &mut ptr.options,
|
|
Stream::StandardError(ref mut ptr) => &mut ptr.options,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn add_lines_read(&mut self, incr_num_lines_read: usize) {
|
|
match self {
|
|
Stream::Byte(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::InputFile(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::OutputFile(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::StaticString(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::NamedTcp(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::NamedTls(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::HttpRead(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::HttpWrite(_) => {}
|
|
Stream::Null(_) => {}
|
|
Stream::Readline(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::StandardOutput(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
Stream::StandardError(ptr) => ptr.lines_read += incr_num_lines_read,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn set_lines_read(&mut self, value: usize) {
|
|
match self {
|
|
Stream::Byte(ptr) => ptr.lines_read = value,
|
|
Stream::InputFile(ptr) => ptr.lines_read = value,
|
|
Stream::OutputFile(ptr) => ptr.lines_read = value,
|
|
Stream::StaticString(ptr) => ptr.lines_read = value,
|
|
Stream::NamedTcp(ptr) => ptr.lines_read = value,
|
|
Stream::NamedTls(ptr) => ptr.lines_read = value,
|
|
Stream::HttpRead(ptr) => ptr.lines_read = value,
|
|
Stream::HttpWrite(_) => {}
|
|
Stream::Null(_) => {}
|
|
Stream::Readline(ptr) => ptr.lines_read = value,
|
|
Stream::StandardOutput(ptr) => ptr.lines_read = value,
|
|
Stream::StandardError(ptr) => ptr.lines_read = value,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn lines_read(&self) -> usize {
|
|
match self {
|
|
Stream::Byte(ptr) => ptr.lines_read,
|
|
Stream::InputFile(ptr) => ptr.lines_read,
|
|
Stream::OutputFile(ptr) => ptr.lines_read,
|
|
Stream::StaticString(ptr) => ptr.lines_read,
|
|
Stream::NamedTcp(ptr) => ptr.lines_read,
|
|
Stream::NamedTls(ptr) => ptr.lines_read,
|
|
Stream::HttpRead(ptr) => ptr.lines_read,
|
|
Stream::HttpWrite(_) => 0,
|
|
Stream::Null(_) => 0,
|
|
Stream::Readline(ptr) => ptr.lines_read,
|
|
Stream::StandardOutput(ptr) => ptr.lines_read,
|
|
Stream::StandardError(ptr) => ptr.lines_read,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl CharRead for Stream {
|
|
fn peek_char(&mut self) -> Option<std::io::Result<char>> {
|
|
match self {
|
|
Stream::InputFile(file) => (*file).peek_char(),
|
|
Stream::NamedTcp(tcp_stream) => (*tcp_stream).peek_char(),
|
|
Stream::NamedTls(tls_stream) => (*tls_stream).peek_char(),
|
|
Stream::HttpRead(http_stream) => (*http_stream).peek_char(),
|
|
Stream::Readline(rl_stream) => (*rl_stream).peek_char(),
|
|
Stream::StaticString(src) => (*src).peek_char(),
|
|
Stream::Byte(cursor) => (*cursor).peek_char(),
|
|
Stream::OutputFile(_) |
|
|
Stream::StandardError(_) |
|
|
Stream::StandardOutput(_) |
|
|
Stream::HttpWrite(_) |
|
|
Stream::Null(_) => Some(Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::ReadFromOutputStream,
|
|
))),
|
|
}
|
|
}
|
|
|
|
fn read_char(&mut self) -> Option<std::io::Result<char>> {
|
|
match self {
|
|
Stream::InputFile(file) => (*file).read_char(),
|
|
Stream::NamedTcp(tcp_stream) => (*tcp_stream).read_char(),
|
|
Stream::NamedTls(tls_stream) => (*tls_stream).read_char(),
|
|
Stream::HttpRead(http_stream) => (*http_stream).read_char(),
|
|
Stream::Readline(rl_stream) => (*rl_stream).read_char(),
|
|
Stream::StaticString(src) => (*src).read_char(),
|
|
Stream::Byte(cursor) => (*cursor).read_char(),
|
|
Stream::OutputFile(_) |
|
|
Stream::StandardError(_) |
|
|
Stream::StandardOutput(_) |
|
|
Stream::HttpWrite(_) |
|
|
Stream::Null(_) => Some(Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::ReadFromOutputStream,
|
|
))),
|
|
}
|
|
}
|
|
|
|
fn put_back_char(&mut self, c: char) {
|
|
match self {
|
|
Stream::InputFile(file) => file.put_back_char(c),
|
|
Stream::NamedTcp(tcp_stream) => tcp_stream.put_back_char(c),
|
|
Stream::NamedTls(tls_stream) => tls_stream.put_back_char(c),
|
|
Stream::HttpRead(http_stream) => http_stream.put_back_char(c),
|
|
Stream::Readline(rl_stream) => rl_stream.put_back_char(c),
|
|
Stream::StaticString(src) => src.put_back_char(c),
|
|
Stream::Byte(cursor) => cursor.put_back_char(c),
|
|
Stream::OutputFile(_) |
|
|
Stream::StandardError(_) |
|
|
Stream::StandardOutput(_) |
|
|
Stream::HttpWrite(_) |
|
|
Stream::Null(_) => {}
|
|
}
|
|
}
|
|
|
|
fn consume(&mut self, nread: usize) {
|
|
match self {
|
|
Stream::InputFile(ref mut file) => file.consume(nread),
|
|
Stream::NamedTcp(ref mut tcp_stream) => tcp_stream.consume(nread),
|
|
Stream::NamedTls(ref mut tls_stream) => tls_stream.consume(nread),
|
|
Stream::HttpRead(ref mut http_stream) => http_stream.consume(nread),
|
|
Stream::Readline(ref mut rl_stream) => rl_stream.consume(nread),
|
|
Stream::StaticString(ref mut src) => src.consume(nread),
|
|
Stream::Byte(ref mut cursor) => cursor.consume(nread),
|
|
Stream::OutputFile(_) |
|
|
Stream::StandardError(_) |
|
|
Stream::StandardOutput(_) |
|
|
Stream::HttpWrite(_) |
|
|
Stream::Null(_) => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Read for Stream {
|
|
#[inline]
|
|
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
|
let bytes_read = match self {
|
|
Stream::InputFile(file) => (*file).read(buf),
|
|
Stream::NamedTcp(tcp_stream) => (*tcp_stream).read(buf),
|
|
Stream::NamedTls(tls_stream) => (*tls_stream).read(buf),
|
|
Stream::HttpRead(http_stream) => (*http_stream).read(buf),
|
|
Stream::Readline(rl_stream) => (*rl_stream).read(buf),
|
|
Stream::StaticString(src) => (*src).read(buf),
|
|
Stream::Byte(cursor) => (*cursor).read(buf),
|
|
Stream::OutputFile(_)
|
|
| Stream::StandardError(_)
|
|
| Stream::StandardOutput(_)
|
|
| Stream::HttpWrite(_)
|
|
| Stream::Null(_) => Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::ReadFromOutputStream,
|
|
)),
|
|
};
|
|
|
|
bytes_read
|
|
}
|
|
}
|
|
|
|
impl Write for Stream {
|
|
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
|
match self {
|
|
Stream::OutputFile(ref mut file) => file.write(buf),
|
|
Stream::NamedTcp(ref mut tcp_stream) => tcp_stream.get_mut().write(buf),
|
|
Stream::NamedTls(ref mut tls_stream) => tls_stream.get_mut().write(buf),
|
|
Stream::Byte(ref mut cursor) => cursor.get_mut().write(buf),
|
|
Stream::StandardOutput(stream) => stream.write(buf),
|
|
Stream::StandardError(stream) => stream.write(buf),
|
|
Stream::HttpWrite(ref mut stream) => stream.get_mut().write(buf),
|
|
Stream::HttpRead(_) |
|
|
Stream::StaticString(_) |
|
|
Stream::Readline(_) |
|
|
Stream::InputFile(..) |
|
|
Stream::Null(_) => Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::WriteToInputStream,
|
|
)),
|
|
}
|
|
}
|
|
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
match self {
|
|
Stream::OutputFile(ref mut file) => file.stream.flush(),
|
|
Stream::NamedTcp(ref mut tcp_stream) => tcp_stream.stream.get_mut().flush(),
|
|
Stream::NamedTls(ref mut tls_stream) => tls_stream.stream.get_mut().flush(),
|
|
Stream::Byte(ref mut cursor) => cursor.stream.get_mut().flush(),
|
|
Stream::StandardError(stream) => stream.stream.flush(),
|
|
Stream::StandardOutput(stream) => stream.stream.flush(),
|
|
Stream::HttpWrite(ref mut stream) => stream.stream.get_mut().flush(),
|
|
Stream::HttpRead(_) |
|
|
Stream::StaticString(_) |
|
|
Stream::Readline(_) |
|
|
Stream::InputFile(_) |
|
|
Stream::Null(_) => Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::FlushToInputStream,
|
|
)),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
enum StreamError {
|
|
PeekByteFailed,
|
|
PeekByteFromNonPeekableStream,
|
|
#[allow(unused)] PeekCharFailed,
|
|
#[allow(unused)] PeekCharFromNonPeekableStream,
|
|
ReadFromOutputStream,
|
|
WriteToInputStream,
|
|
FlushToInputStream,
|
|
}
|
|
|
|
impl fmt::Display for StreamError {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
match self {
|
|
StreamError::PeekByteFailed => {
|
|
write!(f, "peek byte failed!")
|
|
}
|
|
StreamError::PeekByteFromNonPeekableStream => {
|
|
write!(f, "attempted to peek byte from a non-peekable input stream")
|
|
}
|
|
StreamError::PeekCharFailed => {
|
|
write!(f, "peek char failed!")
|
|
}
|
|
StreamError::PeekCharFromNonPeekableStream => {
|
|
write!(f, "attempted to peek char from a non-peekable input stream")
|
|
}
|
|
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 {}
|
|
|
|
impl PartialOrd for Stream {
|
|
#[inline]
|
|
fn partial_cmp(&self, other: &Stream) -> Option<Ordering> {
|
|
Some(self.cmp(other))
|
|
}
|
|
}
|
|
|
|
impl Ord for Stream {
|
|
#[inline]
|
|
fn cmp(&self, other: &Stream) -> Ordering {
|
|
self.as_ptr().cmp(&other.as_ptr())
|
|
}
|
|
}
|
|
|
|
impl PartialEq for Stream {
|
|
#[inline]
|
|
fn eq(&self, other: &Self) -> bool {
|
|
self.as_ptr() == other.as_ptr()
|
|
}
|
|
}
|
|
|
|
impl Eq for Stream {}
|
|
|
|
impl Stream {
|
|
#[inline]
|
|
pub(crate) fn position(&mut self) -> Option<(u64, usize)> {
|
|
// returns lines_read, position.
|
|
let result = match self {
|
|
Stream::InputFile(file_stream) => {
|
|
file_stream.position()
|
|
}
|
|
Stream::NamedTcp(..)
|
|
| Stream::NamedTls(..)
|
|
| Stream::Readline(..)
|
|
| Stream::StaticString(..)
|
|
| Stream::Byte(..) => Some(0),
|
|
_ => None,
|
|
};
|
|
|
|
result.map(|position| (position, self.lines_read()))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn set_position(&mut self, position: u64) {
|
|
match self {
|
|
Stream::InputFile(stream_layout) => {
|
|
let StreamLayout {
|
|
past_end_of_stream,
|
|
stream,
|
|
..
|
|
} = &mut **stream_layout;
|
|
|
|
stream.get_mut().file.seek(SeekFrom::Start(position)).unwrap();
|
|
stream.reset_buffer(); // flush the internal buffer.
|
|
|
|
if let Ok(metadata) = stream.get_ref().file.metadata() {
|
|
*past_end_of_stream = position > metadata.len();
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn past_end_of_stream(&self) -> bool {
|
|
match self {
|
|
Stream::Byte(stream) => stream.past_end_of_stream,
|
|
Stream::InputFile(stream) => stream.past_end_of_stream,
|
|
Stream::OutputFile(stream) => stream.past_end_of_stream,
|
|
Stream::StaticString(stream) => stream.past_end_of_stream,
|
|
Stream::NamedTcp(stream) => stream.past_end_of_stream,
|
|
Stream::NamedTls(stream) => stream.past_end_of_stream,
|
|
Stream::HttpRead(stream) => stream.past_end_of_stream,
|
|
Stream::HttpWrite(stream) => stream.past_end_of_stream,
|
|
Stream::Null(_) => false,
|
|
Stream::Readline(stream) => stream.past_end_of_stream,
|
|
Stream::StandardOutput(stream) => stream.past_end_of_stream,
|
|
Stream::StandardError(stream) => stream.past_end_of_stream,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn at_end_of_stream(&mut self) -> bool {
|
|
self.position_relative_to_end() == AtEndOfStream::At
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn set_past_end_of_stream(&mut self, value: bool) {
|
|
match self {
|
|
Stream::Byte(stream) => stream.past_end_of_stream = value,
|
|
Stream::InputFile(stream) => stream.past_end_of_stream = value,
|
|
Stream::OutputFile(stream) => stream.past_end_of_stream = value,
|
|
Stream::StaticString(stream) => stream.past_end_of_stream = value,
|
|
Stream::NamedTcp(stream) => stream.past_end_of_stream = value,
|
|
Stream::NamedTls(stream) => stream.past_end_of_stream = value,
|
|
Stream::HttpRead(stream) => stream.past_end_of_stream = value,
|
|
Stream::HttpWrite(stream) => stream.past_end_of_stream = value,
|
|
Stream::Null(_) => {}
|
|
Stream::Readline(stream) => stream.past_end_of_stream = value,
|
|
Stream::StandardOutput(stream) => stream.past_end_of_stream = value,
|
|
Stream::StandardError(stream) => stream.past_end_of_stream = value,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn position_relative_to_end(&mut self) -> AtEndOfStream {
|
|
if self.past_end_of_stream() {
|
|
return AtEndOfStream::Past;
|
|
}
|
|
|
|
if let Stream::InputFile(stream_layout) = self {
|
|
let position = stream_layout.position();
|
|
|
|
let StreamLayout {
|
|
past_end_of_stream,
|
|
stream,
|
|
..
|
|
} = &mut **stream_layout;
|
|
|
|
match stream.get_ref().file.metadata() {
|
|
Ok(metadata) => {
|
|
if let Some(position) = position {
|
|
return match position.cmp(&metadata.len()) {
|
|
Ordering::Equal => AtEndOfStream::At,
|
|
Ordering::Less => AtEndOfStream::Not,
|
|
Ordering::Greater => {
|
|
*past_end_of_stream = true;
|
|
AtEndOfStream::Past
|
|
}
|
|
};
|
|
} else {
|
|
*past_end_of_stream = true;
|
|
AtEndOfStream::Past
|
|
}
|
|
}
|
|
_ => {
|
|
*past_end_of_stream = true;
|
|
AtEndOfStream::Past
|
|
}
|
|
}
|
|
} else {
|
|
AtEndOfStream::Not
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn file_name(&self) -> Option<Atom> {
|
|
match self {
|
|
Stream::InputFile(file) => Some(file.stream.get_ref().file_name),
|
|
Stream::OutputFile(file) => Some(file.stream.file_name),
|
|
Stream::NamedTcp(tcp) => Some(tcp.stream.get_ref().address),
|
|
Stream::NamedTls(tls) => Some(tls.stream.get_ref().address),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn mode(&self) -> Atom {
|
|
match self {
|
|
Stream::Byte(_)
|
|
| Stream::Readline(_)
|
|
| Stream::StaticString(_)
|
|
| Stream::HttpRead(_)
|
|
| Stream::InputFile(..) => atom!("read"),
|
|
Stream::NamedTcp(..) | Stream::NamedTls(..) => atom!("read_append"),
|
|
Stream::OutputFile(file) if file.is_append => atom!("append"),
|
|
Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) | Stream::HttpWrite(_) => atom!("write"),
|
|
Stream::Null(_) => atom!(""),
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub fn stdout(arena: &mut Arena) -> Self {
|
|
Stream::StandardOutput(arena_alloc!(
|
|
StreamLayout::new(StandardOutputStream {}),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub fn stderr(arena: &mut Arena) -> Self {
|
|
Stream::StandardError(arena_alloc!(
|
|
StreamLayout::new(StandardErrorStream {}),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_tcp_stream(
|
|
address: Atom,
|
|
tcp_stream: TcpStream,
|
|
arena: &mut Arena,
|
|
) -> Self {
|
|
tcp_stream.set_read_timeout(None).unwrap();
|
|
tcp_stream.set_write_timeout(None).unwrap();
|
|
|
|
Stream::NamedTcp(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(NamedTcpStream {
|
|
address,
|
|
tcp_stream
|
|
})),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_tls_stream(
|
|
address: Atom,
|
|
tls_stream: TlsStream<Stream>,
|
|
arena: &mut Arena,
|
|
) -> Self {
|
|
Stream::NamedTls(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(NamedTlsStream {
|
|
address,
|
|
tls_stream
|
|
})),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_http_stream(
|
|
url: Atom,
|
|
http_stream: Box<dyn BufRead>,
|
|
arena: &mut Arena,
|
|
) -> Self {
|
|
Stream::HttpRead(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(HttpReadStream {
|
|
url,
|
|
body_reader: http_stream
|
|
})),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_http_sender(
|
|
response: TypedArenaPtr<HttpResponse>,
|
|
status_code: u16,
|
|
headers: hyper::HeaderMap,
|
|
arena: &mut Arena,
|
|
) -> Self {
|
|
Stream::HttpWrite(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(HttpWriteStream {
|
|
response,
|
|
status_code,
|
|
headers,
|
|
buffer: Vec::new(),
|
|
})),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_file_as_output(
|
|
file_name: Atom,
|
|
file: File,
|
|
is_append: bool,
|
|
arena: &mut Arena,
|
|
) -> Self {
|
|
Stream::OutputFile(arena_alloc!(
|
|
StreamLayout::new(OutputFileStream {
|
|
file_name,
|
|
file,
|
|
is_append
|
|
}),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn from_file_as_input(file_name: Atom, file: File, arena: &mut Arena) -> Self {
|
|
Stream::InputFile(arena_alloc!(
|
|
StreamLayout::new(CharReader::new(InputFileStream { file_name, file })),
|
|
arena
|
|
))
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn close(&mut self) -> Result<(), std::io::Error> {
|
|
let mut stream = std::mem::replace(self, Stream::Null(StreamOptions::default()));
|
|
|
|
match stream {
|
|
Stream::NamedTcp(ref mut tcp_stream) => {
|
|
tcp_stream.inner_mut().tcp_stream.shutdown(Shutdown::Both)
|
|
},
|
|
Stream::NamedTls(ref mut tls_stream) => {
|
|
tls_stream.inner_mut().tls_stream.shutdown()
|
|
}
|
|
Stream::HttpRead(ref mut http_stream) => {
|
|
unsafe {
|
|
http_stream.set_tag(ArenaHeaderTag::Dropped);
|
|
std::ptr::drop_in_place(&mut http_stream.inner_mut().body_reader as *mut _);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
Stream::HttpWrite(ref mut http_stream) => {
|
|
unsafe {
|
|
http_stream.set_tag(ArenaHeaderTag::Dropped);
|
|
std::ptr::drop_in_place(&mut http_stream.inner_mut().buffer as *mut _);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
Stream::InputFile(mut file_stream) => {
|
|
// close the stream by dropping the inner File.
|
|
unsafe {
|
|
file_stream.set_tag(ArenaHeaderTag::Dropped);
|
|
std::ptr::drop_in_place(&mut file_stream.inner_mut().file as *mut _);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
Stream::OutputFile(mut file_stream) => {
|
|
// close the stream by dropping the inner File.
|
|
unsafe {
|
|
file_stream.set_tag(ArenaHeaderTag::Dropped);
|
|
std::ptr::drop_in_place(&mut file_stream.file as *mut _);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
_ => Ok(())
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn is_null_stream(&self) -> bool {
|
|
if let Stream::Null(_) = self {
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn is_input_stream(&self) -> bool {
|
|
match self {
|
|
Stream::NamedTcp(..)
|
|
| Stream::NamedTls(..)
|
|
| Stream::HttpRead(..)
|
|
| Stream::Byte(_)
|
|
| Stream::Readline(_)
|
|
| Stream::StaticString(_)
|
|
| Stream::InputFile(..) => true,
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn is_output_stream(&self) -> bool {
|
|
match self {
|
|
Stream::StandardError(_)
|
|
| Stream::StandardOutput(_)
|
|
| Stream::NamedTcp(..)
|
|
| Stream::NamedTls(..)
|
|
| Stream::HttpWrite(..)
|
|
| Stream::Byte(_)
|
|
| Stream::OutputFile(..) => true,
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
// returns true on success.
|
|
#[inline]
|
|
pub(super) fn reset(&mut self) -> bool {
|
|
self.set_lines_read(0);
|
|
self.set_past_end_of_stream(false);
|
|
|
|
loop {
|
|
match self {
|
|
Stream::Byte(ref mut cursor) => {
|
|
cursor.stream.get_mut().0.set_position(0);
|
|
return true;
|
|
}
|
|
Stream::InputFile(ref mut file_stream) => {
|
|
file_stream.stream.get_mut().file.seek(SeekFrom::Start(0)).unwrap();
|
|
return true;
|
|
}
|
|
Stream::Readline(ref mut readline_stream) => {
|
|
readline_stream.reset();
|
|
return true;
|
|
}
|
|
_ => {
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn peek_byte(&mut self) -> std::io::Result<u8> {
|
|
match self {
|
|
Stream::Byte(ref mut cursor) => {
|
|
let mut b = [0u8; 1];
|
|
let pos = cursor.stream.get_mut().0.position();
|
|
|
|
match cursor.read(&mut b)? {
|
|
1 => {
|
|
cursor.stream.get_mut().0.set_position(pos);
|
|
Ok(b[0])
|
|
}
|
|
_ => Err(std::io::Error::new(ErrorKind::UnexpectedEof, "end of file")),
|
|
}
|
|
}
|
|
Stream::InputFile(ref mut file) => {
|
|
match file.peek_byte() {
|
|
Some(result) => {
|
|
Ok(result?)
|
|
}
|
|
_ => Err(std::io::Error::new(
|
|
ErrorKind::UnexpectedEof,
|
|
StreamError::PeekByteFailed,
|
|
)),
|
|
}
|
|
}
|
|
Stream::Readline(ref mut stream) => stream.stream.peek_byte(),
|
|
Stream::NamedTcp(ref mut stream) => {
|
|
let mut b = [0u8; 1];
|
|
stream.stream.get_mut().tcp_stream.peek(&mut b)?;
|
|
Ok(b[0])
|
|
}
|
|
_ => Err(std::io::Error::new(
|
|
ErrorKind::PermissionDenied,
|
|
StreamError::PeekByteFromNonPeekableStream,
|
|
)),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl MachineState {
|
|
#[inline]
|
|
pub(crate) fn eof_action(
|
|
&mut self,
|
|
result: HeapCellValue,
|
|
mut stream: Stream,
|
|
caller: Atom,
|
|
arity: usize,
|
|
) -> CallResult {
|
|
let eof_action = stream.options().eof_action();
|
|
|
|
match eof_action {
|
|
EOFAction::Error => {
|
|
stream.set_past_end_of_stream(true);
|
|
return Err(self.open_past_eos_error(stream, caller, arity));
|
|
}
|
|
EOFAction::EOFCode => {
|
|
let end_of_stream = if stream.options().stream_type() == StreamType::Binary {
|
|
fixnum_as_cell!(Fixnum::build_with(-1))
|
|
} else {
|
|
atom_as_cell!(atom!("end_of_file"))
|
|
};
|
|
|
|
stream.set_past_end_of_stream(true);
|
|
Ok(unify!(self, result, end_of_stream))
|
|
}
|
|
EOFAction::Reset => {
|
|
if !stream.reset() {
|
|
stream.set_past_end_of_stream(true);
|
|
}
|
|
|
|
Ok(self.fail = stream.past_end_of_stream())
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) fn to_stream_options(
|
|
&mut self,
|
|
alias: HeapCellValue,
|
|
eof_action: HeapCellValue,
|
|
reposition: HeapCellValue,
|
|
stream_type: HeapCellValue,
|
|
) -> StreamOptions {
|
|
let alias = read_heap_cell!(self.store(MachineState::deref(self, alias)),
|
|
(HeapCellValueTag::Atom, (name, arity)) => {
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
if name != atom!("[]") {
|
|
Some(name)
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
(HeapCellValueTag::Str, s) => {
|
|
let (name, arity) = cell_as_atom_cell!(self.heap[s])
|
|
.get_name_and_arity();
|
|
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
if name != atom!("[]") {
|
|
Some(name)
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
_ => {
|
|
None
|
|
}
|
|
);
|
|
|
|
let eof_action = read_heap_cell!(self.store(MachineState::deref(self, eof_action)),
|
|
(HeapCellValueTag::Atom, (name, arity)) => {
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
match name {
|
|
atom!("eof_code") => EOFAction::EOFCode,
|
|
atom!("error") => EOFAction::Error,
|
|
atom!("reset") => EOFAction::Reset,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
(HeapCellValueTag::Str, s) => {
|
|
let (name, arity) = cell_as_atom_cell!(self.heap[s])
|
|
.get_name_and_arity();
|
|
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
match name {
|
|
atom!("eof_code") => EOFAction::EOFCode,
|
|
atom!("error") => EOFAction::Error,
|
|
atom!("reset") => EOFAction::Reset,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
_ => {
|
|
unreachable!()
|
|
}
|
|
);
|
|
|
|
let reposition = read_heap_cell!(self.store(MachineState::deref(self, reposition)),
|
|
(HeapCellValueTag::Atom, (name, arity)) => {
|
|
debug_assert_eq!(arity, 0);
|
|
name == atom!("true")
|
|
}
|
|
(HeapCellValueTag::Str, s) => {
|
|
let (name, arity) = cell_as_atom_cell!(self.heap[s])
|
|
.get_name_and_arity();
|
|
|
|
debug_assert_eq!(arity, 0);
|
|
name == atom!("true")
|
|
}
|
|
_ => {
|
|
unreachable!()
|
|
}
|
|
);
|
|
|
|
let stream_type = read_heap_cell!(self.store(MachineState::deref(self, stream_type)),
|
|
(HeapCellValueTag::Atom, (name, arity)) => {
|
|
debug_assert_eq!(arity, 0);
|
|
match name {
|
|
atom!("text") => StreamType::Text,
|
|
atom!("binary") => StreamType::Binary,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
(HeapCellValueTag::Str, s) => {
|
|
let (name, arity) = cell_as_atom_cell!(self.heap[s])
|
|
.get_name_and_arity();
|
|
|
|
debug_assert_eq!(arity, 0);
|
|
match name {
|
|
atom!("text") => StreamType::Text,
|
|
atom!("binary") => StreamType::Binary,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
_ => {
|
|
unreachable!()
|
|
}
|
|
);
|
|
|
|
let mut options = StreamOptions::default();
|
|
|
|
options.set_stream_type(stream_type);
|
|
options.set_reposition(reposition);
|
|
options.set_alias_to_atom_opt(alias);
|
|
options.set_eof_action(eof_action);
|
|
|
|
options
|
|
}
|
|
|
|
pub(crate) fn get_stream_or_alias(
|
|
&mut self,
|
|
addr: HeapCellValue,
|
|
stream_aliases: &StreamAliasDir,
|
|
caller: Atom,
|
|
arity: usize,
|
|
) -> Result<Stream, MachineStub> {
|
|
let addr = self.store(MachineState::deref(self, addr));
|
|
|
|
read_heap_cell!(addr,
|
|
(HeapCellValueTag::Atom, (name, arity)) => {
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
return match stream_aliases.get(&name) {
|
|
Some(stream) if !stream.is_null_stream() => Ok(*stream),
|
|
_ => {
|
|
let stub = functor_stub(caller, arity);
|
|
let addr = atom_as_cell!(name);
|
|
|
|
let existence_error = self.existence_error(ExistenceError::Stream(addr));
|
|
|
|
Err(self.error_form(existence_error, stub))
|
|
}
|
|
};
|
|
}
|
|
(HeapCellValueTag::Str, s) => {
|
|
let (name, arity) = cell_as_atom_cell!(self.heap[s])
|
|
.get_name_and_arity();
|
|
|
|
debug_assert_eq!(arity, 0);
|
|
|
|
return match stream_aliases.get(&name) {
|
|
Some(stream) if !stream.is_null_stream() => Ok(*stream),
|
|
_ => {
|
|
let stub = functor_stub(caller, arity);
|
|
let addr = atom_as_cell!(name);
|
|
|
|
let existence_error = self.existence_error(ExistenceError::Stream(addr));
|
|
|
|
Err(self.error_form(existence_error, stub))
|
|
}
|
|
};
|
|
}
|
|
(HeapCellValueTag::Cons, ptr) => {
|
|
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))
|
|
} else {
|
|
Ok(stream)
|
|
};
|
|
}
|
|
(ArenaHeaderTag::Dropped, _value) => {
|
|
let stub = functor_stub(caller, arity);
|
|
let err = self.existence_error(ExistenceError::Stream(addr));
|
|
|
|
return Err(self.error_form(err, stub));
|
|
}
|
|
_ => {
|
|
}
|
|
);
|
|
}
|
|
_ => {
|
|
}
|
|
);
|
|
|
|
let stub = functor_stub(caller, arity);
|
|
|
|
if addr.is_var() {
|
|
let instantiation_error = self.instantiation_error();
|
|
Err(self.error_form(instantiation_error, stub))
|
|
} else {
|
|
let domain_error = self.domain_error(DomainErrorType::StreamOrAlias, addr);
|
|
Err(self.error_form(domain_error, stub))
|
|
}
|
|
}
|
|
|
|
pub(crate) fn open_parsing_stream(&mut self, mut stream: Stream) -> Result<Stream, ParserError> {
|
|
match stream.peek_char() {
|
|
None => Ok(stream), // empty stream is handled gracefully by Lexer::eof
|
|
Some(Err(e)) => Err(ParserError::IO(e)),
|
|
Some(Ok(c)) => {
|
|
if c == '\u{feff}' {
|
|
// skip UTF-8 BOM
|
|
stream.consume(c.len_utf8());
|
|
}
|
|
|
|
Ok(stream)
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) fn stream_permission_error(
|
|
&mut self,
|
|
perm: Permission,
|
|
err_atom: Atom,
|
|
stream: Stream,
|
|
caller: Atom,
|
|
arity: usize,
|
|
) -> MachineStub {
|
|
let stub = functor_stub(caller, arity);
|
|
let err = self.permission_error(
|
|
perm,
|
|
err_atom,
|
|
if let Some(alias) = stream.options().get_alias() {
|
|
atom_as_cell!(alias)
|
|
} else {
|
|
stream_as_cell!(stream)
|
|
},
|
|
);
|
|
|
|
self.error_form(err, stub)
|
|
}
|
|
|
|
#[inline]
|
|
pub(crate) fn open_past_eos_error(
|
|
&mut self,
|
|
stream: Stream,
|
|
caller: Atom,
|
|
arity: usize,
|
|
) -> MachineStub {
|
|
self.stream_permission_error(
|
|
Permission::InputStream,
|
|
atom!("past_end_of_stream"),
|
|
stream,
|
|
caller,
|
|
arity,
|
|
)
|
|
}
|
|
|
|
pub(crate) fn open_permission_error<T: PermissionError>(
|
|
&mut self,
|
|
culprit: T,
|
|
stub_name: Atom,
|
|
stub_arity: usize,
|
|
) -> MachineStub {
|
|
let stub = functor_stub(stub_name, stub_arity);
|
|
let err = self.permission_error(Permission::Open, atom!("source_sink"), culprit);
|
|
|
|
self.error_form(err, stub)
|
|
}
|
|
|
|
pub(crate) fn occupied_alias_permission_error(
|
|
&mut self,
|
|
alias: Atom,
|
|
stub_name: Atom,
|
|
stub_arity: usize,
|
|
) -> MachineStub {
|
|
let stub = functor_stub(stub_name, stub_arity);
|
|
|
|
let err = self.permission_error(
|
|
Permission::Open,
|
|
atom!("source_sink"),
|
|
functor!(atom!("alias"), [atom(alias)]),
|
|
);
|
|
|
|
self.error_form(err, stub)
|
|
}
|
|
|
|
pub(crate) fn reposition_error(&mut self, stub_name: Atom, stub_arity: usize) -> MachineStub {
|
|
let stub = functor_stub(stub_name, stub_arity);
|
|
let rep_stub = functor!(atom!("reposition"), [atom(atom!("true"))]);
|
|
let err = self.permission_error(Permission::Open, atom!("source_sink"), rep_stub);
|
|
|
|
self.error_form(err, stub)
|
|
}
|
|
|
|
pub(crate) fn check_stream_properties(
|
|
&mut self,
|
|
stream: Stream,
|
|
expected_type: StreamType,
|
|
input: Option<HeapCellValue>,
|
|
caller: Atom,
|
|
arity: usize,
|
|
) -> CallResult {
|
|
let opt_err = if input.is_some() && !stream.is_input_stream() {
|
|
Some(atom!("stream")) // 8.14.2.3 g)
|
|
} else if input.is_none() && !stream.is_output_stream() {
|
|
Some(atom!("stream")) // 8.14.2.3 g)
|
|
} else if stream.options().stream_type() != expected_type {
|
|
Some(expected_type.other().as_atom()) // 8.14.2.3 h)
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let permission = if input.is_some() {
|
|
Permission::InputStream
|
|
} else {
|
|
Permission::OutputStream
|
|
};
|
|
|
|
if let Some(err_atom) = opt_err {
|
|
return Err(self.stream_permission_error(permission, err_atom, stream, caller, arity));
|
|
}
|
|
|
|
if let Some(input) = input {
|
|
if stream.past_end_of_stream() {
|
|
self.eof_action(input, stream, caller, arity)?;
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn stream_from_file_spec(
|
|
&mut self,
|
|
file_spec: Atom,
|
|
indices: &mut IndexStore,
|
|
options: &StreamOptions,
|
|
) -> Result<Stream, MachineStub> {
|
|
if file_spec == atom!("") {
|
|
let stub = functor_stub(atom!("open"), 4);
|
|
let err = self.domain_error(DomainErrorType::SourceSink, self[temp_v!(1)]);
|
|
|
|
return Err(self.error_form(err, stub));
|
|
}
|
|
|
|
// 8.11.5.3l)
|
|
if let Some(alias) = options.get_alias() {
|
|
if indices.stream_aliases.contains_key(&alias) {
|
|
return Err(self.occupied_alias_permission_error(alias, atom!("open"), 4));
|
|
}
|
|
}
|
|
|
|
let mode = MachineState::deref(self, self[temp_v!(2)]);
|
|
let mode = cell_as_atom!(self.store(mode));
|
|
|
|
let mut open_options = OpenOptions::new();
|
|
|
|
let (is_input_file, in_append_mode) = match mode {
|
|
atom!("read") => {
|
|
open_options.read(true).write(false).create(false);
|
|
(true, false)
|
|
}
|
|
atom!("write") => {
|
|
open_options
|
|
.read(false)
|
|
.write(true)
|
|
.truncate(true)
|
|
.create(true);
|
|
|
|
(false, false)
|
|
}
|
|
atom!("append") => {
|
|
open_options
|
|
.read(false)
|
|
.write(true)
|
|
.create(true)
|
|
.append(true);
|
|
|
|
(false, true)
|
|
}
|
|
_ => {
|
|
let stub = functor_stub(atom!("open"), 4);
|
|
let err = self.domain_error(DomainErrorType::IOMode, self[temp_v!(2)]);
|
|
|
|
// 8.11.5.3h)
|
|
return Err(self.error_form(err, stub));
|
|
}
|
|
};
|
|
|
|
let file = match open_options.open(file_spec.as_str()) {
|
|
Ok(file) => file,
|
|
Err(err) => {
|
|
match err.kind() {
|
|
ErrorKind::NotFound => {
|
|
// 8.11.5.3j)
|
|
let stub = functor_stub(atom!("open"), 4);
|
|
|
|
let err = self.existence_error(
|
|
ExistenceError::SourceSink(self[temp_v!(1)]),
|
|
);
|
|
|
|
return Err(self.error_form(err, stub));
|
|
}
|
|
ErrorKind::PermissionDenied => {
|
|
// 8.11.5.3k)
|
|
return Err(self.open_permission_error(self.registers[1], atom!("open"), 4));
|
|
}
|
|
_ => {
|
|
let stub = functor_stub(atom!("open"), 4);
|
|
let err = self.syntax_error(ParserError::IO(err));
|
|
|
|
return Err(self.error_form(err, stub));
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
Ok(if is_input_file {
|
|
Stream::from_file_as_input(file_spec, file, &mut self.arena)
|
|
} else {
|
|
Stream::from_file_as_output(file_spec, file, in_append_mode, &mut self.arena)
|
|
})
|
|
}
|
|
}
|