1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-12-23 15:49:44 -05:00
denoland-deno/core/ops.rs

112 lines
2.7 KiB
Rust
Raw Normal View History

2019-09-30 14:59:44 -04:00
// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
pub use crate::libdeno::OpId;
use crate::PinnedBuf;
use futures::Future;
use std::collections::HashMap;
pub type Buf = Box<[u8]>;
pub type OpAsyncFuture<E> = Box<dyn Future<Item = Buf, Error = E> + Send>;
pub(crate) type PendingOpFuture =
Box<dyn Future<Item = (OpId, Buf), Error = CoreError> + Send>;
pub type OpResult<E> = Result<Op<E>, E>;
pub enum Op<E> {
Sync(Buf),
Async(OpAsyncFuture<E>),
}
pub type CoreError = ();
pub type CoreOp = Op<CoreError>;
/// Main type describing op
type OpDispatcher = dyn Fn(&[u8], Option<PinnedBuf>) -> CoreOp;
#[derive(Default)]
pub struct OpRegistry {
dispatchers: Vec<Box<OpDispatcher>>,
name_to_id: HashMap<String, OpId>,
}
impl OpRegistry {
pub fn new() -> Self {
let mut registry = Self::default();
let op_id = registry.register("ops", |_, _| {
// ops is a special op which is handled in call.
unreachable!()
});
assert_eq!(op_id, 0);
registry
}
pub fn register<F>(&mut self, name: &str, op: F) -> OpId
where
F: Fn(&[u8], Option<PinnedBuf>) -> CoreOp + Send + Sync + 'static,
{
let op_id = self.dispatchers.len() as u32;
let existing = self.name_to_id.insert(name.to_string(), op_id);
assert!(
existing.is_none(),
format!("Op already registered: {}", name)
);
self.dispatchers.push(Box::new(op));
op_id
}
fn json_map(&self) -> Buf {
let op_map_json = serde_json::to_string(&self.name_to_id).unwrap();
op_map_json.as_bytes().to_owned().into_boxed_slice()
}
pub fn call(
&self,
op_id: OpId,
control: &[u8],
zero_copy_buf: Option<PinnedBuf>,
) -> CoreOp {
// Op with id 0 has special meaning - it's a special op that is always
// provided to retrieve op id map. The map consists of name to `OpId`
// mappings.
if op_id == 0 {
return Op::Sync(self.json_map());
}
let d = &*self.dispatchers.get(op_id as usize).expect("Op not found!");
d(control, zero_copy_buf)
}
}
#[test]
fn test_op_registry() {
use std::sync::atomic;
use std::sync::Arc;
let mut op_registry = OpRegistry::new();
let c = Arc::new(atomic::AtomicUsize::new(0));
let c_ = c.clone();
let test_id = op_registry.register("test", move |_, _| {
c_.fetch_add(1, atomic::Ordering::SeqCst);
CoreOp::Sync(Box::new([]))
});
assert!(test_id != 0);
let mut expected = HashMap::new();
expected.insert("ops".to_string(), 0);
expected.insert("test".to_string(), 1);
assert_eq!(op_registry.name_to_id, expected);
let res = op_registry.call(test_id, &[], None);
if let Op::Sync(buf) = res {
assert_eq!(buf.len(), 0);
} else {
unreachable!();
}
assert_eq!(c.load(atomic::Ordering::SeqCst), 1);
}