rishi/scryer-prolog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

http_open/3 with method option

Browse Source
This commit is contained in:
Adrián Arroyo Calle
2022-03-19 22:53:24 +01:00
parent c45cdd6ea0
commit 3acbe2a418
11 changed files with 654 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/machine/dispatch.rs
View File

@@ -4127,6 +4127,14 @@ impl Machine {
try_or_throw!(self.machine_st, self.det_length_rundown());
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpOpen(_) => {
self.http_open();
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpOpen(_) => {
self.http_open();
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallCurrentTime(_) => {
self.current_time();
step_or_fail!(self, self.machine_st.p += 1);

6
src/machine/mock_wam.rs
View File

@@ -224,6 +224,11 @@ impl Machine {
let user_output = Stream::from_owned_string("".to_owned(), &mut machine_st.arena);
let user_error = Stream::stderr(&mut machine_st.arena);
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut wam = Machine {
machine_st,
indices: IndexStore::new(),
@@ -232,6 +237,7 @@ impl Machine {
user_output,
user_error,
load_contexts: vec![],
runtime
};
let mut lib_path = current_dir();

8
src/machine/mod.rs
View File

@@ -48,6 +48,7 @@ use std::cmp::Ordering;
use std::env;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use tokio::runtime::Runtime;
lazy_static! {
pub static ref INTERRUPT: AtomicBool = AtomicBool::new(false);
@@ -62,6 +63,7 @@ pub struct Machine {
pub(super) user_output: Stream,
pub(super) user_error: Stream,
pub(super) load_contexts: Vec<LoadContext>,
pub(super) runtime: Runtime,
}
#[derive(Debug)]
@@ -405,6 +407,11 @@ impl Machine {
let user_output = Stream::stdout(&mut machine_st.arena);
let user_error = Stream::stderr(&mut machine_st.arena);
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut wam = Machine {
machine_st,
indices: IndexStore::new(),
@@ -413,6 +420,7 @@ impl Machine {
user_output,
user_error,
load_contexts: vec![],
runtime,
};
let mut lib_path = current_dir();

56
src/machine/streams.rs
View File

@@ -15,10 +15,11 @@ 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::{Cursor, ErrorKind, Read, Seek, SeekFrom, Write};
use std::io::{BufRead, Cursor, ErrorKind, Read, Seek, SeekFrom, Write};
use std::mem;
use std::net::{TcpStream, Shutdown};
use std::ops::{Deref, DerefMut};
@@ -237,6 +238,24 @@ impl Write for NamedTlsStream {
}
}
pub struct NamedHttpClientStream {
url: Atom,
body_reader: Box<dyn BufRead>,
}
impl Debug for NamedHttpClientStream {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Http Client Stream [{}]", self.url.as_str())
}
}
impl Read for NamedHttpClientStream {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.body_reader.read(buf)
}
}
#[derive(Debug)]
pub struct StandardOutputStream {}
@@ -375,6 +394,7 @@ 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<NamedHttpClientStream>, NamedHttpClientStream);
arena_allocated_impl_for_stream!(ReadlineStream, ReadlineStream);
arena_allocated_impl_for_stream!(StaticStringStream, StaticStringStream);
arena_allocated_impl_for_stream!(StandardOutputStream, StandardOutputStream);
@@ -388,6 +408,7 @@ pub enum Stream {
StaticString(TypedArenaPtr<StreamLayout<StaticStringStream>>),
NamedTcp(TypedArenaPtr<StreamLayout<CharReader<NamedTcpStream>>>),
NamedTls(TypedArenaPtr<StreamLayout<CharReader<NamedTlsStream>>>),
NamedHttpClient(TypedArenaPtr<StreamLayout<CharReader<NamedHttpClientStream>>>),
Null(StreamOptions),
Readline(TypedArenaPtr<StreamLayout<ReadlineStream>>),
StandardOutput(TypedArenaPtr<StreamLayout<StandardOutputStream>>),
@@ -442,6 +463,7 @@ impl Stream {
}
ArenaHeaderTag::NamedTcpStream => Stream::NamedTcp(TypedArenaPtr::new(ptr as *mut _)),
ArenaHeaderTag::NamedTlsStream => Stream::NamedTls(TypedArenaPtr::new(ptr as *mut _)),
ArenaHeaderTag::NamedHttpClientStream => Stream::NamedHttpClient(TypedArenaPtr::new(ptr as *mut _)),
ArenaHeaderTag::ReadlineStream => Stream::Readline(TypedArenaPtr::new(ptr as *mut _)),
ArenaHeaderTag::StaticStringStream => {
Stream::StaticString(TypedArenaPtr::new(ptr as *mut _))
@@ -492,6 +514,7 @@ impl Stream {
Stream::StaticString(ptr) => ptr.header_ptr(),
Stream::NamedTcp(ptr) => ptr.header_ptr(),
Stream::NamedTls(ptr) => ptr.header_ptr(),
Stream::NamedHttpClient(ptr) => ptr.header_ptr(),
Stream::Null(_) => ptr::null(),
Stream::Readline(ptr) => ptr.header_ptr(),
Stream::StandardOutput(ptr) => ptr.header_ptr(),
@@ -507,6 +530,7 @@ impl Stream {
Stream::StaticString(ref ptr) => &ptr.options,
Stream::NamedTcp(ref ptr) => &ptr.options,
Stream::NamedTls(ref ptr) => &ptr.options,
Stream::NamedHttpClient(ref ptr) => &ptr.options,
Stream::Null(ref options) => options,
Stream::Readline(ref ptr) => &ptr.options,
Stream::StandardOutput(ref ptr) => &ptr.options,
@@ -522,6 +546,7 @@ impl Stream {
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::NamedHttpClient(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,
@@ -538,6 +563,7 @@ impl Stream {
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::NamedHttpClient(ptr) => ptr.lines_read += incr_num_lines_read,
Stream::Null(_) => {}
Stream::Readline(ptr) => ptr.lines_read += incr_num_lines_read,
Stream::StandardOutput(ptr) => ptr.lines_read += incr_num_lines_read,
@@ -554,6 +580,7 @@ impl Stream {
Stream::StaticString(ptr) => ptr.lines_read = value,
Stream::NamedTcp(ptr) => ptr.lines_read = value,
Stream::NamedTls(ptr) => ptr.lines_read = value,
Stream::NamedHttpClient(ptr) => ptr.lines_read = value,
Stream::Null(_) => {}
Stream::Readline(ptr) => ptr.lines_read = value,
Stream::StandardOutput(ptr) => ptr.lines_read = value,
@@ -570,6 +597,7 @@ impl Stream {
Stream::StaticString(ptr) => ptr.lines_read,
Stream::NamedTcp(ptr) => ptr.lines_read,
Stream::NamedTls(ptr) => ptr.lines_read,
Stream::NamedHttpClient(ptr) => ptr.lines_read,
Stream::Null(_) => 0,
Stream::Readline(ptr) => ptr.lines_read,
Stream::StandardOutput(ptr) => ptr.lines_read,
@@ -584,6 +612,7 @@ impl CharRead for Stream {
Stream::InputFile(file) => (*file).peek_char(),
Stream::NamedTcp(tcp_stream) => (*tcp_stream).peek_char(),
Stream::NamedTls(tls_stream) => (*tls_stream).peek_char(),
Stream::NamedHttpClient(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(),
@@ -602,6 +631,7 @@ impl CharRead for Stream {
Stream::InputFile(file) => (*file).read_char(),
Stream::NamedTcp(tcp_stream) => (*tcp_stream).read_char(),
Stream::NamedTls(tls_stream) => (*tls_stream).read_char(),
Stream::NamedHttpClient(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(),
@@ -620,6 +650,7 @@ impl CharRead for Stream {
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::NamedHttpClient(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),
@@ -635,6 +666,7 @@ impl CharRead for Stream {
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::NamedHttpClient(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),
@@ -653,6 +685,7 @@ impl Read for Stream {
Stream::InputFile(file) => (*file).read(buf),
Stream::NamedTcp(tcp_stream) => (*tcp_stream).read(buf),
Stream::NamedTls(tls_stream) => (*tls_stream).read(buf),
Stream::NamedHttpClient(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),
@@ -678,6 +711,7 @@ impl Write for Stream {
Stream::Byte(ref mut cursor) => cursor.get_mut().write(buf),
Stream::StandardOutput(stream) => stream.write(buf),
Stream::StandardError(stream) => stream.write(buf),
Stream::NamedHttpClient(_) |
Stream::StaticString(_) |
Stream::Readline(_) |
Stream::InputFile(..) |
@@ -696,6 +730,7 @@ impl Write for Stream {
Stream::Byte(ref mut cursor) => cursor.stream.get_mut().flush(),
Stream::StandardError(stream) => stream.stream.flush(),
Stream::StandardOutput(stream) => stream.stream.flush(),
Stream::NamedHttpClient(_) |
Stream::StaticString(_) |
Stream::Readline(_) |
Stream::InputFile(_) |
@@ -820,6 +855,7 @@ impl 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::NamedHttpClient(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,
@@ -841,6 +877,7 @@ impl Stream {
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::NamedHttpClient(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,
@@ -904,6 +941,7 @@ impl Stream {
Stream::Byte(_)
| Stream::Readline(_)
| Stream::StaticString(_)
| Stream::NamedHttpClient(_)
| Stream::InputFile(..) => atom!("read"),
Stream::NamedTcp(..) | Stream::NamedTls(..) => atom!("read_append"),
Stream::OutputFile(file) if file.is_append => atom!("append"),
@@ -961,6 +999,21 @@ impl Stream {
))
}
#[inline]
pub(crate) fn from_http_stream(
url: Atom,
http_stream: Box<dyn BufRead>,
arena: &mut Arena,
) -> Self {
Stream::NamedHttpClient(arena_alloc!(
StreamLayout::new(CharReader::new(NamedHttpClientStream {
url,
body_reader: http_stream
})),
arena
))
}
#[inline]
pub(crate) fn from_file_as_output(
file_name: Atom,
@@ -1033,6 +1086,7 @@ impl Stream {
match self {
Stream::NamedTcp(..)
| Stream::NamedTls(..)
| Stream::NamedHttpClient(..)
| Stream::Byte(_)
| Stream::Readline(_)
| Stream::StaticString(_)

76
src/machine/system_calls.rs
View File

@@ -73,6 +73,10 @@ use base64;
use roxmltree;
use select;
use hyper::{Body, Client, Method, Request, Uri};
use hyper::body::Buf;
use hyper_tls::HttpsConnector;
ref_thread_local! {
pub(crate) static managed RANDOM_STATE: RandState<'static> = RandState::new();
}
@@ -3263,6 +3267,78 @@ impl Machine {
Ok(())
}
#[inline(always)]
pub(crate) fn http_open(&mut self) -> CallResult {
let address_sink = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[1]));
let method = read_heap_cell!(self.machine_st.store(self.machine_st.deref(self.machine_st.registers[3])),
(HeapCellValueTag::Atom, (name, arity)) => {
debug_assert_eq!(arity, 0);
match name {
atom!("get") => Method::GET,
atom!("post") => Method::POST,
atom!("put") => Method::PUT,
atom!("delete") => Method::DELETE,
atom!("patch") => Method::PATCH,
atom!("head") => Method::HEAD,
_ => unreachable!(),
}
}
_ => {
unreachable!()
}
);
if let Some(address_sink) = self.machine_st.value_to_str_like(address_sink) {
let address_string = match address_sink {
AtomOrString::Atom(atom) => {
String::from(atom.as_str())
}
AtomOrString::String(string) => {
String::from(string.as_str())
}
};
let address: Uri = address_string.parse().unwrap();
let stream = self.runtime.block_on(async {
let https = HttpsConnector::new();
let client = Client::builder()
.build::<_, hyper::Body>(https);
let req = Request::builder()
.method(method)
.uri(address)
.body(Body::empty())
.unwrap();
let mut resp = client.request(req).await.unwrap();
let buf = hyper::body::aggregate(resp).await.unwrap();
let mut reader = buf.reader();
let mut stream = Stream::from_http_stream(
self.machine_st.atom_tbl.build_with(&address_string),
Box::new(reader),
&mut self.machine_st.arena
);
*stream.options_mut() = StreamOptions::default();
if let Some(alias) = stream.options().get_alias() {
self.indices.stream_aliases.insert(alias, stream);
}
self.indices.streams.insert(stream);
stream_as_cell!(stream)
});
let stream_addr = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[2]));
self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
} else {
let err = self.machine_st.domain_error(DomainErrorType::SourceSink, address_sink);
let stub = functor_stub(atom!("http_open"), 3);
return Err(self.machine_st.error_form(err, stub));
}
Ok(())
}
#[inline(always)]
pub(crate) fn current_time(&mut self) {
let timestamp = self.systemtime_to_timestamp(SystemTime::now());