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

HTTP Server 2.0

Browse Source
This commit is contained in:
Adrián Arroyo Calle
2022-08-11 20:25:19 +02:00
committed by Mark Thom
parent 91d4e91f53
commit 181be5be3f
12 changed files with 602 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/machine/dispatch.rs
View File

@@ -4197,6 +4197,30 @@ impl Machine {
try_or_throw!(self.machine_st, self.http_open());
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpListen(_) => {
try_or_throw!(self.machine_st, self.http_listen());
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpListen(_) => {
try_or_throw!(self.machine_st, self.http_listen());
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpAccept(_) => {
try_or_throw!(self.machine_st, self.http_accept());
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpAccept(_) => {
try_or_throw!(self.machine_st, self.http_accept());
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpAnswer(_) => {
try_or_throw!(self.machine_st, self.http_answer());
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpAnswer(_) => {
try_or_throw!(self.machine_st, self.http_answer());
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);

4
src/machine/mod.rs
View File

@@ -431,9 +431,7 @@ 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()
let runtime = tokio::runtime::Runtime::new()
.unwrap();
let mut wam = Machine {

128
src/machine/streams.rs
View File

@@ -26,6 +26,7 @@ use std::ops::{Deref, DerefMut};
use std::ptr;
use native_tls::TlsStream;
use hyper::body::{Bytes, Sender};
#[derive(Debug, BitfieldSpecifier, Clone, Copy, PartialEq, Eq, Hash)]
#[bits = 1]
@@ -249,24 +250,51 @@ impl Write for NamedTlsStream {
}
}
pub struct NamedHttpClientStream {
pub struct HttpReadStream {
url: Atom,
body_reader: Box<dyn BufRead>,
}
impl Debug for NamedHttpClientStream {
impl Debug for HttpReadStream {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Http Client Stream [{}]", self.url.as_str())
write!(f, "Http Read Stream [{}]", self.url.as_str())
}
}
impl Read for NamedHttpClientStream {
impl Read for HttpReadStream {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.body_reader.read(buf)
}
}
pub struct HttpWriteStream {
body_writer: Sender,
}
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> {
let bytes = Bytes::copy_from_slice(buf);
let len = bytes.len();
match self.body_writer.try_send_data(bytes) {
Ok(()) => Ok(len),
Err(_) => Err(std::io::Error::from(ErrorKind::Interrupted))
}
}
#[inline]
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[derive(Debug)]
pub struct StandardOutputStream {}
@@ -405,7 +433,8 @@ 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!(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);
@@ -419,7 +448,8 @@ pub enum Stream {
StaticString(TypedArenaPtr<StreamLayout<StaticStringStream>>),
NamedTcp(TypedArenaPtr<StreamLayout<CharReader<NamedTcpStream>>>),
NamedTls(TypedArenaPtr<StreamLayout<CharReader<NamedTlsStream>>>),
NamedHttpClient(TypedArenaPtr<StreamLayout<CharReader<NamedHttpClientStream>>>),
HttpRead(TypedArenaPtr<StreamLayout<CharReader<HttpReadStream>>>),
HttpWrite(TypedArenaPtr<StreamLayout<CharReader<HttpWriteStream>>>),
Null(StreamOptions),
Readline(TypedArenaPtr<StreamLayout<ReadlineStream>>),
StandardOutput(TypedArenaPtr<StreamLayout<StandardOutputStream>>),
@@ -474,7 +504,8 @@ 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::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 _))
@@ -527,7 +558,8 @@ 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::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(),
@@ -543,7 +575,8 @@ 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::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,
@@ -559,7 +592,8 @@ 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::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,
@@ -576,7 +610,8 @@ 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::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,
@@ -593,7 +628,8 @@ 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::HttpRead(ptr) => ptr.lines_read = value,
Stream::HttpWrite(_) => {}
Stream::Null(_) => {}
Stream::Readline(ptr) => ptr.lines_read = value,
Stream::StandardOutput(ptr) => ptr.lines_read = value,
@@ -610,7 +646,8 @@ 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::HttpRead(ptr) => ptr.lines_read,
Stream::HttpWrite(_) => 0,
Stream::Null(_) => 0,
Stream::Readline(ptr) => ptr.lines_read,
Stream::StandardOutput(ptr) => ptr.lines_read,
@@ -625,13 +662,14 @@ 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::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,
@@ -644,13 +682,14 @@ 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::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,
@@ -663,13 +702,14 @@ 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::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(_) => {}
}
}
@@ -679,13 +719,14 @@ 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::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(_) => {}
}
}
@@ -698,13 +739,14 @@ 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::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::StandardOutput(_)
| Stream::HttpWrite(_)
| Stream::Null(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::ReadFromOutputStream,
@@ -724,7 +766,8 @@ 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::HttpWrite(ref mut stream) => stream.get_mut().write(buf),
Stream::HttpRead(_) |
Stream::StaticString(_) |
Stream::Readline(_) |
Stream::InputFile(..) |
@@ -743,7 +786,8 @@ 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::HttpWrite(ref mut stream) => stream.stream.get_mut().flush(),
Stream::HttpRead(_) |
Stream::StaticString(_) |
Stream::Readline(_) |
Stream::InputFile(_) |
@@ -868,7 +912,8 @@ 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::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,
@@ -890,7 +935,8 @@ 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::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,
@@ -956,11 +1002,11 @@ impl Stream {
Stream::Byte(_)
| Stream::Readline(_)
| Stream::StaticString(_)
| Stream::NamedHttpClient(_)
| 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(_) => atom!("write"),
Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) | Stream::HttpWrite(_) => atom!("write"),
Stream::Null(_) => atom!(""),
}
}
@@ -1020,8 +1066,8 @@ impl Stream {
http_stream: Box<dyn BufRead>,
arena: &mut Arena,
) -> Self {
Stream::NamedHttpClient(arena_alloc!(
StreamLayout::new(CharReader::new(NamedHttpClientStream {
Stream::HttpRead(arena_alloc!(
StreamLayout::new(CharReader::new(HttpReadStream {
url,
body_reader: http_stream
})),
@@ -1029,6 +1075,19 @@ impl Stream {
))
}
#[inline]
pub(crate) fn from_http_sender(
body_writer: Sender,
arena: &mut Arena,
) -> Self {
Stream::HttpWrite(arena_alloc!(
StreamLayout::new(CharReader::new(HttpWriteStream {
body_writer
})),
arena
))
}
#[inline]
pub(crate) fn from_file_as_output(
file_name: Atom,
@@ -1065,7 +1124,7 @@ impl Stream {
Stream::NamedTls(ref mut tls_stream) => {
tls_stream.inner_mut().tls_stream.shutdown()
}
Stream::NamedHttpClient(ref mut http_stream) => {
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 _);
@@ -1073,6 +1132,14 @@ impl Stream {
Ok(())
}
Stream::HttpWrite(ref mut http_stream) => {
unsafe {
http_stream.set_tag(ArenaHeaderTag::Dropped);
std::ptr::drop_in_place(&mut http_stream.inner_mut().body_writer as *mut _);
}
Ok(())
}
Stream::InputFile(mut file_stream) => {
// close the stream by dropping the inner File.
unsafe {
@@ -1109,7 +1176,7 @@ impl Stream {
match self {
Stream::NamedTcp(..)
| Stream::NamedTls(..)
| Stream::NamedHttpClient(..)
| Stream::HttpRead(..)
| Stream::Byte(_)
| Stream::Readline(_)
| Stream::StaticString(_)
@@ -1124,7 +1191,8 @@ impl Stream {
Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::NamedTcp(..)
| Stream::NamedTls(..)
| Stream::NamedTls(..)
| Stream::HttpWrite(..)
| Stream::Byte(_)
| Stream::OutputFile(..) => true,
_ => false,

208
src/machine/system_calls.rs
View File

@@ -8,6 +8,7 @@ use crate::atom_table::*;
use crate::forms::*;
use crate::heap_iter::*;
use crate::heap_print::*;
use crate::http::{self, HttpListener, HttpResponse};
use crate::instructions::*;
use crate::machine;
use crate::machine::{Machine, VERIFY_ATTR_INTERRUPT_LOC, get_structure_index};
@@ -37,19 +38,20 @@ use ref_thread_local::{RefThreadLocal, ref_thread_local};
use std::cell::Cell;
use std::cmp::Ordering;
use std::collections::BTreeSet;
use std::convert::TryFrom;
use std::convert::{TryFrom, Infallible};
use std::env;
use std::fs;
use std::hash::{BuildHasher, BuildHasherDefault};
use std::io::{ErrorKind, Read, Write};
use std::iter::{once, FromIterator};
use std::mem;
use std::net::{TcpListener, TcpStream};
use std::net::{TcpListener, TcpStream, SocketAddr, ToSocketAddrs};
use std::num::NonZeroU32;
use std::ops::Sub;
use std::process;
use std::rc::Rc;
use std::str::FromStr;
use std::sync::Arc;
use chrono::{offset::Local, DateTime};
use cpu_time::ProcessTime;
@@ -79,10 +81,13 @@ use base64;
use roxmltree;
use select;
use hyper::{Body, Client, HeaderMap, Method, Request, Uri};
use hyper::{Body, Server, Client, HeaderMap, Method, Request, Response, Uri};
use hyper::header::{HeaderName, HeaderValue};
use hyper::body::Buf;
use hyper::service::{make_service_fn, service_fn};
use hyper_tls::HttpsConnector;
use tokio::sync::Mutex;
use tokio::sync::mpsc::channel;
ref_thread_local! {
pub(crate) static managed RANDOM_STATE: RandState<'static> = RandState::new();
@@ -3901,6 +3906,203 @@ impl Machine {
Ok(())
}
#[inline(always)]
pub(crate) fn http_listen(&mut self) -> CallResult {
let address_sink = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[1]));
if let Some(address_str) = self.machine_st.value_to_str_like(address_sink) {
let address_string = address_str.as_str();
let addr: SocketAddr = match address_string.to_socket_addrs().ok().and_then(|mut s| s.next()) {
Some(addr) => addr,
_ => {
self.machine_st.fail = true;
return Ok(());
}
};
let (tx, rx) = channel(1);
let tx = Arc::new(Mutex::new(tx));
let _guard = self.runtime.enter();
let server = match Server::try_bind(&addr) {
Ok(server) => server,
Err(_) => {
return Err(self.machine_st.open_permission_error(address_sink, atom!("http_listen"), 2));
}
};
self.runtime.spawn(async move {
let make_svc = make_service_fn(move |_conn| {
let tx = tx.clone();
async move { Ok::<_, Infallible>(service_fn(move |req| http::serve_req(req, tx.clone()))) }
});
let server = server.serve(make_svc);
if let Err(_) = server.await {
eprintln!("server error");
}
});
let http_listener = HttpListener { incoming: rx };
let http_listener = arena_alloc!(http_listener, &mut self.machine_st.arena);
let addr = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[2]));
self.machine_st.bind(addr.as_var().unwrap(), typed_arena_ptr_as_cell!(http_listener));
}
Ok(())
}
#[inline(always)]
pub(crate) fn http_accept(&mut self) -> CallResult {
let culprit = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[1]));
let method = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[2]));
let path = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[3]));
let query = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[5]));
let stream_addr = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[6]));
let handle_addr = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[7]));
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
match_untyped_arena_ptr!(cons_ptr,
(ArenaHeaderTag::HttpListener, http_listener) => {
match http_listener.incoming.blocking_recv() {
Some(request) => {
let method_atom = match *request.request.method() {
Method::GET => atom!("get"),
Method::POST => atom!("post"),
Method::PUT => atom!("put"),
Method::DELETE => atom!("delete"),
Method::PATCH => atom!("patch"),
Method::HEAD => atom!("head"),
_ => unreachable!(),
};
let path_atom = self.machine_st.atom_tbl.build_with(request.request.uri().path());
let path_cell = atom_as_cstr_cell!(path_atom);
let headers: Vec<HeapCellValue> = request.request.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());
let query_str = request.request.uri().query().unwrap_or("");
let query_atom = self.machine_st.atom_tbl.build_with(query_str);
let query_cell = string_as_cstr_cell!(query_atom);
let hyper_req = request.request;
let buf = self.runtime.block_on(async {hyper::body::aggregate(hyper_req).await.unwrap()});
let reader = buf.reader();
let mut stream = Stream::from_http_stream(
path_atom,
Box::new(reader),
&mut self.machine_st.arena
);
*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 handle = arena_alloc!(request.response, &mut self.machine_st.arena);
self.machine_st.bind(method.as_var().unwrap(), atom_as_cell!(method_atom));
self.machine_st.bind(path.as_var().unwrap(), path_cell);
unify!(self.machine_st, heap_loc_as_cell!(headers_list), self.machine_st.registers[4]);
self.machine_st.bind(query.as_var().unwrap(), query_cell);
self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
self.machine_st.bind(handle_addr.as_var().unwrap(), typed_arena_ptr_as_cell!(handle));
}
None => {
self.machine_st.fail = true;
}
}
}
_ => {
unreachable!();
}
);
}
_ => {
unreachable!();
}
);
Ok(())
}
#[inline(always)]
pub(crate) fn http_answer(&mut self) -> CallResult {
let culprit = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[1]));
let status_code = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[2]));
let status_code: u16 = match Number::try_from(status_code) {
Ok(Number::Fixnum(n)) => n.get_num() as u16,
Ok(Number::Integer(n)) => match n.to_u16() {
Some(u) => u,
_ => {
self.machine_st.fail = true;
return Ok(());
}
}
_ => unreachable!()
};
let stub_gen = || functor_stub(atom!("http_listen"), 2);
let headers = match self.machine_st.try_from_list(self.machine_st.registers[3], 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)
};
let stream_addr = self.machine_st.store(self.machine_st.deref(self.machine_st.registers[4]));
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
match_untyped_arena_ptr!(cons_ptr,
(ArenaHeaderTag::HttpResponse, http_response) => {
let mut response = Response::builder()
.status(status_code);
*response.headers_mut().unwrap() = headers;
let (sender, body) = Body::channel();
let response = response.body(body).unwrap();
http_response.blocking_send(response).unwrap();
let mut stream = Stream::from_http_sender(
sender,
&mut self.machine_st.arena
);
*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!();
}
);
Ok(())
}
#[inline(always)]
pub(crate) fn current_time(&mut self) {
let timestamp = self.systemtime_to_timestamp(SystemTime::now());