+ Send + Sync + 'static>(
cfg_path: Option,
disk_guid: Arc,
) -> Result {
@@ -192,19 +152,15 @@ impl RpcContext {
let account = AccountInfo::load(&secret_store).await?;
let db = base.db(&account).await?;
tracing::info!("Opened PatchDB");
- let mut docker = Docker::connect_with_unix_defaults()?;
- docker.set_timeout(Duration::from_secs(600));
- tracing::info!("Connected to Docker");
let net_controller = Arc::new(
NetController::init(
base.tor_control
.unwrap_or(SocketAddr::from(([127, 0, 0, 1], 9051))),
tor_proxy,
base.dns_bind
- .as_ref()
- .map(|v| v.as_slice())
+ .as_deref()
.unwrap_or(&[SocketAddr::from(([127, 0, 0, 1], 53))]),
- SslManager::new(&account)?,
+ SslManager::new(&account, root_ca_start_time().await?)?,
&account.hostname,
&account.key,
)
@@ -212,7 +168,7 @@ impl RpcContext {
);
tracing::info!("Initialized Net Controller");
let managers = ManagerMap::default();
- let metrics_cache = RwLock::new(None);
+ let metrics_cache = RwLock::>::new(None);
let notification_manager = NotificationManager::new(secret_store.clone());
tracing::info!("Initialized Notification Manager");
let tor_proxy_url = format!("socks5h://{tor_proxy}");
@@ -228,7 +184,6 @@ impl RpcContext {
db,
secret_store,
account: RwLock::new(account),
- docker,
net_controller,
managers,
metrics_cache,
@@ -250,7 +205,6 @@ impl RpcContext {
)
})?,
),
- config_hooks: Mutex::new(BTreeMap::new()),
client: Client::builder()
.proxy(Proxy::custom(move |url| {
if url.host_str().map_or(false, |h| h.ends_with(".onion")) {
@@ -262,19 +216,12 @@ impl RpcContext {
.build()
.with_kind(crate::ErrorKind::ParseUrl)?,
hardware: Hardware { devices, ram },
+ start_time: Instant::now(),
});
- let res = Self(seed);
- res.cleanup().await?;
+ let res = Self(seed.clone());
+ res.cleanup_and_initialize().await?;
tracing::info!("Cleaned up transient states");
- res.managers
- .init(
- &res,
- &mut res.db.handle(),
- &mut res.secret_store.acquire().await?,
- )
- .await?;
- tracing::info!("Initialized Package Managers");
Ok(res)
}
@@ -288,118 +235,150 @@ impl RpcContext {
Ok(())
}
- #[instrument(skip_all)]
- pub async fn cleanup(&self) -> Result<(), Error> {
- let mut db = self.db.handle();
- let receipts = RpcCleanReceipts::new(&mut db).await?;
- let packages = receipts.packages.get(&mut db).await?.0;
- let mut current_dependents = packages
- .keys()
- .map(|k| (k.clone(), BTreeMap::new()))
- .collect::>();
- for (package_id, package) in packages {
- for (k, v) in package
- .into_installed()
- .into_iter()
- .flat_map(|i| i.current_dependencies.0)
- {
- let mut entry: BTreeMap<_, _> = current_dependents.remove(&k).unwrap_or_default();
- entry.insert(package_id.clone(), v);
- current_dependents.insert(k, entry);
- }
- }
- for (package_id, current_dependents) in current_dependents {
- if let Some(deps) = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(&package_id)
- .and_then(|pde| pde.installed())
- .map::<_, CurrentDependents>(|i| i.current_dependents())
- .check(&mut db)
- .await?
- {
- deps.put(&mut db, &CurrentDependents(current_dependents))
- .await?;
- } else if let Some(deps) = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(&package_id)
- .and_then(|pde| pde.removing())
- .map::<_, CurrentDependents>(|i| i.current_dependents())
- .check(&mut db)
- .await?
- {
- deps.put(&mut db, &CurrentDependents(current_dependents))
- .await?;
- }
- }
- for (package_id, package) in receipts.packages.get(&mut db).await?.0 {
- if let Err(e) = async {
- match package {
- PackageDataEntry::Installing { .. }
- | PackageDataEntry::Restoring { .. }
- | PackageDataEntry::Updating { .. } => {
- cleanup_failed(self, &mut db, &package_id, &receipts.cleanup_receipts)
- .await?;
- }
- PackageDataEntry::Removing { .. } => {
- uninstall(
- self,
- &mut db,
- &mut self.secret_store.acquire().await?,
- &package_id,
- )
- .await?;
- }
- PackageDataEntry::Installed {
- installed,
- static_files,
- manifest,
- } => {
- for (volume_id, volume_info) in &*manifest.volumes {
- let tmp_path = to_tmp_path(volume_info.path_for(
- &self.datadir,
- &package_id,
- &manifest.version,
- &volume_id,
- ))
- .with_kind(ErrorKind::Filesystem)?;
- if tokio::fs::metadata(&tmp_path).await.is_ok() {
- tokio::fs::remove_dir_all(&tmp_path).await?;
- }
- }
- let status = installed.status;
- let main = match status.main {
- MainStatus::BackingUp { started, .. } => {
- if let Some(_) = started {
- MainStatus::Starting
- } else {
- MainStatus::Stopped
- }
- }
- MainStatus::Running { .. } => MainStatus::Starting,
- a => a.clone(),
- };
- let new_package = PackageDataEntry::Installed {
- installed: InstalledPackageDataEntry {
- status: Status { main, ..status },
- ..installed
- },
- static_files,
- manifest,
- };
- receipts
- .package
- .set(&mut db, new_package, &package_id)
- .await?;
+ #[instrument(skip(self))]
+ pub async fn cleanup_and_initialize(&self) -> Result<(), Error> {
+ self.db
+ .mutate(|f| {
+ let mut current_dependents = f
+ .as_package_data()
+ .keys()?
+ .into_iter()
+ .map(|k| (k.clone(), BTreeMap::new()))
+ .collect::>();
+ for (package_id, package) in f.as_package_data_mut().as_entries_mut()? {
+ for (k, v) in package
+ .as_installed_mut()
+ .into_iter()
+ .flat_map(|i| i.clone().into_current_dependencies().into_entries())
+ .flatten()
+ {
+ let mut entry: BTreeMap<_, _> =
+ current_dependents.remove(&k).unwrap_or_default();
+ entry.insert(package_id.clone(), v.de()?);
+ current_dependents.insert(k, entry);
}
}
- Ok::<_, Error>(())
- }
- .await
- {
+ for (package_id, current_dependents) in current_dependents {
+ if let Some(deps) = f
+ .as_package_data_mut()
+ .as_idx_mut(&package_id)
+ .and_then(|pde| pde.expect_as_installed_mut().ok())
+ .map(|i| i.as_installed_mut().as_current_dependents_mut())
+ {
+ deps.ser(&CurrentDependents(current_dependents))?;
+ } else if let Some(deps) = f
+ .as_package_data_mut()
+ .as_idx_mut(&package_id)
+ .and_then(|pde| pde.expect_as_removing_mut().ok())
+ .map(|i| i.as_removing_mut().as_current_dependents_mut())
+ {
+ deps.ser(&CurrentDependents(current_dependents))?;
+ }
+ }
+ Ok(())
+ })
+ .await?;
+
+ let peek = self.db.peek().await;
+
+ for (package_id, package) in peek.as_package_data().as_entries()?.into_iter() {
+ let action = match package.as_match() {
+ PackageDataEntryMatchModelRef::Installing(_)
+ | PackageDataEntryMatchModelRef::Restoring(_)
+ | PackageDataEntryMatchModelRef::Updating(_) => {
+ cleanup_failed(self, &package_id).await
+ }
+ PackageDataEntryMatchModelRef::Removing(_) => {
+ uninstall(
+ self,
+ self.secret_store.acquire().await?.as_mut(),
+ &package_id,
+ )
+ .await
+ }
+ PackageDataEntryMatchModelRef::Installed(m) => {
+ let version = m.as_manifest().as_version().clone().de()?;
+ let volumes = m.as_manifest().as_volumes().de()?;
+ for (volume_id, volume_info) in &*volumes {
+ let tmp_path = to_tmp_path(volume_info.path_for(
+ &self.datadir,
+ &package_id,
+ &version,
+ volume_id,
+ ))
+ .with_kind(ErrorKind::Filesystem)?;
+ if tokio::fs::metadata(&tmp_path).await.is_ok() {
+ tokio::fs::remove_dir_all(&tmp_path).await?;
+ }
+ }
+ Ok(())
+ }
+ _ => continue,
+ };
+ if let Err(e) = action {
tracing::error!("Failed to clean up package {}: {}", package_id, e);
tracing::debug!("{:?}", e);
}
}
+ let peek = self
+ .db
+ .mutate(|v| {
+ for (_, pde) in v.as_package_data_mut().as_entries_mut()? {
+ let status = pde
+ .expect_as_installed_mut()?
+ .as_installed_mut()
+ .as_status_mut()
+ .as_main_mut();
+ let running = status.clone().de()?.running();
+ status.ser(&if running {
+ MainStatus::Starting
+ } else {
+ MainStatus::Stopped
+ })?;
+ }
+ Ok(v.clone())
+ })
+ .await?;
+ self.managers.init(self.clone(), peek.clone()).await?;
+ tracing::info!("Initialized Package Managers");
+
+ let mut all_dependency_config_errs = BTreeMap::new();
+ for (package_id, package) in peek.as_package_data().as_entries()?.into_iter() {
+ let package = package.clone();
+ if let Some(current_dependencies) = package
+ .as_installed()
+ .and_then(|x| x.as_current_dependencies().de().ok())
+ {
+ let manifest = package.as_manifest().de()?;
+ all_dependency_config_errs.insert(
+ package_id.clone(),
+ compute_dependency_config_errs(
+ self,
+ &peek,
+ &manifest,
+ ¤t_dependencies,
+ &Default::default(),
+ )
+ .await?,
+ );
+ }
+ }
+ self.db
+ .mutate(|v| {
+ for (package_id, errs) in all_dependency_config_errs {
+ if let Some(config_errors) = v
+ .as_package_data_mut()
+ .as_idx_mut(&package_id)
+ .and_then(|pde| pde.as_installed_mut())
+ .map(|i| i.as_status_mut().as_dependency_config_errors_mut())
+ {
+ config_errors.ser(&errs)?;
+ }
+ }
+ Ok(())
+ })
+ .await?;
+
Ok(())
}
@@ -457,7 +436,7 @@ impl RpcContext {
}
impl AsRef for RpcContext {
fn as_ref(&self) -> &Jwk {
- &*CURRENT_SECRET
+ &CURRENT_SECRET
}
}
impl Context for RpcContext {}
@@ -471,7 +450,7 @@ impl Deref for RpcContext {
tracing_error::SpanTrace::capture()
);
}
- &*self.0
+ &self.0
}
}
impl Drop for RpcContext {
diff --git a/backend/src/context/sdk.rs b/backend/src/context/sdk.rs
index 87adaee1f..7ba7a6bfa 100644
--- a/backend/src/context/sdk.rs
+++ b/backend/src/context/sdk.rs
@@ -7,8 +7,8 @@ use rpc_toolkit::Context;
use serde::Deserialize;
use tracing::instrument;
+use crate::prelude::*;
use crate::util::config::{load_config_from_paths, local_config_path};
-use crate::{Error, ResultExt};
#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "kebab-case")]
@@ -50,21 +50,21 @@ impl SdkContext {
}
/// BLOCKING
#[instrument(skip_all)]
- pub fn developer_key(&self) -> Result {
+ pub fn developer_key(&self) -> Result {
if !self.developer_key_path.exists() {
- return Err(Error::new(eyre!("Developer Key does not exist! Please run `embassy-sdk init` before running this command."), crate::ErrorKind::Uninitialized));
+ return Err(Error::new(eyre!("Developer Key does not exist! Please run `start-sdk init` before running this command."), crate::ErrorKind::Uninitialized));
}
let pair = ::from_pkcs8_pem(
&std::fs::read_to_string(&self.developer_key_path)?,
)
.with_kind(crate::ErrorKind::Pem)?;
- let secret = ed25519_dalek::SecretKey::from_bytes(&pair.secret_key[..])?;
- let public = if let Some(public) = pair.public_key {
- ed25519_dalek::PublicKey::from_bytes(&public[..])?
- } else {
- (&secret).into()
- };
- Ok(ed25519_dalek::Keypair { secret, public })
+ let secret = ed25519_dalek::SecretKey::try_from(&pair.secret_key[..]).map_err(|_| {
+ Error::new(
+ eyre!("pkcs8 key is of incorrect length"),
+ ErrorKind::OpenSsl,
+ )
+ })?;
+ Ok(secret.into())
}
}
impl std::ops::Deref for SdkContext {
diff --git a/backend/src/control.rs b/backend/src/control.rs
index 794afc64a..58e39ac14 100644
--- a/backend/src/control.rs
+++ b/backend/src/control.rs
@@ -1,220 +1,85 @@
-use std::collections::BTreeMap;
-
use color_eyre::eyre::eyre;
-use patch_db::{DbHandle, LockReceipt, LockType};
use rpc_toolkit::command;
use tracing::instrument;
use crate::context::RpcContext;
-use crate::dependencies::{
- break_all_dependents_transitive, heal_all_dependents_transitive, BreakageRes, DependencyError,
- DependencyReceipt, TaggedDependencyError,
-};
+use crate::prelude::*;
use crate::s9pk::manifest::PackageId;
use crate::status::MainStatus;
use crate::util::display_none;
-use crate::util::serde::display_serializable;
use crate::Error;
-#[derive(Clone)]
-pub struct StartReceipts {
- dependency_receipt: DependencyReceipt,
- status: LockReceipt,
- version: LockReceipt,
-}
-
-impl StartReceipts {
- pub async fn new(db: &mut impl DbHandle, id: &PackageId) -> Result {
- let mut locks = Vec::new();
-
- let setup = Self::setup(&mut locks, id);
- setup(&db.lock_all(locks).await?)
- }
-
- pub fn setup(
- locks: &mut Vec,
- id: &PackageId,
- ) -> impl FnOnce(&patch_db::Verifier) -> Result {
- let dependency_receipt = DependencyReceipt::setup(locks);
- let status = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(id)
- .and_then(|x| x.installed())
- .map(|x| x.status().main())
- .make_locker(LockType::Write)
- .add_to_keys(locks);
- let version = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(id)
- .and_then(|x| x.installed())
- .map(|x| x.manifest().version())
- .make_locker(LockType::Read)
- .add_to_keys(locks);
- move |skeleton_key| {
- Ok(Self {
- dependency_receipt: dependency_receipt(skeleton_key)?,
- status: status.verify(skeleton_key)?,
- version: version.verify(skeleton_key)?,
- })
- }
- }
-}
-
#[command(display(display_none), metadata(sync_db = true))]
#[instrument(skip_all)]
pub async fn start(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(), Error> {
- let mut db = ctx.db.handle();
- let mut tx = db.begin().await?;
- let receipts = StartReceipts::new(&mut tx, &id).await?;
- let version = receipts.version.get(&mut tx).await?;
- receipts.status.set(&mut tx, MainStatus::Starting).await?;
- heal_all_dependents_transitive(&ctx, &mut tx, &id, &receipts.dependency_receipt).await?;
-
- tx.commit().await?;
- drop(receipts);
+ let peek = ctx.db.peek().await;
+ let version = peek
+ .as_package_data()
+ .as_idx(&id)
+ .or_not_found(&id)?
+ .as_installed()
+ .or_not_found(&id)?
+ .as_manifest()
+ .as_version()
+ .de()?;
ctx.managers
.get(&(id, version))
.await
.ok_or_else(|| Error::new(eyre!("Manager not found"), crate::ErrorKind::InvalidRequest))?
- .start();
+ .start()
+ .await;
Ok(())
}
-#[derive(Clone)]
-pub struct StopReceipts {
- breaks: crate::dependencies::BreakTransitiveReceipts,
- status: LockReceipt,
-}
-impl StopReceipts {
- pub async fn new<'a>(db: &'a mut impl DbHandle, id: &PackageId) -> Result {
- let mut locks = Vec::new();
+#[command(display(display_none), metadata(sync_db = true))]
+pub async fn stop(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result {
+ let peek = ctx.db.peek().await;
+ let version = peek
+ .as_package_data()
+ .as_idx(&id)
+ .or_not_found(&id)?
+ .as_installed()
+ .or_not_found(&id)?
+ .as_manifest()
+ .as_version()
+ .de()?;
- let setup = Self::setup(&mut locks, id);
- setup(&db.lock_all(locks).await?)
- }
+ let last_statuts = ctx
+ .db
+ .mutate(|v| {
+ v.as_package_data_mut()
+ .as_idx_mut(&id)
+ .and_then(|x| x.as_installed_mut())
+ .ok_or_else(|| Error::new(eyre!("{} is not installed", id), ErrorKind::NotFound))?
+ .as_status_mut()
+ .as_main_mut()
+ .replace(&MainStatus::Stopping)
+ })
+ .await?;
- pub fn setup(
- locks: &mut Vec,
- id: &PackageId,
- ) -> impl FnOnce(&patch_db::Verifier) -> Result {
- let breaks = crate::dependencies::BreakTransitiveReceipts::setup(locks);
- let status = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(id)
- .and_then(|x| x.installed())
- .map(|x| x.status().main())
- .make_locker(LockType::Write)
- .add_to_keys(locks);
- move |skeleton_key| {
- Ok(Self {
- breaks: breaks(skeleton_key)?,
- status: status.verify(skeleton_key)?,
- })
- }
- }
-}
-
-#[instrument(skip_all)]
-pub async fn stop_common(
- db: &mut Db,
- id: &PackageId,
- breakages: &mut BTreeMap,
-) -> Result {
- let mut tx = db.begin().await?;
- let receipts = StopReceipts::new(&mut tx, id).await?;
- let last_status = receipts.status.get(&mut tx).await?;
- receipts.status.set(&mut tx, MainStatus::Stopping).await?;
-
- tx.save().await?;
- break_all_dependents_transitive(
- db,
- id,
- DependencyError::NotRunning,
- breakages,
- &receipts.breaks,
- )
- .await?;
-
- Ok(last_status)
-}
-
-#[command(
- subcommands(self(stop_impl(async)), stop_dry),
- display(display_none),
- metadata(sync_db = true)
-)]
-pub fn stop(#[arg] id: PackageId) -> Result {
- Ok(id)
-}
-
-#[command(rename = "dry", display(display_serializable))]
-#[instrument(skip_all)]
-pub async fn stop_dry(
- #[context] ctx: RpcContext,
- #[parent_data] id: PackageId,
-) -> Result {
- let mut db = ctx.db.handle();
- let mut tx = db.begin().await?;
-
- let mut breakages = BTreeMap::new();
- stop_common(&mut tx, &id, &mut breakages).await?;
-
- tx.abort().await?;
-
- Ok(BreakageRes(breakages))
-}
-
-#[instrument(skip_all)]
-pub async fn stop_impl(ctx: RpcContext, id: PackageId) -> Result {
- let mut db = ctx.db.handle();
- let mut tx = db.begin().await?;
- let version = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(&id)
- .expect(&mut tx)
- .await?
- .installed()
- .expect(&mut tx)
- .await?
- .manifest()
- .version()
- .get(&mut tx)
- .await?
- .clone();
-
- let last_statuts = stop_common(&mut tx, &id, &mut BTreeMap::new()).await?;
-
- tx.commit().await?;
ctx.managers
.get(&(id, version))
.await
.ok_or_else(|| Error::new(eyre!("Manager not found"), crate::ErrorKind::InvalidRequest))?
- .stop();
+ .stop()
+ .await;
Ok(last_statuts)
}
#[command(display(display_none), metadata(sync_db = true))]
pub async fn restart(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(), Error> {
- let mut db = ctx.db.handle();
- let mut tx = db.begin().await?;
- let version = crate::db::DatabaseModel::new()
- .package_data()
- .idx_model(&id)
- .expect(&mut tx)
- .await?
- .installed()
- .expect(&mut tx)
- .await?
- .manifest()
- .version()
- .get(&mut tx)
- .await?
- .clone();
-
- tx.commit().await?;
+ let peek = ctx.db.peek().await;
+ let version = peek
+ .as_package_data()
+ .as_idx(&id)
+ .or_not_found(&id)?
+ .expect_as_installed()?
+ .as_manifest()
+ .as_version()
+ .de()?;
ctx.managers
.get(&(id, version))
diff --git a/backend/src/db/mod.rs b/backend/src/db/mod.rs
index a8cc11d1f..03ad94338 100644
--- a/backend/src/db/mod.rs
+++ b/backend/src/db/mod.rs
@@ -1,13 +1,14 @@
pub mod model;
pub mod package;
+pub mod prelude;
use std::future::Future;
+use std::path::PathBuf;
use std::sync::Arc;
-use color_eyre::eyre::eyre;
use futures::{FutureExt, SinkExt, StreamExt};
use patch_db::json_ptr::JsonPointer;
-use patch_db::{DbHandle, Dump, LockType, Revision};
+use patch_db::{Dump, Revision};
use rpc_toolkit::command;
use rpc_toolkit::hyper::upgrade::Upgraded;
use rpc_toolkit::hyper::{Body, Error as HyperError, Request, Response};
@@ -22,12 +23,11 @@ use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::WebSocketStream;
use tracing::instrument;
-pub use self::model::DatabaseModel;
-use crate::context::RpcContext;
+use crate::context::{CliContext, RpcContext};
use crate::middleware::auth::{HasValidSession, HashSessionToken};
+use crate::prelude::*;
use crate::util::display_none;
use crate::util::serde::{display_serializable, IoFormat};
-use crate::{Error, ResultExt};
#[instrument(skip_all)]
async fn ws_handler<
@@ -37,11 +37,11 @@ async fn ws_handler<
session: Option<(HasValidSession, HashSessionToken)>,
ws_fut: WSFut,
) -> Result<(), Error> {
- let (dump, sub) = ctx.db.dump_and_sub().await?;
+ let (dump, sub) = ctx.db.dump_and_sub().await;
let mut stream = ws_fut
.await
- .with_kind(crate::ErrorKind::Network)?
- .with_kind(crate::ErrorKind::Unknown)?;
+ .with_kind(ErrorKind::Network)?
+ .with_kind(ErrorKind::Unknown)?;
if let Some((session, token)) = session {
let kill = subscribe_to_session_kill(&ctx, token).await;
@@ -55,7 +55,7 @@ async fn ws_handler<
reason: "UNAUTHORIZED".into(),
}))
.await
- .with_kind(crate::ErrorKind::Network)?;
+ .with_kind(ErrorKind::Network)?;
}
Ok(())
@@ -82,6 +82,8 @@ async fn deal_with_messages(
mut sub: patch_db::Subscriber,
mut stream: WebSocketStream,
) -> Result<(), Error> {
+ let mut timer = tokio::time::interval(tokio::time::Duration::from_secs(5));
+
loop {
futures::select! {
_ = (&mut kill).fuse() => {
@@ -92,18 +94,18 @@ async fn deal_with_messages(
reason: "UNAUTHORIZED".into(),
}))
.await
- .with_kind(crate::ErrorKind::Network)?;
+ .with_kind(ErrorKind::Network)?;
return Ok(())
}
new_rev = sub.recv().fuse() => {
let rev = new_rev.expect("UNREACHABLE: patch-db is dropped");
stream
- .send(Message::Text(serde_json::to_string(&rev).with_kind(crate::ErrorKind::Serialization)?))
+ .send(Message::Text(serde_json::to_string(&rev).with_kind(ErrorKind::Serialization)?))
.await
- .with_kind(crate::ErrorKind::Network)?;
+ .with_kind(ErrorKind::Network)?;
}
message = stream.next().fuse() => {
- let message = message.transpose().with_kind(crate::ErrorKind::Network)?;
+ let message = message.transpose().with_kind(ErrorKind::Network)?;
match message {
None => {
tracing::info!("Closing WebSocket: Stream Finished");
@@ -112,6 +114,13 @@ async fn deal_with_messages(
_ => (),
}
}
+ // This is trying to give a health checks to the home to keep the ui alive.
+ _ = timer.tick().fuse() => {
+ stream
+ .send(Message::Ping(vec![]))
+ .await
+ .with_kind(crate::ErrorKind::Network)?;
+ }
}
}
}
@@ -123,10 +132,10 @@ async fn send_dump(
) -> Result<(), Error> {
stream
.send(Message::Text(
- serde_json::to_string(&dump).with_kind(crate::ErrorKind::Serialization)?,
+ serde_json::to_string(&dump).with_kind(ErrorKind::Serialization)?,
))
.await
- .with_kind(crate::ErrorKind::Network)?;
+ .with_kind(ErrorKind::Network)?;
Ok(())
}
@@ -141,7 +150,7 @@ pub async fn subscribe(ctx: RpcContext, req: Request) -> Result
+ 
+ 
-
- 
+
+ 
-
+
+ 
+
+
- 
+ 
- 
+ 
- 
+ 
- 
+ 
- 
+ 