710 lines
22 KiB
Rust
710 lines
22 KiB
Rust
use std::collections::hash_map::DefaultHasher;
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::hash::{Hash, Hasher};
|
|
use std::sync::LazyLock;
|
|
use std::time::Duration;
|
|
|
|
use anyhow::{Error, Result};
|
|
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
|
|
use nostr_sdk::prelude::*;
|
|
use smallvec::{SmallVec, smallvec};
|
|
use state::NostrRegistry;
|
|
|
|
mod group;
|
|
mod protocol;
|
|
|
|
pub use group::*;
|
|
pub use protocol::*;
|
|
|
|
pub(crate) static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
|
|
|
|
const LIST_SUBSCRIPTION: &str = "nip29-list";
|
|
const GROUP_PREFIX: &str = "grp-";
|
|
const BROWSE_PREFIX: &str = "brw-";
|
|
const BROWSE_LIMIT: usize = 500;
|
|
const BROWSE_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
pub fn init(cx: &mut App) {
|
|
GroupsRegistry::set_global(cx.new(GroupsRegistry::new), cx);
|
|
}
|
|
|
|
struct GlobalGroupsRegistry(Entity<GroupsRegistry>);
|
|
|
|
impl Global for GlobalGroupsRegistry {}
|
|
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum GroupsEvent {
|
|
Updated,
|
|
Error(String),
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct Browse {
|
|
pub relay: RelayUrl,
|
|
pub loading: bool,
|
|
pub groups: Vec<GroupCandidate>,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
enum Signal {
|
|
Event(SubscriptionId, Event),
|
|
Eose(SubscriptionId),
|
|
Closed(SubscriptionId, String),
|
|
}
|
|
|
|
fn list_subscription_id() -> SubscriptionId {
|
|
SubscriptionId::new(LIST_SUBSCRIPTION)
|
|
}
|
|
|
|
fn subscription_id(key: &GroupKey) -> SubscriptionId {
|
|
SubscriptionId::new(format!("{GROUP_PREFIX}{:08x}", key.uniq_id() as u32))
|
|
}
|
|
|
|
fn browse_subscription(relay: &RelayUrl) -> SubscriptionId {
|
|
let mut hasher = DefaultHasher::new();
|
|
relay.hash(&mut hasher);
|
|
|
|
SubscriptionId::new(format!("{BROWSE_PREFIX}{:08x}", hasher.finish() as u32))
|
|
}
|
|
|
|
fn is_own_subscription(id: &SubscriptionId) -> bool {
|
|
let id = id.as_str();
|
|
id == LIST_SUBSCRIPTION || id.starts_with(GROUP_PREFIX) || id.starts_with(BROWSE_PREFIX)
|
|
}
|
|
|
|
impl EventEmitter<GroupsEvent> for GroupsRegistry {}
|
|
|
|
pub struct GroupsRegistry {
|
|
groups: Vec<Entity<Group>>,
|
|
list: GroupList,
|
|
browse: HashMap<SubscriptionId, Browse>,
|
|
routes: HashMap<SubscriptionId, GroupKey>,
|
|
observers: HashMap<GroupKey, Subscription>,
|
|
signal_tx: flume::Sender<Signal>,
|
|
signal_rx: flume::Receiver<Signal>,
|
|
notification_listener: Option<Task<()>>,
|
|
signal_consumer: Option<Task<()>>,
|
|
tasks: SmallVec<[Task<()>; 2]>,
|
|
_subscriptions: SmallVec<[Subscription; 2]>,
|
|
}
|
|
|
|
impl GroupsRegistry {
|
|
pub fn global(cx: &App) -> Entity<Self> {
|
|
cx.global::<GlobalGroupsRegistry>().0.clone()
|
|
}
|
|
|
|
fn set_global(state: Entity<Self>, cx: &mut App) {
|
|
cx.set_global(GlobalGroupsRegistry(state));
|
|
}
|
|
|
|
fn new(cx: &mut Context<Self>) -> Self {
|
|
let nostr = NostrRegistry::global(cx);
|
|
let this = cx.entity().downgrade();
|
|
|
|
let (tx, rx) = flume::unbounded::<Signal>();
|
|
let mut subscriptions = smallvec![];
|
|
|
|
subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| {
|
|
if event.signer_changed() {
|
|
this.reset(cx);
|
|
this.handle_notifications(cx);
|
|
this.load(cx);
|
|
}
|
|
}));
|
|
|
|
cx.defer(move |cx| {
|
|
if let Err(error) = this.update(cx, |this, cx| {
|
|
this.handle_notifications(cx);
|
|
|
|
if nostr.read(cx).current_user().is_some() {
|
|
this.load(cx);
|
|
}
|
|
}) {
|
|
log::warn!("nip29: initializing the registry failed: {error}");
|
|
}
|
|
});
|
|
|
|
Self {
|
|
groups: Vec::new(),
|
|
list: GroupList::default(),
|
|
browse: HashMap::new(),
|
|
routes: HashMap::new(),
|
|
observers: HashMap::new(),
|
|
signal_tx: tx,
|
|
signal_rx: rx,
|
|
notification_listener: None,
|
|
signal_consumer: None,
|
|
tasks: smallvec![],
|
|
_subscriptions: subscriptions,
|
|
}
|
|
}
|
|
|
|
pub fn groups(&self) -> &[Entity<Group>] {
|
|
&self.groups
|
|
}
|
|
|
|
pub fn group(&self, key: &GroupKey, cx: &App) -> Option<Entity<Group>> {
|
|
self.groups
|
|
.iter()
|
|
.find(|group| group.read(cx).key() == key)
|
|
.cloned()
|
|
}
|
|
|
|
pub fn join(&mut self, reference: GroupReference, cx: &mut Context<Self>) {
|
|
self.join_key(reference.key(), reference.invite().map(str::to_owned), cx);
|
|
}
|
|
|
|
pub fn join_key(&mut self, key: GroupKey, invite: Option<String>, cx: &mut Context<Self>) {
|
|
self.ensure(key.clone(), cx);
|
|
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
self.upsert_entry(&key, None);
|
|
self.publish_list(cx);
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
|
|
let builder = key.id().join_request(invite.as_deref(), "");
|
|
let sent = group.read(cx).publish(builder, cx);
|
|
|
|
group.update(cx, |group, cx| {
|
|
group.mark_pending(Timestamp::now(), cx);
|
|
});
|
|
|
|
self.tasks.push(cx.spawn(async move |this, cx| {
|
|
if let Err(error) = sent.await {
|
|
let reason = error.to_string();
|
|
|
|
if let Err(error) = this.update(cx, |this, cx| {
|
|
if let Some(group) = this.group(&key, cx) {
|
|
group.update(cx, |group, cx| {
|
|
group.mark_refused(reason, cx);
|
|
});
|
|
}
|
|
}) {
|
|
log::warn!("nip29: marking the refused join failed: {error}");
|
|
}
|
|
}
|
|
}));
|
|
}
|
|
|
|
pub fn create(&mut self, key: GroupKey, metadata: GroupMetadata, cx: &mut Context<Self>) {
|
|
self.ensure(key.clone(), cx);
|
|
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
self.upsert_entry(&key, metadata.name().map(str::to_owned));
|
|
self.publish_list(cx);
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
|
|
let builder = key.id().create_group(&metadata, None);
|
|
let published = group.read(cx).publish(builder, cx);
|
|
let logged = key.clone();
|
|
|
|
self.tasks.push(cx.spawn(async move |this, cx| {
|
|
if let Err(e) = published.await {
|
|
this.update(cx, |this, cx| {
|
|
this.forget(logged, cx);
|
|
cx.emit(GroupsEvent::Error(e.to_string()));
|
|
})
|
|
.ok();
|
|
}
|
|
}));
|
|
}
|
|
|
|
pub fn browse(&mut self, relay: RelayUrl, cx: &mut Context<Self>) {
|
|
let nostr = NostrRegistry::global(cx);
|
|
let client = nostr.read(cx).client();
|
|
let id = browse_subscription(&relay);
|
|
|
|
self.browse.insert(
|
|
id.clone(),
|
|
Browse {
|
|
relay: relay.clone(),
|
|
loading: true,
|
|
groups: Vec::new(),
|
|
},
|
|
);
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
|
|
self.tasks.push(cx.background_spawn(async move {
|
|
client.add_relay(relay.clone()).and_connect().await.ok();
|
|
|
|
let filter = Filter::new().kind(Kind::GroupMetadata).limit(BROWSE_LIMIT);
|
|
let opts = SubscribeAutoCloseOptions::default()
|
|
.exit_policy(ReqExitPolicy::ExitOnEOSE)
|
|
.timeout(Some(BROWSE_TIMEOUT));
|
|
|
|
if let Err(error) = client
|
|
.subscribe(ReqTarget::single(relay, [filter]))
|
|
.with_id(id)
|
|
.close_on(opts)
|
|
.await
|
|
{
|
|
log::warn!("nip29: browsing failed: {error}");
|
|
}
|
|
}));
|
|
}
|
|
|
|
pub fn browsed(&self, relay: &RelayUrl) -> Option<&Browse> {
|
|
self.browse.get(&browse_subscription(relay))
|
|
}
|
|
|
|
pub fn leave(&mut self, key: &GroupKey, cx: &mut Context<Self>) {
|
|
let Some(group) = self.group(key, cx) else {
|
|
return;
|
|
};
|
|
|
|
let builder = key.id().leave_request("");
|
|
let sent = group.read_with(cx, |group, cx| group.publish(builder, cx));
|
|
let key = key.clone();
|
|
|
|
self.tasks.push(cx.spawn(async move |this, cx| {
|
|
if let Err(error) = sent.await {
|
|
log::warn!("nip29: leaving {} failed: {error}", key.id());
|
|
return;
|
|
}
|
|
this.update(cx, |this, cx| {
|
|
this.forget(key, cx);
|
|
})
|
|
.ok();
|
|
}));
|
|
}
|
|
|
|
pub fn reset(&mut self, cx: &mut Context<Self>) {
|
|
self.notification_listener = None;
|
|
self.signal_consumer = None;
|
|
self.tasks.clear();
|
|
|
|
let routes = std::mem::take(&mut self.routes);
|
|
|
|
self.groups.clear();
|
|
self.observers.clear();
|
|
self.list = GroupList::default();
|
|
|
|
let nostr = NostrRegistry::global(cx);
|
|
let client = nostr.read(cx).client();
|
|
|
|
let mut ids: Vec<SubscriptionId> = routes.into_keys().collect();
|
|
ids.push(list_subscription_id());
|
|
ids.extend(self.browse.keys().cloned());
|
|
|
|
self.browse.clear();
|
|
|
|
self.tasks.push(cx.background_spawn(async move {
|
|
for id in ids {
|
|
if let Err(error) = client.unsubscribe(&id).await {
|
|
log::warn!("nip29: unsubscribing {id} failed: {error}");
|
|
}
|
|
}
|
|
}));
|
|
|
|
cx.notify();
|
|
}
|
|
|
|
fn load(&mut self, cx: &mut Context<Self>) {
|
|
let nostr = NostrRegistry::global(cx);
|
|
let client = nostr.read(cx).client();
|
|
let signer = nostr.read(cx).signer();
|
|
|
|
let task: Task<Result<GroupList, Error>> = cx.background_spawn(async move {
|
|
let me = signer.get_public_key_async().await?;
|
|
let filter = Filter::new().author(me).kind(Kind::SimpleGroups).limit(1);
|
|
|
|
client
|
|
.subscribe(ReqTarget::auto([filter.clone()]))
|
|
.with_id(list_subscription_id())
|
|
.await?;
|
|
|
|
match client.database().query(filter).await?.into_iter().next() {
|
|
Some(event) => GroupList::parse(&event),
|
|
None => Ok(GroupList::default()),
|
|
}
|
|
});
|
|
|
|
self.tasks
|
|
.push(cx.spawn(async move |this, cx| match task.await {
|
|
Ok(list) => {
|
|
if let Err(error) = this.update(cx, |this, cx| this.merge(list, cx)) {
|
|
log::warn!("nip29: applying the group list failed: {error}");
|
|
}
|
|
}
|
|
Err(error) => log::warn!("nip29: loading the group list failed: {error}"),
|
|
}));
|
|
}
|
|
|
|
fn merge(&mut self, list: GroupList, cx: &mut Context<Self>) {
|
|
let entries = list.entries().to_vec();
|
|
|
|
for entry in entries {
|
|
self.list.upsert(entry.clone());
|
|
self.ensure(GroupKey::new(entry.relay.clone(), entry.id.clone()), cx);
|
|
}
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
}
|
|
|
|
fn ensure(&mut self, key: GroupKey, cx: &mut Context<Self>) {
|
|
if self.group(&key, cx).is_some() {
|
|
return;
|
|
}
|
|
|
|
let group = cx.new(|_| Group::new(key.clone()));
|
|
self.groups.push(group.clone());
|
|
|
|
self.observers.insert(
|
|
key.clone(),
|
|
cx.subscribe(&group, |this, group, event, cx| match event {
|
|
GroupEvent::Sent(event) => {
|
|
let key = group.read(cx).key().clone();
|
|
this.reload(key, Some(event.clone()), cx);
|
|
}
|
|
GroupEvent::Updated => {
|
|
this.refresh_entry(group, cx);
|
|
}
|
|
_ => {}
|
|
}),
|
|
);
|
|
|
|
self.subscribe(key.clone(), cx);
|
|
self.reload(key, None, cx);
|
|
}
|
|
|
|
fn subscribe(&mut self, key: GroupKey, cx: &mut Context<Self>) {
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
self.routes.insert(subscription_id(&key), key);
|
|
|
|
group.update(cx, |group, cx| {
|
|
group.connect(cx);
|
|
});
|
|
}
|
|
|
|
fn forget(&mut self, key: GroupKey, cx: &mut Context<Self>) {
|
|
let id = subscription_id(&key);
|
|
let cache_tag = key.cache_tag();
|
|
|
|
self.routes.remove(&id);
|
|
self.observers.remove(&key);
|
|
self.groups.retain(|group| group.read(cx).key() != &key);
|
|
self.list.remove(&key);
|
|
|
|
let nostr = NostrRegistry::global(cx);
|
|
let client = nostr.read(cx).client();
|
|
|
|
self.tasks.push(cx.background_spawn(async move {
|
|
if let Err(error) = client.unsubscribe(&id).await {
|
|
log::warn!("nip29: unsubscribing {id} failed: {error}");
|
|
}
|
|
|
|
let filter = Filter::new()
|
|
.kind(Kind::ApplicationSpecificData)
|
|
.custom_tag(SingleLetterTag::LOWERCASE_R, cache_tag);
|
|
|
|
if let Err(error) = client.database().delete(filter).await {
|
|
log::warn!("nip29: clearing the group cache failed: {error}");
|
|
}
|
|
}));
|
|
|
|
self.publish_list(cx);
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
}
|
|
|
|
fn entry_name(&self, key: &GroupKey) -> Option<String> {
|
|
self.list
|
|
.entries()
|
|
.iter()
|
|
.find(|entry| entry.id == *key.id() && entry.relay == *key.relay())
|
|
.and_then(|entry| entry.name.clone())
|
|
}
|
|
|
|
fn upsert_entry(&mut self, key: &GroupKey, name: Option<String>) {
|
|
let name = name.or_else(|| self.entry_name(key));
|
|
|
|
self.list.upsert(GroupListEntry {
|
|
id: key.id().clone(),
|
|
relay: key.relay().clone(),
|
|
name,
|
|
});
|
|
}
|
|
|
|
fn refresh_entry(&mut self, group: Entity<Group>, cx: &mut Context<Self>) {
|
|
let (key, Some(name)) = ({
|
|
let group = group.read(cx);
|
|
|
|
(
|
|
group.key().clone(),
|
|
group
|
|
.metadata()
|
|
.and_then(|metadata| metadata.name().map(str::to_owned)),
|
|
)
|
|
}) else {
|
|
return;
|
|
};
|
|
|
|
if self.entry_name(&key).as_deref() == Some(name.as_str()) {
|
|
return;
|
|
}
|
|
|
|
self.upsert_entry(&key, Some(name));
|
|
self.publish_list(cx);
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
}
|
|
|
|
fn publish_list(&mut self, cx: &mut Context<Self>) {
|
|
let nostr = NostrRegistry::global(cx);
|
|
let client = nostr.read(cx).client();
|
|
let signer = nostr.read(cx).signer();
|
|
|
|
let builder = self.list.to_builder();
|
|
|
|
self.tasks.push(cx.spawn(async move |_this, _cx| {
|
|
let event = match builder.finalize_async(&signer).await {
|
|
Ok(event) => event,
|
|
Err(error) => {
|
|
log::warn!("nip29: signing the group list failed: {error}");
|
|
return;
|
|
}
|
|
};
|
|
|
|
match client.send_event(&event).to_nip65().await {
|
|
Ok(output) if output.failed.is_empty() => {}
|
|
Ok(output) => log::warn!(
|
|
"nip29: {} relay(s) rejected the group list",
|
|
output.failed.len()
|
|
),
|
|
Err(error) => log::warn!("nip29: publishing the group list failed: {error}"),
|
|
}
|
|
}));
|
|
}
|
|
|
|
fn reload(&mut self, key: GroupKey, event: Option<Event>, cx: &mut Context<Self>) {
|
|
let nostr = NostrRegistry::global(cx);
|
|
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
let Some(me) = nostr.read(cx).current_user() else {
|
|
return;
|
|
};
|
|
|
|
let logged = key.clone();
|
|
let load = group.read(cx).load_envelopes(cx);
|
|
|
|
self.tasks.push(cx.spawn(async move |_this, cx| {
|
|
if let Some(event) = event
|
|
&& let Err(error) = group
|
|
.read_with(cx, |group, cx| group.save_envelope(event, cx))
|
|
.await
|
|
{
|
|
log::warn!("nip29: caching failed: {error}");
|
|
return;
|
|
}
|
|
|
|
match load.await {
|
|
Ok(events) => {
|
|
group.update(cx, |group, cx| {
|
|
group.hydrate(events, me, cx);
|
|
});
|
|
}
|
|
Err(error) => {
|
|
log::warn!("nip29: reloading {} failed: {error}", logged.id());
|
|
}
|
|
}
|
|
}));
|
|
}
|
|
|
|
fn handle_event(&mut self, id: SubscriptionId, event: Event, cx: &mut Context<Self>) {
|
|
if id == list_subscription_id() {
|
|
match GroupList::parse(&event) {
|
|
Ok(list) => self.merge(list, cx),
|
|
Err(error) => log::warn!("nip29: ignoring a malformed group list: {error}"),
|
|
}
|
|
return;
|
|
}
|
|
|
|
if self.browse.contains_key(&id) {
|
|
self.absorb_candidate(&id, &event, cx);
|
|
return;
|
|
}
|
|
|
|
let Some(key) = self.routes.get(&id).cloned() else {
|
|
return;
|
|
};
|
|
|
|
self.reload(key, Some(event), cx);
|
|
}
|
|
|
|
fn absorb_candidate(&mut self, id: &SubscriptionId, event: &Event, cx: &mut Context<Self>) {
|
|
let Some(browse) = self.browse.get_mut(id) else {
|
|
return;
|
|
};
|
|
|
|
let Ok(metadata) = GroupMetadata::parse(event) else {
|
|
log::warn!(
|
|
"nip29: ignoring malformed group metadata from {}",
|
|
browse.relay
|
|
);
|
|
return;
|
|
};
|
|
|
|
let Some(group) = metadata.id().cloned() else {
|
|
return;
|
|
};
|
|
|
|
let candidate = GroupCandidate {
|
|
key: GroupKey::new(browse.relay.clone(), group),
|
|
metadata,
|
|
};
|
|
|
|
match browse.groups.iter_mut().find(|e| e.key == candidate.key) {
|
|
Some(existing) => *existing = candidate,
|
|
None => browse.groups.push(candidate),
|
|
}
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
}
|
|
|
|
fn handle_eose(&mut self, id: SubscriptionId, cx: &mut Context<Self>) {
|
|
if let Some(browse) = self.browse.get_mut(&id) {
|
|
browse.loading = false;
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
return;
|
|
}
|
|
|
|
let Some(key) = self.routes.get(&id).cloned() else {
|
|
return;
|
|
};
|
|
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
group.update(cx, |group, cx| {
|
|
group.eose(cx);
|
|
});
|
|
|
|
cx.notify();
|
|
}
|
|
|
|
fn handle_closed(&mut self, id: SubscriptionId, reason: String, cx: &mut Context<Self>) {
|
|
if let Some(browse) = self.browse.get_mut(&id) {
|
|
log::warn!("nip29: browsing {} was refused: {reason}", browse.relay);
|
|
browse.loading = false;
|
|
|
|
cx.emit(GroupsEvent::Updated);
|
|
cx.notify();
|
|
return;
|
|
}
|
|
|
|
let Some(key) = self.routes.get(&id).cloned() else {
|
|
return;
|
|
};
|
|
|
|
let Some(group) = self.group(&key, cx) else {
|
|
return;
|
|
};
|
|
|
|
group.update(cx, |group, cx| {
|
|
group.refuse(reason, cx);
|
|
});
|
|
|
|
cx.notify();
|
|
}
|
|
|
|
fn handle_notifications(&mut self, cx: &mut Context<Self>) {
|
|
self.notification_listener = None;
|
|
self.signal_consumer = None;
|
|
|
|
let nostr = NostrRegistry::global(cx).read(cx);
|
|
let client = nostr.client();
|
|
|
|
let tx = self.signal_tx.clone();
|
|
let rx = self.signal_rx.clone();
|
|
|
|
self.notification_listener = Some(cx.background_spawn(async move {
|
|
let mut notifications = client.notifications();
|
|
let mut processed = HashSet::new();
|
|
const MAX_PROCESSED: usize = 10_000;
|
|
|
|
while let Some(notification) = notifications.next().await {
|
|
let ClientNotification::Message { message, .. } = notification else {
|
|
continue;
|
|
};
|
|
|
|
let signal = match *message {
|
|
RelayMessage::Event {
|
|
subscription_id,
|
|
event,
|
|
} => {
|
|
let id = subscription_id.into_owned();
|
|
|
|
if !is_own_subscription(&id) {
|
|
None
|
|
} else {
|
|
if processed.len() >= MAX_PROCESSED {
|
|
processed.clear();
|
|
}
|
|
processed
|
|
.insert(event.id)
|
|
.then(|| Signal::Event(id, event.into_owned()))
|
|
}
|
|
}
|
|
RelayMessage::EndOfStoredEvents(id) => {
|
|
let id = id.into_owned();
|
|
is_own_subscription(&id).then_some(Signal::Eose(id))
|
|
}
|
|
RelayMessage::Closed {
|
|
subscription_id,
|
|
message,
|
|
} => {
|
|
let id = subscription_id.into_owned();
|
|
is_own_subscription(&id).then_some(Signal::Closed(id, message.into_owned()))
|
|
}
|
|
_ => None,
|
|
};
|
|
|
|
if let Some(signal) = signal {
|
|
tx.send_async(signal).await.ok();
|
|
}
|
|
}
|
|
}));
|
|
|
|
self.signal_consumer = Some(cx.spawn(async move |this, cx| {
|
|
while let Ok(signal) = rx.recv_async().await {
|
|
this.update(cx, |this, cx| match signal {
|
|
Signal::Event(id, event) => {
|
|
this.handle_event(id, event, cx);
|
|
}
|
|
Signal::Eose(id) => {
|
|
this.handle_eose(id, cx);
|
|
}
|
|
Signal::Closed(id, reason) => {
|
|
this.handle_closed(id, reason, cx);
|
|
}
|
|
})
|
|
.ok();
|
|
}
|
|
}));
|
|
}
|
|
}
|