// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license. mod in_memory_broadcast_channel; pub use in_memory_broadcast_channel::InMemoryBroadcastChannel; pub use in_memory_broadcast_channel::InMemoryBroadcastChannelResource; use std::cell::RefCell; use std::path::PathBuf; use std::rc::Rc; use async_trait::async_trait; use deno_core::op2; use deno_core::JsBuffer; use deno_core::OpState; use deno_core::Resource; use deno_core::ResourceId; use tokio::sync::broadcast::error::SendError as BroadcastSendError; use tokio::sync::mpsc::error::SendError as MpscSendError; pub const UNSTABLE_FEATURE_NAME: &str = "broadcast-channel"; #[derive(Debug, thiserror::Error)] pub enum BroadcastChannelError { #[error(transparent)] Resource(deno_core::error::AnyError), #[error(transparent)] MPSCSendError(MpscSendError>), #[error(transparent)] BroadcastSendError( BroadcastSendError>, ), #[error(transparent)] Other(deno_core::error::AnyError), } impl From> for BroadcastChannelError { fn from(value: MpscSendError) -> Self { BroadcastChannelError::MPSCSendError(MpscSendError(Box::new(value.0))) } } impl From> for BroadcastChannelError { fn from(value: BroadcastSendError) -> Self { BroadcastChannelError::BroadcastSendError(BroadcastSendError(Box::new( value.0, ))) } } #[async_trait] pub trait BroadcastChannel: Clone { type Resource: Resource; fn subscribe(&self) -> Result; fn unsubscribe( &self, resource: &Self::Resource, ) -> Result<(), BroadcastChannelError>; async fn send( &self, resource: &Self::Resource, name: String, data: Vec, ) -> Result<(), BroadcastChannelError>; async fn recv( &self, resource: &Self::Resource, ) -> Result, BroadcastChannelError>; } pub type Message = (String, Vec); #[op2(fast)] #[smi] pub fn op_broadcast_subscribe( state: &mut OpState, ) -> Result where BC: BroadcastChannel + 'static, { state .feature_checker .check_or_exit(UNSTABLE_FEATURE_NAME, "BroadcastChannel"); let bc = state.borrow::(); let resource = bc.subscribe()?; Ok(state.resource_table.add(resource)) } #[op2(fast)] pub fn op_broadcast_unsubscribe( state: &mut OpState, #[smi] rid: ResourceId, ) -> Result<(), BroadcastChannelError> where BC: BroadcastChannel + 'static, { let resource = state .resource_table .get::(rid) .map_err(BroadcastChannelError::Resource)?; let bc = state.borrow::(); bc.unsubscribe(&resource) } #[op2(async)] pub async fn op_broadcast_send( state: Rc>, #[smi] rid: ResourceId, #[string] name: String, #[buffer] buf: JsBuffer, ) -> Result<(), BroadcastChannelError> where BC: BroadcastChannel + 'static, { let resource = state .borrow() .resource_table .get::(rid) .map_err(BroadcastChannelError::Resource)?; let bc = state.borrow().borrow::().clone(); bc.send(&resource, name, buf.to_vec()).await } #[op2(async)] #[serde] pub async fn op_broadcast_recv( state: Rc>, #[smi] rid: ResourceId, ) -> Result, BroadcastChannelError> where BC: BroadcastChannel + 'static, { let resource = state .borrow() .resource_table .get::(rid) .map_err(BroadcastChannelError::Resource)?; let bc = state.borrow().borrow::().clone(); bc.recv(&resource).await } deno_core::extension!(deno_broadcast_channel, deps = [ deno_webidl, deno_web ], parameters = [BC: BroadcastChannel], ops = [ op_broadcast_subscribe, op_broadcast_unsubscribe, op_broadcast_send, op_broadcast_recv, ], esm = [ "01_broadcast_channel.js" ], options = { bc: BC, }, state = |state, options| { state.put(options.bc); }, ); pub fn get_declaration() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .join("lib.deno_broadcast_channel.d.ts") }