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

Replace Hyper with Warp for HTTP server

Browse Source 
- Use Warp
- Optimize clones
- HTTPS server
- Content-Length limit
- HTTP Basic Auth
- Stop server with Ctrl-C
This commit is contained in:
Adrián Arroyo Calle
2023-09-05 20:40:58 +02:00
parent 841aaa78b4
commit 660860bccf
9 changed files with 682 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/http.rs
View File

@@ -1,55 +1,23 @@
use bytes::Bytes;
use http_body_util::Full;
use hyper::service::Service;
use hyper::{body::Incoming as IncomingBody, Request, Response};
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Condvar, Mutex};
use std::sync::{Arc, Mutex, Condvar};
use std::io::BufRead;
use warp::http;
pub struct HttpListener {
pub incoming: std::sync::mpsc::Receiver<HttpRequest>,
}
#[derive(Debug)]
pub struct HttpRequest {
pub request: Request<IncomingBody>,
pub request_data: HttpRequestData,
pub response: HttpResponse,
}
pub type HttpResponse = Arc<(Mutex<bool>, Mutex<Option<Response<Full<Bytes>>>>, Condvar)>;
pub type HttpResponse = Arc<(Mutex<bool>, Mutex<Option<warp::reply::Response>>, Condvar)>;
pub struct HttpService {
pub tx: std::sync::mpsc::SyncSender<HttpRequest>,
}
impl Service<Request<IncomingBody>> for HttpService {
type Response = Response<Full<Bytes>>;
type Error = hyper::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn call(&mut self, req: Request<IncomingBody>) -> Self::Future {
// new connection!
// we send the Request info to Prolog
let response = Arc::new((Mutex::new(false), Mutex::new(None), Condvar::new()));
let http_request = HttpRequest {
request: req,
response: Arc::clone(&response),
};
self.tx.send(http_request).unwrap();
// we wait for the Response info from Prolog
{
let (ready, _response, cvar) = &*response;
let mut ready = ready.lock().unwrap();
while !*ready {
ready = cvar.wait(ready).unwrap();
}
}
{
let (_, response, _) = &*response;
let response = response.lock().unwrap().take();
let res = response.expect("Data race error in HTTP Server");
Box::pin(async move { Ok(res) })
}
}
pub struct HttpRequestData {
pub method: http::Method,
pub headers: http::HeaderMap,
pub path: String,
pub query: String,
pub body: Box<dyn BufRead + Send>,
}

105
src/lib/http/http_server.pl
View File

@@ -6,8 +6,8 @@
*/
/** This library provides an starting point to build HTTP server based applications.
It is based on [Hyper](https://hyper.rs/), which allows for HTTP/1.0, HTTP/1.1 and HTTP/2. However,
some advanced features that Hyper provides are still not accesible.
It is based on [Warp](https://github.com/seanmonstar/warp), which allows for HTTP/1.0, HTTP/1.1 and HTTP/2. However,
some advanced features that Warp provides are still not accesible.
## Usage
@@ -46,7 +46,6 @@ recommeded to use the helper predicates, which are easier to understand and clea
Some things that are still missing:
- Read forms in multipart format
- HTTP Basic Auth
- Session handling via cookies
- HTML Templating (but you can use [Teruel](https://github.com/aarroyoc/teruel/), [Marquete](https://github.com/aarroyoc/marquete/) or [Djota](https://github.com/aarroyoc/djota) for that)
*/
@@ -54,14 +53,19 @@ Some things that are still missing:
:- module(http_server, [
http_listen/2,
http_listen/3,
http_headers/2,
http_status_code/2,
http_body/2,
http_redirect/2,
http_query/3
http_query/3,
http_basic_auth/4
]).
:- meta_predicate http_listen(?, :).
:- meta_predicate http_listen(?, :, ?).
:- meta_predicate http_basic_auth(:, :, ?, ?).
:- use_module(library(charsio)).
:- use_module(library(crypto)).
@@ -74,25 +78,58 @@ Some things that are still missing:
%% http_listen(+Port, +Handlers).
%
% Listens for HTTP connections on port Port. Each handler on the list Handlers should be of the form: `HttpVerb(PathUnification, Predicate)`.
% For example: `get(user/User, get_info(User))` will match an HTTP request that is a GET, the path unifies with /user/User (where User is a variable)
% and it will call `get_info` with three arguments: an `http_request` term, an `http_response` term and User.
% Equivalent to `http_listen(Port, Handlers, [])`.
http_listen(Port, Module:Handlers0) :-
must_be(integer, Port),
must_be(list, Handlers0),
maplist(module_qualification(Module), Handlers0, Handlers),
http_listen_(Port, Handlers).
http_listen_(Port, Handlers, []).
%% http_listen(+Port, +Handlers, +Options).
%
% Listens for HTTP connections on port Port. Each handler on the list Handlers should be of the form: `HttpVerb(PathUnification, Predicate)`.
% For example: `get(user/User, get_info(User))` will match an HTTP request that is a GET, the path unifies with /user/User (where User is a variable)
% and it will call `get_info` with three arguments: an `http_request` term, an `http_response` term and User.
%
% The following options are supported:
%
% - `tls_key(+Key)` - a TLS key for HTTPS (string)
% - `tls_cert(+Cert)` - a TLS cert for HTTPS (string)
% - `content_length_limit(+Limit)` - maximum length (in bytes) for the incoming bodies. By default, 32KB.
%
% In order to have a HTTPS server (instead of plain HTTP), both `tls_key` and `tls_cert` options must be provided.
http_listen(Port, Module:Handlers0, Options) :-
must_be(integer, Port),
must_be(list, Handlers0),
must_be(list, Options),
maplist(module_qualification(Module), Handlers0, Handlers),
http_listen_(Port, Handlers, Options).
module_qualification(M, H0, H) :-
H0 =.. [Method, Path, Goal],
H =.. [Method, Path, M:Goal].
http_listen_(Port, Handlers) :-
http_listen_(Port, Handlers, Options) :-
parse_options(Options, TLSKey, TLSCert, ContentLengthLimit),
phrase(format_("0.0.0.0:~d", [Port]), Addr),
'$http_listen'(Addr, HttpListener),!,
'$http_listen'(Addr, HttpListener, TLSKey, TLSCert, ContentLengthLimit),!,
format("Listening at ~s\n", [Addr]),
http_loop(HttpListener, Handlers).
parse_options(Options, TLSKey, TLSCert, ContentLengthLimit) :-
member_option_default(tls_key, Options, "", TLSKey),
member_option_default(tls_cert, Options, "", TLSCert),
member_option_default(content_length_limit, Options, 32768, ContentLengthLimit),
must_be(integer, ContentLengthLimit).
member_option_default(Key, List, _Default, Value) :-
X =.. [Key, Value],
member(X, List).
member_option_default(Key, List, Default, Default) :-
X =.. [Key, _],
\+ member(X, List).
http_loop(HttpListener, Handlers) :-
'$http_accept'(HttpListener, RequestMethod, RequestPath, RequestHeaders, RequestQuery, RequestStream, ResponseHandle),
current_time(Time),
@@ -114,7 +151,7 @@ http_loop(HttpListener, Handlers) :-
)
; (
'$http_answer'(ResponseHandle, 404, [], ResponseStream),
call_cleanup(format(ResponseStream, "Not Found"), close(ResponseStream)))
call_cleanup(format(ResponseStream, "Not Found", []), close(ResponseStream)))
),
http_loop(HttpListener, Handlers).
@@ -352,3 +389,49 @@ url_decode([Char|Chars]) -->
url_decode(Chars).
url_decode([]) --> [].
%% http_basic_auth(+LoginPredicate, +Handler, +Request, -Response)
%
% Metapredicate that wraps an existing Handler with an HTTP Basic Auth flow.
% Checks if a given user + password is authorized to execute that handler, returning 401
% if it's not satisfied.
%
% `LoginPredicate` must be a predicate of arity 2 that takes a User and a Password.
% `Handler` will have, in addition to the Request and Response arguments, a User argument
% containing the User given in the authentication.
%
% Example:
%
% ```
% main :-
% http_listen(8800,[get('/', http_basic_auth(login, inside_handler("data")))]).
%
% login(User, Pass) :-
% User = "aarroyoc",
% Pass = "123456".
%
% inside_handler(Data, User, Request, Response) :-
% http_body(Response, text(User)).
% ```
http_basic_auth(LoginPredicate, Handler, Request, Response) :-
http_headers(Request, Headers),
member("authorization"-AuthorizationStr, Headers),
append("Basic ", Coded, AuthorizationStr),
chars_base64(UserPass, Coded, []),
append(User, [':'|Password], UserPass),
(
call(LoginPredicate, User, Password) ->
call(Handler, User, Request, Response)
; http_basic_auth_unauthorized_response(Response)
).
http_basic_auth(_LoginPredicate, _Handler, Request, Response) :-
http_headers(Request, Headers),
\+ member("authorization"-_, Headers),
http_basic_auth_unauthorized_response(Response).
http_basic_auth_unauthorized_response(Response) :-
http_status_code(Response, 401),
http_headers(Response, ["www-authenticate"-"Basic realm=\"Scryer Prolog\", charset=\"UTF-8\""]),
http_body(Response, text("Unauthorized")).

11
src/machine/dispatch.rs
View File

@@ -5499,6 +5499,17 @@ impl Machine {
if interruption {
self.machine_st.throw_interrupt_exception();
self.machine_st.backtrack();
#[cfg(not(target_arch = "wasm32"))]
let runtime = tokio::runtime::Runtime::new().unwrap();
#[cfg(target_arch = "wasm32")]
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let old_runtime = std::mem::replace(&mut self.runtime, runtime);
old_runtime.shutdown_background();
}
}
Err(_) => unreachable!(),

67
src/machine/streams.rs
View File

@@ -30,6 +30,9 @@ use std::ptr;
#[cfg(feature = "tls")]
use native_tls::TlsStream;
#[cfg(feature = "http")]
use warp::hyper;
#[derive(Debug, BitfieldSpecifier, Clone, Copy, PartialEq, Eq, Hash)]
#[bits = 1]
pub enum StreamType {
@@ -290,9 +293,9 @@ impl Read for HttpReadStream {
#[cfg(feature = "http")]
pub struct HttpWriteStream {
status_code: u16,
headers: hyper::HeaderMap,
headers: mem::ManuallyDrop<hyper::HeaderMap>,
response: TypedArenaPtr<HttpResponse>,
buffer: Vec<u8>,
buffer: mem::ManuallyDrop<Vec<u8>>,
}
#[cfg(feature = "http")]
@@ -312,24 +315,33 @@ impl Write for HttpWriteStream {
#[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(())
Ok(())
}
}
#[cfg(feature = "http")]
impl HttpWriteStream {
fn drop(&mut self) {
let headers = unsafe { mem::ManuallyDrop::take(&mut self.headers) };
let buffer = unsafe { mem::ManuallyDrop::take(&mut self.buffer) };
let (ready, response, cvar) = &**self.response;
let mut ready = ready.lock().unwrap();
{
let mut response = response.lock().unwrap();
let mut response_ = warp::http::Response::builder()
.status(self.status_code);
*response_.headers_mut().unwrap() = headers;
*response = Some(response_.body(warp::hyper::Body::from(buffer)).unwrap());
}
*ready = true;
cvar.notify_one();
}
}
#[derive(Debug)]
pub struct StandardOutputStream {}
@@ -1243,15 +1255,15 @@ impl Stream {
headers: hyper::HeaderMap,
arena: &mut Arena,
) -> Self {
Stream::HttpWrite(arena_alloc!(
StreamLayout::new(CharReader::new(HttpWriteStream {
response,
status_code,
headers,
buffer: Vec::new(),
})),
arena
))
Stream::HttpWrite(arena_alloc!(
StreamLayout::new(CharReader::new(HttpWriteStream {
response,
status_code,
headers: mem::ManuallyDrop::new(headers),
buffer: mem::ManuallyDrop::new(Vec::new()),
})),
arena
))
}
#[inline]
@@ -1299,7 +1311,8 @@ impl Stream {
Ok(())
}
#[cfg(feature = "http")]
Stream::HttpWrite(ref mut http_stream) => {
Stream::HttpWrite(ref mut http_stream) => {
http_stream.inner_mut().drop();
unsafe {
http_stream.set_tag(ArenaHeaderTag::Dropped);
std::ptr::drop_in_place(&mut http_stream.inner_mut().buffer as *mut _);

297
src/machine/system_calls.rs
View File

@@ -14,7 +14,7 @@ use crate::forms::*;
use crate::heap_iter::*;
use crate::heap_print::*;
#[cfg(feature = "http")]
use crate::http::{HttpListener, HttpResponse, HttpService};
use crate::http::{HttpRequestData, HttpListener, HttpResponse, HttpRequest};
use crate::instructions::*;
use crate::machine;
use crate::machine::code_walker::*;
@@ -44,14 +44,14 @@ pub(crate) use ref_thread_local::RefThreadLocal;
use std::cell::Cell;
use std::cmp::Ordering;
use std::collections::BTreeSet;
use std::collections::{BTreeSet};
use std::convert::TryFrom;
use std::env;
#[cfg(feature = "ffi")]
use std::ffi::CString;
use std::fs;
use std::hash::{BuildHasher, BuildHasherDefault};
use std::io::{ErrorKind, Read, Write};
use std::io::{ErrorKind, Read, BufRead, Write};
use std::iter::{once, FromIterator};
use std::mem;
use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
@@ -59,6 +59,7 @@ use std::num::NonZeroU32;
use std::ops::Sub;
use std::process;
use std::str::FromStr;
use std::sync::{Mutex, Arc, Condvar};
use chrono::{offset::Local, DateTime};
#[cfg(not(target_arch = "wasm32"))]
@@ -91,16 +92,15 @@ use base64;
use roxmltree;
use select;
use bytes::Buf;
use http_body_util::BodyExt;
#[cfg(feature = "http")]
use hyper::header::{HeaderName, HeaderValue};
use warp::hyper::header::{HeaderValue, HeaderName};
#[cfg(feature = "http")]
use hyper::server::conn::http1;
use warp::hyper::{HeaderMap, Method};
#[cfg(feature = "http")]
use hyper::{HeaderMap, Method};
use warp::{Buf, Filter};
#[cfg(feature = "http")]
use reqwest::Url;
use futures::future;
#[cfg(feature = "repl")]
pub(crate) fn get_key() -> KeyEvent {
@@ -4412,54 +4412,119 @@ impl Machine {
#[inline(always)]
pub(crate) fn http_listen(&mut self) -> CallResult {
let address_sink = self.deref_register(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,
let tls_key = self.deref_register(3);
let tls_cert = self.deref_register(4);
let content_length_limit = self.deref_register(5);
const CONTENT_LENGTH_LIMIT_DEFAULT: u64 = 32768;
let content_length_limit = match Number::try_from(content_length_limit) {
Ok(Number::Fixnum(n)) => if n.get_num() >= 0 {
n.get_num() as u64
} else {
CONTENT_LENGTH_LIMIT_DEFAULT
},
Ok(Number::Integer(n)) => {
let n: Result<u64, _> = (&*n).try_into();
match n {
Ok(u) => u,
Err(_) => CONTENT_LENGTH_LIMIT_DEFAULT,
}
}
_ => CONTENT_LENGTH_LIMIT_DEFAULT,
};
let ssl_server: Option<(String,String)> = {
match self.machine_st.value_to_str_like(tls_key) {
Some(key) => {
match self.machine_st.value_to_str_like(tls_cert) {
Some(cert) => {
let key_str = key.as_str();
let cert_str = cert.as_str();
if key_str.is_empty() || cert_str.is_empty() {
None
} else {
Some((key_str.to_string(), cert_str.to_string()))
}
}
None => None
}
}
None => None
}
};
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) = std::sync::mpsc::sync_channel(1024);
let (tx, rx) = std::sync::mpsc::sync_channel(1024);
let _guard = self.runtime.enter();
let listener = match self
.runtime
.block_on(async { tokio::net::TcpListener::bind(addr).await })
{
Ok(listener) => listener,
Err(_) => {
return Err(self.machine_st.open_permission_error(
address_sink,
atom!("http_listen"),
2,
));
}
};
fn get_reader(body: impl Buf + Send + 'static) -> Box<dyn BufRead + Send> {
Box::new(body.reader())
}
self.runtime.spawn(async move {
loop {
let tx = tx.clone();
let (stream, _) = listener.accept().await.unwrap();
let serve = warp::body::aggregate()
.and(warp::header::optional::<u64>(warp::http::header::CONTENT_LENGTH.as_str()))
.and(warp::method())
.and(warp::header::headers_cloned())
.and(warp::path::full())
.and(warp::query::raw().or_else(|_| future::ready(Ok::<(String,), warp::Rejection>(("".to_string(),)))))
.map(move |body, content_length, method, headers: warp::http::HeaderMap, path: warp::filters::path::FullPath, query| {
if let Some(content_length) = content_length {
if content_length > content_length_limit {
return warp::http::Response::builder()
.status(413)
.body(warp::hyper::Body::empty())
.unwrap();
}
}
let http_request_data = HttpRequestData {
method,
headers,
path: path.as_str().to_string(),
query,
body: get_reader(body),
};
let response = Arc::new((Mutex::new(false), Mutex::new(None), Condvar::new()));
let http_request = HttpRequest { request_data: http_request_data, response: Arc::clone(&response) };
// we send the request to http_accept
tx.send(http_request).unwrap();
tokio::task::spawn(async move {
if let Err(err) = http1::Builder::new()
.serve_connection(stream, HttpService { tx })
.await
{
eprintln!("Error serving connection: {:?}", err);
}
});
}
});
let http_listener = HttpListener { incoming: rx };
let http_listener = arena_alloc!(http_listener, &mut self.machine_st.arena);
// we wait for the Response info from Prolog
{
let (ready, _response, cvar) = &*response;
let mut ready = ready.lock().unwrap();
while !*ready {
ready = cvar.wait(ready).unwrap();
}
}
{
let (_, response, _) = &*response;
let response = response.lock().unwrap().take();
response.expect("Data race error in HTTP server")
}
});
self.runtime.spawn(async move {
match ssl_server {
Some((key, cert)) => {
warp::serve(serve).tls().key(key).cert(cert).run(addr).await
}
None => {
warp::serve(serve).run(addr).await
}
}
});
let http_listener = HttpListener { incoming: rx };
let http_listener = arena_alloc!(http_listener, &mut self.machine_st.arena);
let addr = self.deref_register(2);
self.machine_st.bind(
addr.as_var().unwrap(),
@@ -4472,74 +4537,94 @@ impl Machine {
#[cfg(feature = "http")]
#[inline(always)]
pub(crate) fn http_accept(&mut self) -> CallResult {
let culprit = self.deref_register(1);
let method = self.deref_register(2);
let path = self.deref_register(3);
let query = self.deref_register(5);
let stream_addr = self.deref_register(6);
let handle_addr = self.deref_register(7);
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
match_untyped_arena_ptr!(cons_ptr,
(ArenaHeaderTag::HttpListener, http_listener) => {
match http_listener.incoming.recv() {
Ok(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 = AtomTable::build_with(&self.machine_st.atom_tbl, 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 culprit = self.deref_register(1);
let method = self.deref_register(2);
let path = self.deref_register(3);
let query = self.deref_register(5);
let stream_addr = self.deref_register(6);
let handle_addr = self.deref_register(7);
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
match_untyped_arena_ptr!(cons_ptr,
(ArenaHeaderTag::HttpListener, http_listener) => {
loop {
match http_listener.incoming.recv_timeout(std::time::Duration::from_millis(200)) {
Ok(request) => {
let method_atom = match request.request_data.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"),
Method::OPTIONS => atom!("options"),
Method::TRACE => atom!("trace"),
Method::CONNECT => atom!("connect"),
_ => atom!("unsupported_extension"),
};
let path_atom = AtomTable::build_with(&self.machine_st.atom_tbl, &request.request_data.path);
let path_cell = atom_as_cstr_cell!(path_atom);
let headers: Vec<HeapCellValue> = request.request_data.headers.iter().map(|(header_name, header_value)| {
let h = self.machine_st.heap.len();
let header_term = functor!(AtomTable::build_with(&self.machine_st.atom_tbl, header_name.as_str()), [cell(string_as_cstr_cell!(AtomTable::build_with(&self.machine_st.atom_tbl, header_value.to_str().unwrap())))]);
let header_term = functor!(
AtomTable::build_with(&self.machine_st.atom_tbl, header_name.as_str()),
[cell(string_as_cstr_cell!(AtomTable::build_with(&self.machine_st.atom_tbl, header_value.to_str().unwrap())))]
);
self.machine_st.heap.extend(header_term.into_iter());
str_loc_as_cell!(h)
}).collect();
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 headers_list = iter_to_heap_list(&mut self.machine_st.heap, headers.into_iter());
let query_str = request.request_data.query;
let query_atom = AtomTable::build_with(&self.machine_st.atom_tbl, &query_str);
let query_cell = string_as_cstr_cell!(query_atom);
let query_str = request.request.uri().query().unwrap_or("");
let query_atom = AtomTable::build_with(&self.machine_st.atom_tbl, query_str);
let query_cell = string_as_cstr_cell!(query_atom);
let mut stream = Stream::from_http_stream(
path_atom,
request.request_data.body,
&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 hyper_req = request.request;
let buf = self.runtime.block_on(async {hyper_req.collect().await.unwrap().aggregate()});
let reader = buf.reader();
let handle = arena_alloc!(request.response, &mut self.machine_st.arena);
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);
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));
break
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
let interrupted = machine::INTERRUPT.load(std::sync::atomic::Ordering::Relaxed);
let handle = arena_alloc!(request.response, &mut self.machine_st.arena);
match machine::INTERRUPT.compare_exchange(
interrupted,
false,
std::sync::atomic::Ordering::Relaxed,
std::sync::atomic::Ordering::Relaxed,
) {
Ok(interruption) => {
if interruption {
self.machine_st.throw_interrupt_exception();
self.machine_st.backtrack();
let old_runtime = std::mem::replace(&mut self.runtime, tokio::runtime::Runtime::new().unwrap());
old_runtime.shutdown_background();
break
}
}
Err(_) => unreachable!(),
}
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));
}
Err(_) => {
self.machine_st.fail = true;
}
}
}
Err(_) => {
self.machine_st.fail = true;
}
}
}
_ => {
unreachable!();