main
use tokio::sync::mpsc;
use crate::backend::SocketError;
use super::super::EmitPacket;
use super::super::session::{self, Input, Output};
use super::core::{SocketHub, namespace_key};
const INPUT_CAPACITY: usize = 256;
impl SocketHub {
pub(crate) async fn spawn_transport(
&self,
future: impl std::future::Future<Output = ()> + Send + 'static,
) {
let mut transports = self.inner.transport_shard().lock().await;
while let Some(result) = transports.try_join_next() {
if let Err(error) = result {
eprintln!("kenevent: WebSocket task failed: {error}");
}
}
transports.spawn(future);
}
pub(crate) async fn finish_transports(&self) {
let mut transports = Vec::with_capacity(self.inner.transports.len());
for shard in &self.inner.transports {
transports.push(shard.lock().await);
}
if tokio::time::timeout(std::time::Duration::from_secs(5), async {
for set in &mut transports {
while set.join_next().await.is_some() {}
}
})
.await
.is_err()
{
for set in &mut transports {
set.abort_all();
}
for set in &mut transports {
while set.join_next().await.is_some() {}
}
}
}
pub(crate) async fn authenticate_token(&self, token: &str) -> Result<i64, SocketError> {
tokio::time::timeout(
self.inner.options.auth_timeout,
self.inner.backend.authenticate(token),
)
.await
.map_err(|_| SocketError::Unavailable("authentication timed out".to_string()))?
}
pub(crate) async fn create_session(
&self,
sid: String,
user_id: i64,
token: String,
output: Output,
send_open: bool,
) -> mpsc::Sender<Input> {
let (input, receiver) = mpsc::channel(INPUT_CAPACITY);
self.inner
.sessions_shard(&sid)
.lock()
.await
.insert(sid.clone(), input.clone());
let hub = self.clone();
let session_input = input.clone();
tokio::spawn(async move {
session::run(
hub,
sid,
user_id,
token,
session_input,
receiver,
output,
send_open,
)
.await;
});
input
}
pub(crate) async fn remove_session(&self, sid: &str) {
self.inner.sessions_shard(sid).lock().await.remove(sid);
}
pub(crate) async fn input(&self, sid: &str) -> Option<mpsc::Sender<Input>> {
self.inner
.sessions_shard(sid)
.lock()
.await
.get(sid)
.cloned()
}
pub(crate) async fn register(
&self,
user_id: i64,
namespace: &str,
session_id: String,
input: mpsc::Sender<Input>,
) {
self.inner
.sockets_shard(user_id)
.lock()
.await
.entry(user_id)
.or_default()
.namespaces
.entry(namespace_key(namespace))
.or_default()
.insert(session_id, input);
}
pub(crate) async fn unregister(&self, user_id: i64, namespace: &str, session_id: &str) {
let mut sockets = self.inner.sockets_shard(user_id).lock().await;
if let Some(user) = sockets.get_mut(&user_id) {
let key = namespace_key(namespace);
if let Some(entries) = user.namespaces.get_mut(&key) {
entries.remove(session_id);
if entries.is_empty() {
user.namespaces.remove(&key);
}
}
if user.namespaces.is_empty() {
sockets.remove(&user_id);
}
}
}
pub(super) async fn inputs(&self, user_id: i64, namespace: &str) -> Vec<mpsc::Sender<Input>> {
self.inner
.sockets_shard(user_id)
.lock()
.await
.get(&user_id)
.and_then(|user| user.namespaces.get(namespace))
.map(|entries| entries.values().cloned().collect())
.unwrap_or_default()
}
pub(super) async fn deliver(&self, user_id: i64, namespace: &str, packet: EmitPacket) -> usize {
let mut delivered = 0;
for input in self.inputs(user_id, namespace).await {
if input
.try_send(Input::Emit {
packet: packet.clone(),
ack: None,
accepted: None,
})
.is_ok()
{
delivered += 1;
}
}
delivered
}
}