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

Merge pull request #1373 from aarroyoc/http-open

Browse Source 
Add Hyper based `http_open/3`
This commit is contained in:
Mark Thom
2022-03-28 23:10:08 -06:00
committed by GitHub
parent 48960f1beb 45c35eca3e
commit 32bdedde6d
11 changed files with 740 additions and 124 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(_) => {
try_or_throw!(self.machine_st, self.http_open());
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpOpen(_) => {
try_or_throw!(self.machine_st, 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();

64
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 _))
@@ -494,6 +516,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(),
@@ -509,6 +532,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,
@@ -524,6 +548,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,
@@ -540,6 +565,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,
@@ -556,6 +582,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,
@@ -572,6 +599,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,
@@ -586,6 +614,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(),
@@ -604,6 +633,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(),
@@ -622,6 +652,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),
@@ -637,6 +668,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),
@@ -655,6 +687,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),
@@ -680,6 +713,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(..) |
@@ -698,6 +732,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(_) |
@@ -822,6 +857,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,
@@ -843,6 +879,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,
@@ -906,6 +943,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"),
@@ -963,6 +1001,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,
@@ -999,6 +1052,14 @@ impl Stream {
Stream::NamedTls(ref mut tls_stream) => {
tls_stream.inner_mut().tls_stream.shutdown()
}
Stream::NamedHttpClient(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::InputFile(mut file_stream) => {
// close the stream by dropping the inner File.
unsafe {
@@ -1035,6 +1096,7 @@ impl Stream {
match self {
Stream::NamedTcp(..)
| Stream::NamedTls(..)
| Stream::NamedHttpClient(..)
| Stream::Byte(_)
| Stream::Readline(_)
| Stream::StaticString(_)

129
src/machine/system_calls.rs
View File

@@ -43,6 +43,7 @@ use std::mem;
use std::net::{TcpListener, TcpStream};
use std::num::NonZeroU32;
use std::ops::Sub;
use std::str::FromStr;
use std::process;
use chrono::{offset::Local, DateTime};
@@ -73,6 +74,11 @@ use base64;
use roxmltree;
use select;
use hyper::{Body, Client, HeaderMap, Method, Request, Uri};
use hyper::header::{HeaderName, HeaderValue};
use hyper::body::Buf;
use hyper_tls::HttpsConnector;
ref_thread_local! {
pub(crate) static managed RANDOM_STATE: RandState<'static> = RandState::new();
}
@@ -3263,6 +3269,129 @@ 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!()
}
);
let address_status = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[4]));
let address_data = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[5]));
let mut bytes: Vec<u8> = Vec::new();
if let Some(string) = self.machine_st.value_to_str_like(address_data) {
bytes = string.as_str().bytes().collect();
}
let stub_gen = || functor_stub(atom!("http_open"), 3);
let headers = match self.machine_st.try_from_list(self.machine_st.registers[7], stub_gen) {
Ok(addrs) => {
let mut header_map = HeaderMap::new();
for heap_cell in addrs{
read_heap_cell!(heap_cell,
(HeapCellValueTag::Str, s) => {
let name = cell_as_atom_cell!(self.machine_st.heap[s]).get_name();
let value = self.machine_st.value_to_str_like(self.machine_st.heap[s + 1]).unwrap();
header_map.insert(HeaderName::from_str(name.as_str()).unwrap(), HeaderValue::from_str(value.as_str()).unwrap());
}
_ => {
unreachable!()
}
)
}
header_map
},
Err(e) => return Err(e)
};
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);
// request
let mut req = Request::builder()
.method(method)
.uri(address)
.body(Body::from(bytes))
.unwrap();
// request headers
*req.headers_mut() = headers;
// do it!
let resp = client.request(req).await.unwrap();
// status code
let status = resp.status().as_u16();
self.machine_st.unify_fixnum(Fixnum::build_with(status as i64), address_status);
// headers
let headers: Vec<HeapCellValue> = resp.headers().iter().map(|(header_name, header_value)| {
let h = self.machine_st.heap.len();
let header_term = functor!(
self.machine_st.atom_tbl.build_with(header_name.as_str()),
[cell(string_as_cstr_cell!(self.machine_st.atom_tbl.build_with(header_value.to_str().unwrap())))]
);
self.machine_st.heap.extend(header_term.into_iter());
str_loc_as_cell!(h)
}).collect();
let headers_list = iter_to_heap_list(&mut self.machine_st.heap, headers.into_iter());
unify!(self.machine_st, heap_loc_as_cell!(headers_list), self.machine_st.registers[6]);
// body
let buf = hyper::body::aggregate(resp).await.unwrap();
let 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());