summaryrefslogtreecommitdiff
path: root/crates
diff options
context:
space:
mode:
Diffstat (limited to 'crates')
-rw-r--r--crates/teleia/src/script.rs5
-rw-r--r--crates/teleia/src/script/bencode.rs148
-rw-r--r--crates/teleia/src/script/nrepl.rs142
3 files changed, 295 insertions, 0 deletions
diff --git a/crates/teleia/src/script.rs b/crates/teleia/src/script.rs
index 0631e31..96e8f6b 100644
--- a/crates/teleia/src/script.rs
+++ b/crates/teleia/src/script.rs
@@ -1,3 +1,8 @@
+pub mod bencode;
+
+#[cfg(not(target_arch = "wasm32"))]
+pub mod nrepl;
+
use std::{ffi::c_int, mem::MaybeUninit};
use crate::utils;
diff --git a/crates/teleia/src/script/bencode.rs b/crates/teleia/src/script/bencode.rs
new file mode 100644
index 0000000..260ffed
--- /dev/null
+++ b/crates/teleia/src/script/bencode.rs
@@ -0,0 +1,148 @@
+use std::{collections::BTreeMap, io::Write, io::BufRead};
+
+#[derive(Debug)]
+pub enum Error {
+ Encoding(String),
+ Decoding(String),
+ IO(std::io::Error),
+ Misc(simple_eyre::Report),
+}
+impl std::fmt::Display for Error {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Encoding(msg) => write!(f, "bencode encoding error: {}", msg),
+ Self::Decoding(msg) => write!(f, "bencode decoding error: {}", msg),
+ Self::IO(e) => write!(f, "{}", e),
+ Self::Misc(e) => write!(f, "{}", e),
+ }
+ }
+}
+impl std::error::Error for Error {}
+impl Error {
+ fn misc(x: impl Into<simple_eyre::Report>) -> Self { Self::Misc(x.into()) }
+}
+
+#[derive(Debug, Clone)]
+pub enum Value {
+ Integer(i64),
+ Bytestring(Vec<u8>),
+ List(Vec<Value>),
+ Dictionary(BTreeMap<Vec<u8>, Value>),
+}
+impl std::fmt::Display for Value {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Integer(i) => write!(f, "{}", i),
+ Self::Bytestring(bs) => if let Ok(s) = str::from_utf8(bs) {
+ write!(f, "{:?}", s)
+ } else {
+ write!(f, "{:?}", bs)
+ },
+ Self::List(xs) => {
+ let mut write_sep = false;
+ f.write_str("[")?;
+ for x in xs {
+ if write_sep {
+ f.write_str(" ")?;
+ } else { write_sep = true }
+ write!(f, "{}", x)?;
+ }
+ f.write_str("]")?;
+ Ok(())
+ },
+ Self::Dictionary(m) => {
+ let mut write_sep = false;
+ f.write_str("{")?;
+ for (kbs, v) in m.iter() {
+ if write_sep {
+ f.write_str(" ")?;
+ } else { write_sep = true }
+ let s = format!("{:?}", kbs);
+ let k = str::from_utf8(kbs).unwrap_or(&s);
+ write!(f, ":{} {}", k, v)?;
+ }
+ f.write_str("}")?;
+ Ok(())
+ },
+ }
+ }
+}
+impl Value {
+ fn decode_helper<R>(first: u8, r: &mut R) -> Result<Self, Error> where R: BufRead {
+ match first {
+ b'i' => {
+ let mut buf = Vec::new();
+ r.read_until(b'e', &mut buf).map_err(Error::IO)?;
+ Ok(Self::Integer(
+ str::from_utf8(&buf[..buf.len()-1]).map_err(Error::misc)?
+ .parse().map_err(Error::misc)?))
+ },
+ b'l' => {
+ let mut ret = Vec::new();
+ let mut ch: [u8; 1] = [0];
+ loop {
+ r.read_exact(&mut ch).map_err(Error::IO)?;
+ if ch[0] == b'e' { break }
+ ret.push(Self::decode_helper(ch[0], r)?);
+ }
+ Ok(Self::List(ret))
+ },
+ b'd' => {
+ let mut ret = BTreeMap::new();
+ let mut ch: [u8; 1] = [0];
+ loop {
+ r.read_exact(&mut ch).map_err(Error::IO)?;
+ if ch[0] == b'e' { break }
+ let vk = Self::decode_helper(ch[0], r)?;
+ if let Self::Bytestring(k) = vk {
+ r.read_exact(&mut ch).map_err(Error::IO)?;
+ let v = Self::decode_helper(ch[0], r)?;
+ ret.insert(k, v);
+ } else {
+ return Err(Error::Decoding(format!("non-bytestring key: {:?}", vk)));
+ }
+ }
+ Ok(Self::Dictionary(ret))
+ },
+ _ => {
+ let mut buf = Vec::new();
+ buf.push(first);
+ r.read_until(b':', &mut buf).map_err(Error::IO)?;
+ let len: usize = str::from_utf8(&buf[..buf.len()-1]).map_err(Error::misc)?
+ .parse().map_err(Error::misc)?;
+ buf = vec![0; len];
+ r.read_exact(&mut buf[..]).map_err(Error::IO)?;
+ Ok(Self::Bytestring(buf))
+ }
+ }
+ }
+ pub fn decode<R>(r: &mut R) -> Result<Self, Error> where R: BufRead {
+ let mut ch: [u8; 1] = [0];
+ r.read_exact(&mut ch).map_err(Error::IO)?;
+ Self::decode_helper(ch[0], r)
+ }
+
+ pub fn encode<W>(&self, w: &mut W) -> Result<(), Error> where W: Write {
+ match self {
+ Self::Integer(i) => w.write_all(format!("i{}e", i).as_bytes()).map_err(Error::IO)?,
+ Self::Bytestring(bs) => {
+ w.write_all(format!("{}:", bs.len()).as_bytes()).map_err(Error::IO)?;
+ w.write_all(&bs).map_err(Error::IO)?;
+ },
+ Self::List(xs) => {
+ w.write_all(b"l").map_err(Error::IO)?;
+ for x in xs { x.encode(w)?; }
+ w.write_all(b"e").map_err(Error::IO)?;
+ },
+ Self::Dictionary(m) => {
+ w.write_all(b"d").map_err(Error::IO)?;
+ for (k, v) in m {
+ Self::Bytestring(k.clone()).encode(w)?;
+ v.encode(w)?;
+ }
+ w.write_all(b"e").map_err(Error::IO)?;
+ }
+ }
+ Ok(())
+ }
+}
diff --git a/crates/teleia/src/script/nrepl.rs b/crates/teleia/src/script/nrepl.rs
new file mode 100644
index 0000000..633fa0a
--- /dev/null
+++ b/crates/teleia/src/script/nrepl.rs
@@ -0,0 +1,142 @@
+use super::bencode;
+use crate::{utils, Erm, WrapErr};
+
+use std::io::{BufReader, Read, Write};
+use std::collections::HashMap;
+use std::net::{TcpListener, TcpStream};
+use std::sync::mpsc::{Sender, Receiver, channel};
+use std::sync::{Arc, Mutex};
+use std::thread::spawn;
+
+const ACCEPT_KEY: usize = 1;
+
+type ClientId = usize;
+type ClientMessage = (ClientId, bencode::Value);
+struct ClientConnection {
+ writer: TcpStream,
+ reader: BufReader<TcpStream>,
+}
+impl ClientConnection {
+ fn new(s: TcpStream) -> Erm<Self> {
+ let reader = BufReader::new(s.try_clone()?);
+ Ok(Self {
+ writer: s,
+ reader,
+ })
+ }
+}
+
+pub struct Server {
+ clients: Arc<Mutex<HashMap<ClientId, ClientConnection>>>,
+ channels: Arc<Mutex<Option<(Sender<ClientMessage>, Receiver<ClientMessage>)>>>,
+}
+
+impl Server {
+ pub fn new() -> Self {
+ Self {
+ clients: Arc::new(Mutex::new(HashMap::new())),
+ channels: Arc::new(Mutex::new(None)),
+ }
+ }
+ pub fn start(&mut self, addr: &str) {
+ let recv_clients_ref = self.clients.clone();
+ let send_clients_ref = self.clients.clone();
+ let (outgoing_sender, outgoing_receiver) = channel();
+ let (incoming_sender, incoming_receiver) = channel();
+ *self.channels.lock().unwrap() = Some((outgoing_sender, incoming_receiver));
+ let channels_ref = self.channels.clone();
+ let listener = TcpListener::bind(addr).expect("failed to bind server socket");
+ let poller = polling::Poller::new().expect("failed to create poller for nREPL server");
+ unsafe {
+ poller.add(&listener, polling::Event::readable(ACCEPT_KEY))
+ .expect("failed to add poll server socket");
+ }
+ let mut events = polling::Events::new();
+ let mut next_id = ACCEPT_KEY + 1;
+ spawn(move || loop {
+ // write all outgoing messages
+ while let Ok((cid, msg)) = outgoing_receiver.recv() {
+ let res: Erm<()> = try {
+ if let Some(sock) = send_clients_ref.lock().unwrap().get_mut(&cid) {
+ let res = msg.encode(&mut sock.writer);
+ match res {
+ Ok(_) => {},
+ Err(bencode::Error::IO(e)) if e.kind() == std::io::ErrorKind::WouldBlock => {
+ log::info!("client {} receive send block", cid);
+ },
+ Err(e) => utils::erm(e)?,
+ }
+ }
+ };
+ if let Err(e) = res {
+ log::warn!("error when sending to client {}, disconnecting: {}", cid, e);
+ send_clients_ref.lock().unwrap().remove(&cid);
+ }
+ }
+ });
+ spawn(move || loop {
+ // wait for and receive all new connections and incoming messages
+ let res: Erm<()> = try {
+ events.clear();
+ poller.wait(&mut events, None).wrap_err("failed to wait on nREPL server poller")?;
+ for ev in events.iter() {
+ if ev.key == ACCEPT_KEY {
+ let (sock, client_addr) = listener.accept().wrap_err("failed to accept on socket")?;
+ let id = next_id; next_id += 1;
+ let res: Erm<()> = try {
+ unsafe { poller.add(&sock, polling::Event::readable(id))
+ .wrap_err("failed to add socket to poller")?; }
+ recv_clients_ref.lock().unwrap().insert(id, ClientConnection::new(sock)?);
+ log::info!("new connected client! {}", id);
+ };
+ if let Err(e) = res {
+ log::warn!("error during connection from {}: {}", client_addr, e);
+ };
+ poller.modify(&listener, polling::Event::readable(ACCEPT_KEY))
+ .wrap_err("failed to modify event to poll")?;
+ } else {
+ let res: Erm<()> = try {
+ if let Some(sock) = recv_clients_ref.lock().unwrap().get_mut(&ev.key) {
+ match bencode::Value::decode(&mut sock.reader) {
+ Ok(v) => incoming_sender.send((ev.key, v))
+ .wrap_err("failed to send incoming message on channel")?,
+ Err(e) => utils::erm(e)?,
+ }
+ poller.modify(&sock.writer, polling::Event::readable(ev.key))
+ .wrap_err("failed to modify event to poll")?;
+ } else {
+ log::warn!("poller indicated event for unknown client: {}", ev.key);
+ }
+ };
+ if let Err(e) = res {
+ log::warn!("error on client {} connection, disconnecting: {}", ev.key, e);
+ recv_clients_ref.lock().unwrap().remove(&ev.key);
+ }
+ }
+ }
+ };
+ if let Err(e) = res {
+ log::warn!("unhandled error in nREPL thread, stopping server: {}", e);
+ *channels_ref.lock().unwrap() = None;
+ return;
+ }
+ });
+ }
+ pub fn clients(&self) -> Vec<ClientId> {
+ self.clients.lock().unwrap().keys().copied().collect()
+ }
+ pub fn poll(&mut self) -> Option<(ClientId, bencode::Value)> {
+ if let Some((_, incoming)) = &*self.channels.lock().unwrap() {
+ incoming.try_recv().ok()
+ } else { None }
+ }
+ pub fn send(&self, client: ClientId, msg: bencode::Value) {
+ if let Some((outgoing, _)) = &mut *self.channels.lock().unwrap() {
+ if let Err(e) = outgoing.send((client, msg)) {
+ log::warn!("failed to send nREPL message to client {} on channel: {}", client, e);
+ }
+ } else {
+ log::warn!("tried to send message to client {}, but nREPL is not connected", client);
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-07-07 10:47:05 +0000