kent/consciousness
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Upgrade capnp 0.20 → 0.25, capnp-rpc 0.20 → 0.25

Browse source 
RPC trait methods changed from &mut self to self: Rc<Self> and
return types from Promise<(), Error> to impl Future<Output = Result<...>>.

Updated all Server impls across 6 files: DaemonImpl (rpc.rs),
NotifyForwarder (channels.rs), and ChannelServerImpl in all channel
crates (irc, telegram, tmux, socat). Local pry! macro replaces
capnp_rpc::pry to match the new impl Future return type.

Warning-clean workspace build.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-07 12:02:59 -04:00
parent 382ebc95aa
commit a421c3c9f3
12 changed files with 221 additions and 192 deletions
Download patch file Download diff file Expand all files Collapse all files

4
channels/irc/Cargo.toml
View file

@ -4,8 +4,8 @@ version.workspace = true
edition.workspace = true
[dependencies]
capnp = "0.20"
capnp-rpc = "0.20"
capnp = "0.25"
capnp-rpc = "0.25"
dirs = "6"
futures = "0.3"
json5 = "1.3"

57
channels/irc/src/main.rs
View file

@ -12,14 +12,12 @@
// Socket: ~/.consciousness/channels/irc.sock
use std::cell::RefCell;
use std::collections::VecDeque;
use std::io;
use std::path::PathBuf;
use std::rc::Rc;
use std::sync::Arc;
use capnp::capability::Promise;
use capnp_rpc::{pry, rpc_twoparty_capnp, twoparty, RpcSystem};
use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
use futures::AsyncReadExt;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixListener;
@ -481,12 +479,21 @@ struct ChannelServerImpl {
state: SharedState,
}
macro_rules! pry {
($e:expr) => {
match $e {
Ok(v) => v,
Err(e) => return std::future::ready(Err(e.into())),
}
};
}
impl channel_server::Server for ChannelServerImpl {
fn recv(
&mut self,
self: Rc<Self>,
params: channel_server::RecvParams,
mut results: channel_server::RecvResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let all_new = params.get_all_new();
@ -501,25 +508,25 @@ impl channel_server::Server for ChannelServerImpl {
};
results.get().set_text(&text);
Promise::ok(())
std::future::ready(Ok(()))
}
fn send(
&mut self,
self: Rc<Self>,
params: channel_server::SendParams,
_results: channel_server::SendResults,
) -> Promise<(), capnp::Error> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let message = pry!(pry!(params.get_message()).to_str()).to_string();
// Parse channel path to IRC target:
// irc.#bcachefs -> #bcachefs
// irc.pm.nick -> nick (PRIVMSG)
let target = channel_to_target(&channel);
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let state = self.state.clone();
Promise::from_future(async move {
async move {
let params = params.get()?;
let channel = params.get_channel()?.to_str()?.to_string();
let message = params.get_message()?.to_str()?.to_string();
// Parse channel path to IRC target:
// irc.#bcachefs -> #bcachefs
// irc.pm.nick -> nick (PRIVMSG)
let target = channel_to_target(&channel);
{
let mut s = state.borrow_mut();
s.send_privmsg(&target, &message).await
@ -540,25 +547,25 @@ impl channel_server::Server for ChannelServerImpl {
.push_own(log_line);
Ok(())
})
}
}
fn subscribe(
&mut self,
self: Rc<Self>,
params: channel_server::SubscribeParams,
_results: channel_server::SubscribeResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let callback = pry!(pry!(params.get()).get_callback());
self.state.borrow_mut().subscribers.push(callback);
info!("client subscribed for notifications");
Promise::ok(())
std::future::ready(Ok(()))
}
fn list(
&mut self,
self: Rc<Self>,
_params: channel_server::ListParams,
mut results: channel_server::ListResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let s = self.state.borrow();
let connected = s.connected;
@ -574,7 +581,7 @@ impl channel_server::Server for ChannelServerImpl {
);
}
Promise::ok(())
std::future::ready(Ok(()))
}
}

4
channels/socat/Cargo.toml
View file

@ -4,8 +4,8 @@ version.workspace = true
edition.workspace = true
[dependencies]
capnp = "0.20"
capnp-rpc = "0.20"
capnp = "0.25"
capnp-rpc = "0.25"
dirs = "6"
futures = "0.3"
poc-memory = { path = "../.." }

56
channels/socat/src/main.rs
View file

@ -11,8 +11,7 @@ use std::cell::RefCell;
use std::collections::BTreeMap;
use std::rc::Rc;
use capnp::capability::Promise;
use capnp_rpc::{pry, rpc_twoparty_capnp, twoparty, RpcSystem};
use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
use futures::AsyncReadExt;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
use tokio::net::{TcpStream, UnixListener, UnixStream};
@ -163,10 +162,19 @@ async fn connect_outbound(state: SharedState, label: String, addr: String) -> Re
struct ChannelServerImpl { state: SharedState }
macro_rules! pry {
($e:expr) => {
match $e {
Ok(v) => v,
Err(e) => return std::future::ready(Err(e.into())),
}
};
}
impl channel_server::Server for ChannelServerImpl {
fn recv(
&mut self, params: channel_server::RecvParams, mut results: channel_server::RecvResults,
) -> Promise<(), capnp::Error> {
self: Rc<Self>, params: channel_server::RecvParams, mut results: channel_server::RecvResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let all_new = params.get_all_new();
@ -178,12 +186,12 @@ impl channel_server::Server for ChannelServerImpl {
.unwrap_or_default();
results.get().set_text(&text);
Promise::ok(())
std::future::ready(Ok(()))
}
fn send(
&mut self, params: channel_server::SendParams, _results: channel_server::SendResults,
) -> Promise<(), capnp::Error> {
self: Rc<Self>, params: channel_server::SendParams, _results: channel_server::SendResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let message = pry!(pry!(params.get_message()).to_str()).to_string();
@ -195,12 +203,12 @@ impl channel_server::Server for ChannelServerImpl {
}
ch.log.push_own(format!("> {}", message));
}
Promise::ok(())
std::future::ready(Ok(()))
}
fn list(
&mut self, _params: channel_server::ListParams, mut results: channel_server::ListResults,
) -> Promise<(), capnp::Error> {
self: Rc<Self>, _params: channel_server::ListParams, mut results: channel_server::ListResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let s = self.state.borrow();
let channels: Vec<_> = s.channels.iter()
.map(|(name, ch)| (name.clone(), ch.writer.is_some(), ch.log.unread()))
@ -213,33 +221,33 @@ impl channel_server::Server for ChannelServerImpl {
entry.set_connected(*connected);
entry.set_unread(*unread as u32);
}
Promise::ok(())
std::future::ready(Ok(()))
}
fn subscribe(
&mut self, params: channel_server::SubscribeParams, _results: channel_server::SubscribeResults,
) -> Promise<(), capnp::Error> {
self: Rc<Self>, params: channel_server::SubscribeParams, _results: channel_server::SubscribeResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let callback = pry!(pry!(params.get()).get_callback());
self.state.borrow_mut().subscribers.push(callback);
Promise::ok(())
std::future::ready(Ok(()))
}
fn open(
&mut self, params: channel_server::OpenParams, _results: channel_server::OpenResults,
) -> Promise<(), capnp::Error> {
let params = pry!(params.get());
let label = pry!(pry!(params.get_label()).to_str()).to_string();
self: Rc<Self>, params: channel_server::OpenParams, _results: channel_server::OpenResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let state = self.state.clone();
Promise::from_future(async move {
async move {
let params = params.get()?;
let label = params.get_label()?.to_str()?.to_string();
connect_outbound(state, label.clone(), label).await
.map_err(|e| capnp::Error::failed(e))
})
}
}
fn close(
&mut self, params: channel_server::CloseParams, _results: channel_server::CloseResults,
) -> Promise<(), capnp::Error> {
self: Rc<Self>, params: channel_server::CloseParams, _results: channel_server::CloseResults,
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
@ -248,7 +256,7 @@ impl channel_server::Server for ChannelServerImpl {
info!("closing {}", channel);
ch.writer = None;
}
Promise::ok(())
std::future::ready(Ok(()))
}
}

4
channels/telegram/Cargo.toml
View file

@ -4,8 +4,8 @@ version.workspace = true
edition.workspace = true
[dependencies]
capnp = "0.20"
capnp-rpc = "0.20"
capnp = "0.25"
capnp-rpc = "0.25"
dirs = "6"
futures = "0.3"
poc-memory = { path = "../.." }

49
channels/telegram/src/main.rs
View file

@ -8,16 +8,14 @@
// don't kill the Telegram connection.
use std::cell::RefCell;
use std::collections::VecDeque;
use std::path::PathBuf;
use std::rc::Rc;
use capnp::capability::Promise;
use capnp_rpc::{pry, rpc_twoparty_capnp, twoparty, RpcSystem};
use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
use futures::AsyncReadExt;
use tokio::net::UnixListener;
use tokio_util::compat::TokioAsyncReadCompatExt;
use log::{info, warn, error};
use log::{info, error};
use poc_memory::channel_capnp::{channel_client, channel_server};
@ -228,12 +226,21 @@ struct ChannelServerImpl {
state: SharedState,
}
macro_rules! pry {
($e:expr) => {
match $e {
Ok(v) => v,
Err(e) => return std::future::ready(Err(e.into())),
}
};
}
impl channel_server::Server for ChannelServerImpl {
fn recv(
&mut self,
self: Rc<Self>,
params: channel_server::RecvParams,
mut results: channel_server::RecvResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let all_new = params.get_all_new();
@ -248,20 +255,20 @@ impl channel_server::Server for ChannelServerImpl {
};
results.get().set_text(&text);
Promise::ok(())
std::future::ready(Ok(()))
}
fn send(
&mut self,
self: Rc<Self>,
params: channel_server::SendParams,
_results: channel_server::SendResults,
) -> Promise<(), capnp::Error> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let message = pry!(pry!(params.get_message()).to_str()).to_string();
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let state = self.state.clone();
Promise::from_future(async move {
async move {
let params = params.get()?;
let _channel = params.get_channel()?.to_str()?.to_string();
let message = params.get_message()?.to_str()?.to_string();
let (url, client, chat_id) = {
let s = state.borrow();
(s.api_url("sendMessage"), s.client.clone(), s.config.chat_id)
@ -280,25 +287,25 @@ impl channel_server::Server for ChannelServerImpl {
.push_own(format!("[agent] {}", message));
}
Ok(())
})
}
}
fn subscribe(
&mut self,
self: Rc<Self>,
params: channel_server::SubscribeParams,
_results: channel_server::SubscribeResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let callback = pry!(pry!(params.get()).get_callback());
self.state.borrow_mut().subscribers.push(callback);
info!("client subscribed for notifications");
Promise::ok(())
std::future::ready(Ok(()))
}
fn list(
&mut self,
self: Rc<Self>,
_params: channel_server::ListParams,
mut results: channel_server::ListResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let s = self.state.borrow();
let connected = s.connected;
@ -313,7 +320,7 @@ impl channel_server::Server for ChannelServerImpl {
);
}
Promise::ok(())
std::future::ready(Ok(()))
}
}

4
channels/tmux/Cargo.toml
View file

@ -4,8 +4,8 @@ version.workspace = true
edition.workspace = true
[dependencies]
capnp = "0.20"
capnp-rpc = "0.20"
capnp = "0.25"
capnp-rpc = "0.25"
dirs = "6"
libc = "0.2"
scopeguard = "1"

54
channels/tmux/src/main.rs
View file

@ -12,8 +12,7 @@ use std::cell::RefCell;
use std::collections::BTreeMap;
use std::rc::Rc;
use capnp::capability::Promise;
use capnp_rpc::{pry, rpc_twoparty_capnp, twoparty, RpcSystem};
use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
use futures::AsyncReadExt;
use tokio::io::AsyncBufReadExt;
use tokio::net::UnixListener;
@ -145,12 +144,21 @@ struct ChannelServerImpl {
state: SharedState,
}
macro_rules! pry {
($e:expr) => {
match $e {
Ok(v) => v,
Err(e) => return std::future::ready(Err(e.into())),
}
};
}
impl channel_server::Server for ChannelServerImpl {
fn recv(
&mut self,
self: Rc<Self>,
params: channel_server::RecvParams,
mut results: channel_server::RecvResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let all_new = params.get_all_new();
@ -165,14 +173,14 @@ impl channel_server::Server for ChannelServerImpl {
};
results.get().set_text(&text);
Promise::ok(())
std::future::ready(Ok(()))
}
fn send(
&mut self,
self: Rc<Self>,
params: channel_server::SendParams,
_results: channel_server::SendResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let message = pry!(pry!(params.get_message()).to_str()).to_string();
@ -193,14 +201,14 @@ impl channel_server::Server for ChannelServerImpl {
log.push_own(format!("> {}", message));
}
Promise::ok(())
std::future::ready(Ok(()))
}
fn list(
&mut self,
self: Rc<Self>,
_params: channel_server::ListParams,
mut results: channel_server::ListResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let s = self.state.borrow();
let channels: Vec<_> = s.panes.keys().map(|label| {
let key = format!("tmux.{}", label);
@ -215,22 +223,22 @@ impl channel_server::Server for ChannelServerImpl {
entry.set_connected(*connected);
entry.set_unread(*unread as u32);
}
Promise::ok(())
std::future::ready(Ok(()))
}
fn subscribe(
&mut self,
self: Rc<Self>,
_params: channel_server::SubscribeParams,
_results: channel_server::SubscribeResults,
) -> Promise<(), capnp::Error> {
Promise::ok(())
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
std::future::ready(Ok(()))
}
fn open(
&mut self,
self: Rc<Self>,
params: channel_server::OpenParams,
_results: channel_server::OpenResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let label = pry!(pry!(params.get_label()).to_str()).to_string();
@ -238,15 +246,15 @@ impl channel_server::Server for ChannelServerImpl {
{
let s = self.state.borrow();
if s.panes.contains_key(&label) {
return Promise::ok(());
return std::future::ready(Ok(()));
}
}
// Find the tmux pane by name (window or pane title)
let pane_id = match find_pane_by_name(&label) {
Some(id) => id,
None => return Promise::err(capnp::Error::failed(
format!("no tmux pane named '{}'", label))),
None => return std::future::ready(Err(capnp::Error::failed(
format!("no tmux pane named '{}'", label)))),
};
info!("opening channel tmux.{} (pane {})", label, pane_id);
@ -264,14 +272,14 @@ impl channel_server::Server for ChannelServerImpl {
pipe_pane_reader(reader_state, pane).await;
});
Promise::ok(())
std::future::ready(Ok(()))
}
fn close(
&mut self,
self: Rc<Self>,
params: channel_server::CloseParams,
_results: channel_server::CloseResults,
) -> Promise<(), capnp::Error> {
) -> impl std::future::Future<Output = Result<(), capnp::Error>> {
let params = pry!(params.get());
let channel = pry!(pry!(params.get_channel()).to_str()).to_string();
let label = channel.strip_prefix("tmux.").unwrap_or(&channel).to_string();
@ -287,7 +295,7 @@ impl channel_server::Server for ChannelServerImpl {
.output();
}
Promise::ok(())
std::future::ready(Ok(()))
}
}