use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use crate::error::Result;
use crate::object::{ObjRef, Object, Stream};
pub const MAX_RESOLVE_DEPTH: usize = 32;
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
pub trait ObjectSource {
fn get(&self, r: ObjRef) -> Result<Object>;
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>>;
fn resolve(&self, o: &Object) -> Result<Object> {
resolve_sync_with(self, o)
}
}
impl<T: ObjectSource + ?Sized> ObjectSource for &T {
fn get(&self, r: ObjRef) -> Result<Object> {
(**self).get(r)
}
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>> {
(**self).stream_data(s)
}
fn resolve(&self, o: &Object) -> Result<Object> {
(**self).resolve(o)
}
}
pub trait AsyncObjectSource {
fn get(&self, r: ObjRef) -> BoxFuture<'_, Result<Object>>;
fn stream_data<'a>(&'a self, s: &'a Stream) -> BoxFuture<'a, Result<Vec<u8>>>;
fn resolve<'a>(&'a self, o: &'a Object) -> BoxFuture<'a, Result<Object>>;
}
impl<T: AsyncObjectSource + ?Sized> AsyncObjectSource for &T {
fn get(&self, r: ObjRef) -> BoxFuture<'_, Result<Object>> {
(**self).get(r)
}
fn stream_data<'a>(&'a self, s: &'a Stream) -> BoxFuture<'a, Result<Vec<u8>>> {
(**self).stream_data(s)
}
fn resolve<'a>(&'a self, o: &'a Object) -> BoxFuture<'a, Result<Object>> {
(**self).resolve(o)
}
}
pub fn resolve_sync_with<S>(src: &S, o: &Object) -> Result<Object>
where
S: ObjectSource + ?Sized,
{
let mut current = o.clone();
let mut last_num = 0;
for _ in 0..MAX_RESOLVE_DEPTH {
match current {
Object::Ref(r) => {
last_num = r.num;
current = match src.get(r) {
Ok(object) => object,
Err(crate::error::Error::CircularReference(n)) => {
return Err(crate::error::Error::CircularReference(n))
}
Err(_) => return Ok(Object::Null),
};
}
other => return Ok(other),
}
}
Err(crate::error::Error::CircularReference(last_num))
}
pub async fn resolve_with<S>(src: &S, o: &Object) -> Result<Object>
where
S: AsyncObjectSource + Sync + ?Sized,
{
let mut current = o.clone();
let mut last_num = 0;
for _ in 0..MAX_RESOLVE_DEPTH {
match current {
Object::Ref(r) => {
last_num = r.num;
current = match src.get(r).await {
Ok(object) => object,
Err(crate::error::Error::CircularReference(n)) => {
return Err(crate::error::Error::CircularReference(n))
}
Err(_) => return Ok(Object::Null),
};
}
other => return Ok(other),
}
}
Err(crate::error::Error::CircularReference(last_num))
}
#[derive(Debug, Clone, Copy)]
pub struct Immediate<S>(pub S);
impl<S: ObjectSource> AsyncObjectSource for Immediate<S> {
fn get(&self, r: ObjRef) -> BoxFuture<'_, Result<Object>> {
Box::pin(std::future::ready(self.0.get(r)))
}
fn stream_data<'a>(&'a self, s: &'a Stream) -> BoxFuture<'a, Result<Vec<u8>>> {
Box::pin(std::future::ready(self.0.stream_data(s)))
}
fn resolve<'a>(&'a self, o: &'a Object) -> BoxFuture<'a, Result<Object>> {
Box::pin(std::future::ready(self.0.resolve(o)))
}
}
struct ThreadWaker {
thread: std::thread::Thread,
notified: AtomicBool,
}
impl ThreadWaker {
fn notify(&self) {
if !self.notified.swap(true, Ordering::Release) {
self.thread.unpark();
}
}
}
impl Wake for ThreadWaker {
fn wake(self: Arc<Self>) {
self.notify();
}
fn wake_by_ref(self: &Arc<Self>) {
self.notify();
}
}
pub fn block_on<F: Future>(future: F) -> F::Output {
let waker_state = Arc::new(ThreadWaker {
thread: std::thread::current(),
notified: AtomicBool::new(false),
});
let waker = Waker::from(Arc::clone(&waker_state));
let mut cx = Context::from_waker(&waker);
let mut future = std::pin::pin!(future);
loop {
match future.as_mut().poll(&mut cx) {
Poll::Ready(value) => {
if waker_state.notified.swap(false, Ordering::Acquire) {
std::thread::park_timeout(std::time::Duration::ZERO);
}
return value;
}
Poll::Pending => {
while !waker_state.notified.swap(false, Ordering::Acquire) {
std::thread::park();
}
}
}
}
}
#[cfg(test)]
mod tests {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use crate::document::Document;
use crate::error::{Error, Result};
use crate::object::{ObjRef, Object, Stream};
use crate::source::{
block_on, resolve_sync_with, resolve_with, AsyncObjectSource, BoxFuture, Immediate,
ObjectSource, MAX_RESOLVE_DEPTH,
};
fn content_stream(doc: &Document) -> Stream {
match ObjectSource::get(doc, ObjRef { num: 4, gen: 0 }).unwrap() {
Object::Stream(s) => s,
other => panic!("expected object 4 to be the content stream, got {other:?}"),
}
}
#[test]
fn document_trait_get_matches_inherent_get() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let r = ObjRef { num: 1, gen: 0 };
let inherent = doc.get(r).unwrap();
let through_trait = ObjectSource::get(&doc, r).unwrap();
assert_eq!(inherent, through_trait);
}
#[test]
fn document_trait_resolve_is_lenient_about_missing_targets() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let missing = Object::Ref(ObjRef { num: 9_999, gen: 0 });
assert_eq!(
ObjectSource::resolve(&doc, &missing).unwrap(),
Object::Null,
"a dangling reference must resolve to Null through the trait, \
matching Document::resolve"
);
}
#[test]
fn immediate_get_matches_the_sync_source() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let r = ObjRef { num: 1, gen: 0 };
let expected = ObjectSource::get(&doc, r).unwrap();
let src = Immediate(&doc);
let actual = block_on(src.get(r)).unwrap();
assert_eq!(actual, expected);
}
#[test]
fn immediate_resolve_is_lenient_about_missing_targets() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let missing = Object::Ref(ObjRef { num: 9_999, gen: 0 });
let src = Immediate(&doc);
assert_eq!(block_on(src.resolve(&missing)).unwrap(), Object::Null);
}
#[test]
fn immediate_stream_data_matches_the_sync_source() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let stream = content_stream(&doc);
let expected = ObjectSource::stream_data(&doc, &stream).unwrap();
let src = Immediate(&doc);
let actual = block_on(src.stream_data(&stream)).unwrap();
assert_eq!(actual, expected);
assert!(
!actual.is_empty(),
"the fixture's content stream must decode to something"
);
}
#[test]
fn block_on_drives_a_nested_future_tree() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let src = Immediate(&doc);
async fn depth_three<S: AsyncObjectSource>(src: &S, r: ObjRef) -> Object {
async fn inner<S: AsyncObjectSource>(src: &S, r: ObjRef) -> Object {
src.resolve(&Object::Ref(r)).await.unwrap()
}
inner(src, r).await
}
let got = block_on(depth_three(&src, ObjRef { num: 1, gen: 0 }));
assert!(
!matches!(got, Object::Null),
"object 1 of a simple document must resolve to something"
);
}
struct SelfReferential;
impl AsyncObjectSource for SelfReferential {
fn get(&self, r: ObjRef) -> BoxFuture<'_, Result<Object>> {
Box::pin(std::future::ready(Ok(Object::Ref(r))))
}
fn stream_data<'a>(&'a self, s: &'a Stream) -> BoxFuture<'a, Result<Vec<u8>>> {
Box::pin(std::future::ready(Ok(s.data.clone())))
}
fn resolve<'a>(&'a self, o: &'a Object) -> BoxFuture<'a, Result<Object>> {
Box::pin(resolve_with(self, o))
}
}
#[test]
fn resolve_with_stops_at_the_depth_cap() {
let chain = Object::Ref(ObjRef { num: 7, gen: 0 });
let err = block_on(resolve_with(&SelfReferential, &chain)).unwrap_err();
assert!(
matches!(err, Error::CircularReference(7)),
"a self-referential chain must exhaust the cap and report \
CircularReference for the last reference seen, got {err:?}"
);
}
#[test]
fn resolve_with_returns_a_direct_object_unchanged() {
let direct = Object::Int(42);
assert_eq!(
block_on(resolve_with(&SelfReferential, &direct)).unwrap(),
Object::Int(42)
);
}
struct OwnedStub {
payload: Vec<u8>,
}
impl AsyncObjectSource for OwnedStub {
fn get(&self, _r: ObjRef) -> BoxFuture<'_, Result<Object>> {
Box::pin(std::future::ready(Ok(Object::Int(
self.payload.len() as i64
))))
}
fn stream_data<'a>(&'a self, s: &'a Stream) -> BoxFuture<'a, Result<Vec<u8>>> {
Box::pin(std::future::ready(Ok(s.data.clone())))
}
fn resolve<'a>(&'a self, o: &'a Object) -> BoxFuture<'a, Result<Object>> {
Box::pin(resolve_with(self, o))
}
}
async fn fetch_one<S: AsyncObjectSource>(src: S, r: ObjRef) -> Result<Object> {
src.get(r).await
}
#[test]
fn an_owned_source_yields_a_spawnable_future() {
fn assert_send_static<F: Future + Send + 'static>(_: &F) {}
let future = fetch_one(
OwnedStub {
payload: vec![1, 2, 3],
},
ObjRef { num: 3, gen: 0 },
);
assert_send_static(&future);
assert!(block_on(future).is_ok());
}
#[test]
fn the_same_function_serves_a_borrowed_synchronous_source() {
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let r = ObjRef { num: 1, gen: 0 };
let expected = ObjectSource::get(&doc, r).unwrap();
let got = block_on(fetch_one(Immediate(&doc), r)).unwrap();
assert_eq!(got, expected);
}
#[test]
fn a_reference_to_a_source_is_a_source() {
let stub = OwnedStub {
payload: vec![7, 7],
};
let r = ObjRef { num: 1, gen: 0 };
let borrowed = block_on(fetch_one(&stub, r)).unwrap();
let owned = block_on(fetch_one(stub, r)).unwrap();
assert_eq!(borrowed, Object::Int(2));
assert_eq!(borrowed, owned);
}
struct SyncSelfReferential;
impl ObjectSource for SyncSelfReferential {
fn get(&self, r: ObjRef) -> Result<Object> {
Ok(Object::Ref(r))
}
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>> {
Ok(s.data.clone())
}
}
struct CircularSource;
impl ObjectSource for CircularSource {
fn get(&self, r: ObjRef) -> Result<Object> {
Err(Error::CircularReference(r.num))
}
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>> {
Ok(s.data.clone())
}
}
struct UnreadableSource;
impl ObjectSource for UnreadableSource {
fn get(&self, _: ObjRef) -> Result<Object> {
Err(Error::MissingKey("Length"))
}
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>> {
Ok(s.data.clone())
}
}
struct Chain;
impl ObjectSource for Chain {
fn get(&self, r: ObjRef) -> Result<Object> {
if r.num == 0 {
Ok(Object::Int(0))
} else {
Ok(Object::Ref(ObjRef {
num: r.num - 1,
gen: 0,
}))
}
}
fn stream_data(&self, s: &Stream) -> Result<Vec<u8>> {
Ok(s.data.clone())
}
}
#[test]
fn resolve_sync_with_stops_at_the_depth_cap() {
let chain = Object::Ref(ObjRef { num: 7, gen: 0 });
let err = resolve_sync_with(&SyncSelfReferential, &chain).unwrap_err();
assert!(
matches!(err, Error::CircularReference(7)),
"a self-referential chain must exhaust the cap and report \
CircularReference for the last reference seen, got {err:?}"
);
}
#[test]
fn resolve_sync_with_returns_a_direct_object_unchanged() {
assert_eq!(
resolve_sync_with(&SyncSelfReferential, &Object::Int(42)).unwrap(),
Object::Int(42)
);
}
#[test]
fn resolve_sync_with_is_lenient_but_propagates_cycles() {
let r = Object::Ref(ObjRef { num: 5, gen: 0 });
assert_eq!(
resolve_sync_with(&UnreadableSource, &r).unwrap(),
Object::Null,
"a fetch failing for any reason other than a cycle must flatten \
to Null"
);
assert!(matches!(
resolve_sync_with(&CircularSource, &r).unwrap_err(),
Error::CircularReference(5)
));
}
#[test]
fn the_provided_resolve_is_the_shared_chase() {
let direct = Object::Int(9);
assert_eq!(
ObjectSource::resolve(&SyncSelfReferential, &direct).unwrap(),
resolve_sync_with(&SyncSelfReferential, &direct).unwrap()
);
let chain = Object::Ref(ObjRef { num: 3, gen: 0 });
assert!(matches!(
ObjectSource::resolve(&SyncSelfReferential, &chain).unwrap_err(),
Error::CircularReference(3)
));
}
#[test]
fn both_chases_share_one_depth_cap() {
let longest = u32::try_from(MAX_RESOLVE_DEPTH - 2).expect("the cap fits in a u32");
let fits = Object::Ref(ObjRef {
num: longest,
gen: 0,
});
let too_long = Object::Ref(ObjRef {
num: longest + 1,
gen: 0,
});
assert_eq!(resolve_sync_with(&Chain, &fits).unwrap(), Object::Int(0));
assert_eq!(
block_on(resolve_with(&Immediate(&Chain), &fits)).unwrap(),
Object::Int(0)
);
assert!(matches!(
resolve_sync_with(&Chain, &too_long).unwrap_err(),
Error::CircularReference(0)
));
assert!(matches!(
block_on(resolve_with(&Immediate(&Chain), &too_long)).unwrap_err(),
Error::CircularReference(0)
));
}
#[test]
fn block_on_resumes_after_parking() {
struct YieldOnce {
yielded: bool,
}
impl Future for YieldOnce {
type Output = u32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
if self.yielded {
Poll::Ready(7)
} else {
self.yielded = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
assert_eq!(
block_on(YieldOnce { yielded: false }),
7,
"block_on must re-poll after parking rather than panicking or hanging"
);
}
#[test]
fn nested_block_on_does_not_strand_the_outer_call() {
struct SelfWaking {
yielded: bool,
}
impl Future for SelfWaking {
type Output = u32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
if self.yielded {
return Poll::Ready(7);
}
self.yielded = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}
struct Outer {
polled: bool,
inner: u32,
}
impl Future for Outer {
type Output = u32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
if self.polled {
return Poll::Ready(self.inner);
}
self.polled = true;
cx.waker().wake_by_ref();
self.inner = block_on(SelfWaking { yielded: false });
Poll::Pending
}
}
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = tx.send(block_on(Outer {
polled: false,
inner: 0,
}));
});
match rx.recv_timeout(std::time::Duration::from_secs(30)) {
Ok(inner) => assert_eq!(
inner, 7,
"the nested block_on must have run its future to completion"
),
Err(e) => panic!(
"the outer block_on never finished ({e:?}): the nested call \
consumed its wakeup"
),
}
}
#[test]
fn immediate_futures_are_send() {
fn assert_send<T: Send>(_: &T) {}
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
let stream = content_stream(&doc);
let src = Immediate(&doc);
let object = Object::Ref(ObjRef { num: 1, gen: 0 });
assert_send(&src.get(ObjRef { num: 1, gen: 0 }));
assert_send(&src.stream_data(&stream));
assert_send(&src.resolve(&object));
}
#[test]
fn async_object_source_is_object_safe() {
fn assert_dyn(_: &dyn AsyncObjectSource) {}
let doc = Document::load(pdfboss_testkit::simple_doc("Hello")).unwrap();
assert_dyn(&Immediate(&doc));
assert_dyn(&SelfReferential);
}
}