840 lines
29 KiB
Rust
840 lines
29 KiB
Rust
mod channel_index;
|
|
|
|
use crate::{channel_buffer::ChannelBuffer, channel_chat::ChannelChat};
|
|
use anyhow::{anyhow, Result};
|
|
use client::{Client, Subscription, User, UserId, UserStore};
|
|
use collections::{hash_map, HashMap, HashSet};
|
|
use futures::{channel::mpsc, future::Shared, Future, FutureExt, StreamExt};
|
|
use gpui::{AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle, Task, WeakModelHandle};
|
|
use rpc::{proto, TypedEnvelope};
|
|
use std::{mem, sync::Arc, time::Duration};
|
|
use util::ResultExt;
|
|
|
|
use self::channel_index::ChannelIndex;
|
|
pub use self::channel_index::ChannelPath;
|
|
|
|
pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30);
|
|
|
|
pub type ChannelId = u64;
|
|
|
|
pub struct ChannelStore {
|
|
channel_index: ChannelIndex,
|
|
channel_invitations: Vec<Arc<Channel>>,
|
|
channel_participants: HashMap<ChannelId, Vec<Arc<User>>>,
|
|
channels_with_admin_privileges: HashSet<ChannelId>,
|
|
outgoing_invites: HashSet<(ChannelId, UserId)>,
|
|
update_channels_tx: mpsc::UnboundedSender<proto::UpdateChannels>,
|
|
opened_buffers: HashMap<ChannelId, OpenedModelHandle<ChannelBuffer>>,
|
|
opened_chats: HashMap<ChannelId, OpenedModelHandle<ChannelChat>>,
|
|
client: Arc<Client>,
|
|
user_store: ModelHandle<UserStore>,
|
|
_rpc_subscription: Subscription,
|
|
_watch_connection_status: Task<Option<()>>,
|
|
disconnect_channel_buffers_task: Option<Task<()>>,
|
|
_update_channels: Task<()>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
pub struct Channel {
|
|
pub id: ChannelId,
|
|
pub name: String,
|
|
}
|
|
|
|
pub struct ChannelMembership {
|
|
pub user: Arc<User>,
|
|
pub kind: proto::channel_member::Kind,
|
|
pub admin: bool,
|
|
}
|
|
|
|
pub enum ChannelEvent {
|
|
ChannelCreated(ChannelId),
|
|
ChannelRenamed(ChannelId),
|
|
}
|
|
|
|
impl Entity for ChannelStore {
|
|
type Event = ChannelEvent;
|
|
}
|
|
|
|
enum OpenedModelHandle<E: Entity> {
|
|
Open(WeakModelHandle<E>),
|
|
Loading(Shared<Task<Result<ModelHandle<E>, Arc<anyhow::Error>>>>),
|
|
}
|
|
|
|
impl ChannelStore {
|
|
pub fn new(
|
|
client: Arc<Client>,
|
|
user_store: ModelHandle<UserStore>,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Self {
|
|
let rpc_subscription =
|
|
client.add_message_handler(cx.handle(), Self::handle_update_channels);
|
|
|
|
let mut connection_status = client.status();
|
|
let (update_channels_tx, mut update_channels_rx) = mpsc::unbounded();
|
|
let watch_connection_status = cx.spawn_weak(|this, mut cx| async move {
|
|
while let Some(status) = connection_status.next().await {
|
|
let this = this.upgrade(&cx)?;
|
|
if status.is_connected() {
|
|
this.update(&mut cx, |this, cx| this.handle_connect(cx))
|
|
.await
|
|
.log_err()?;
|
|
} else {
|
|
this.update(&mut cx, |this, cx| this.handle_disconnect(cx));
|
|
}
|
|
}
|
|
Some(())
|
|
});
|
|
|
|
Self {
|
|
channel_invitations: Vec::default(),
|
|
channel_index: ChannelIndex::default(),
|
|
channel_participants: Default::default(),
|
|
channels_with_admin_privileges: Default::default(),
|
|
outgoing_invites: Default::default(),
|
|
opened_buffers: Default::default(),
|
|
opened_chats: Default::default(),
|
|
update_channels_tx,
|
|
client,
|
|
user_store,
|
|
_rpc_subscription: rpc_subscription,
|
|
_watch_connection_status: watch_connection_status,
|
|
disconnect_channel_buffers_task: None,
|
|
_update_channels: cx.spawn_weak(|this, mut cx| async move {
|
|
while let Some(update_channels) = update_channels_rx.next().await {
|
|
if let Some(this) = this.upgrade(&cx) {
|
|
let update_task = this.update(&mut cx, |this, cx| {
|
|
this.update_channels(update_channels, cx)
|
|
});
|
|
if let Some(update_task) = update_task {
|
|
update_task.await.log_err();
|
|
}
|
|
}
|
|
}
|
|
}),
|
|
}
|
|
}
|
|
|
|
pub fn client(&self) -> Arc<Client> {
|
|
self.client.clone()
|
|
}
|
|
|
|
pub fn has_children(&self, channel_id: ChannelId) -> bool {
|
|
self.channel_index.iter().any(|path| {
|
|
if let Some(ix) = path.iter().position(|id| *id == channel_id) {
|
|
path.len() > ix + 1
|
|
} else {
|
|
false
|
|
}
|
|
})
|
|
}
|
|
|
|
pub fn channel_count(&self) -> usize {
|
|
self.channel_index.len()
|
|
}
|
|
|
|
pub fn index_of_channel(&self, channel_id: ChannelId) -> Option<usize> {
|
|
self.channel_index
|
|
.iter()
|
|
.position(|path| path.ends_with(&[channel_id]))
|
|
}
|
|
|
|
pub fn channels(&self) -> impl '_ + Iterator<Item = (usize, &Arc<Channel>)> {
|
|
self.channel_index.iter().map(move |path| {
|
|
let id = path.last().unwrap();
|
|
let channel = self.channel_for_id(*id).unwrap();
|
|
(path.len() - 1, channel)
|
|
})
|
|
}
|
|
|
|
pub fn channel_at_index(&self, ix: usize) -> Option<(&Arc<Channel>, &ChannelPath)> {
|
|
let path = self.channel_index.get(ix)?;
|
|
let id = path.last().unwrap();
|
|
let channel = self.channel_for_id(*id).unwrap();
|
|
|
|
Some((channel, path))
|
|
}
|
|
|
|
pub fn channel_invitations(&self) -> &[Arc<Channel>] {
|
|
&self.channel_invitations
|
|
}
|
|
|
|
pub fn channel_for_id(&self, channel_id: ChannelId) -> Option<&Arc<Channel>> {
|
|
self.channel_index.by_id().get(&channel_id)
|
|
}
|
|
|
|
pub fn has_open_channel_buffer(&self, channel_id: ChannelId, cx: &AppContext) -> bool {
|
|
if let Some(buffer) = self.opened_buffers.get(&channel_id) {
|
|
if let OpenedModelHandle::Open(buffer) = buffer {
|
|
return buffer.upgrade(cx).is_some();
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
pub fn open_channel_buffer(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<ModelHandle<ChannelBuffer>>> {
|
|
let client = self.client.clone();
|
|
self.open_channel_resource(
|
|
channel_id,
|
|
|this| &mut this.opened_buffers,
|
|
|channel, cx| ChannelBuffer::new(channel, client, cx),
|
|
cx,
|
|
)
|
|
}
|
|
|
|
pub fn open_channel_chat(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<ModelHandle<ChannelChat>>> {
|
|
let client = self.client.clone();
|
|
let user_store = self.user_store.clone();
|
|
self.open_channel_resource(
|
|
channel_id,
|
|
|this| &mut this.opened_chats,
|
|
|channel, cx| ChannelChat::new(channel, user_store, client, cx),
|
|
cx,
|
|
)
|
|
}
|
|
|
|
/// Asynchronously open a given resource associated with a channel.
|
|
///
|
|
/// Make sure that the resource is only opened once, even if this method
|
|
/// is called multiple times with the same channel id while the first task
|
|
/// is still running.
|
|
fn open_channel_resource<T: Entity, F, Fut>(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
get_map: fn(&mut Self) -> &mut HashMap<ChannelId, OpenedModelHandle<T>>,
|
|
load: F,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<ModelHandle<T>>>
|
|
where
|
|
F: 'static + FnOnce(Arc<Channel>, AsyncAppContext) -> Fut,
|
|
Fut: Future<Output = Result<ModelHandle<T>>>,
|
|
{
|
|
let task = loop {
|
|
match get_map(self).entry(channel_id) {
|
|
hash_map::Entry::Occupied(e) => match e.get() {
|
|
OpenedModelHandle::Open(model) => {
|
|
if let Some(model) = model.upgrade(cx) {
|
|
break Task::ready(Ok(model)).shared();
|
|
} else {
|
|
get_map(self).remove(&channel_id);
|
|
continue;
|
|
}
|
|
}
|
|
OpenedModelHandle::Loading(task) => {
|
|
break task.clone();
|
|
}
|
|
},
|
|
hash_map::Entry::Vacant(e) => {
|
|
let task = cx
|
|
.spawn(|this, cx| async move {
|
|
let channel = this.read_with(&cx, |this, _| {
|
|
this.channel_for_id(channel_id).cloned().ok_or_else(|| {
|
|
Arc::new(anyhow!("no channel for id: {}", channel_id))
|
|
})
|
|
})?;
|
|
|
|
load(channel, cx).await.map_err(Arc::new)
|
|
})
|
|
.shared();
|
|
|
|
e.insert(OpenedModelHandle::Loading(task.clone()));
|
|
cx.spawn({
|
|
let task = task.clone();
|
|
|this, mut cx| async move {
|
|
let result = task.await;
|
|
this.update(&mut cx, |this, _| match result {
|
|
Ok(model) => {
|
|
get_map(this).insert(
|
|
channel_id,
|
|
OpenedModelHandle::Open(model.downgrade()),
|
|
);
|
|
}
|
|
Err(_) => {
|
|
get_map(this).remove(&channel_id);
|
|
}
|
|
});
|
|
}
|
|
})
|
|
.detach();
|
|
break task;
|
|
}
|
|
}
|
|
};
|
|
cx.foreground()
|
|
.spawn(async move { task.await.map_err(|error| anyhow!("{}", error)) })
|
|
}
|
|
|
|
pub fn is_user_admin(&self, channel_id: ChannelId) -> bool {
|
|
self.channel_index.iter().any(|path| {
|
|
if let Some(ix) = path.iter().position(|id| *id == channel_id) {
|
|
path[..=ix]
|
|
.iter()
|
|
.any(|id| self.channels_with_admin_privileges.contains(id))
|
|
} else {
|
|
false
|
|
}
|
|
})
|
|
}
|
|
|
|
pub fn channel_participants(&self, channel_id: ChannelId) -> &[Arc<User>] {
|
|
self.channel_participants
|
|
.get(&channel_id)
|
|
.map_or(&[], |v| v.as_slice())
|
|
}
|
|
|
|
pub fn create_channel(
|
|
&self,
|
|
name: &str,
|
|
parent_id: Option<ChannelId>,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<ChannelId>> {
|
|
let client = self.client.clone();
|
|
let name = name.trim_start_matches("#").to_owned();
|
|
cx.spawn(|this, mut cx| async move {
|
|
let channel = client
|
|
.request(proto::CreateChannel { name, parent_id })
|
|
.await?
|
|
.channel
|
|
.ok_or_else(|| anyhow!("missing channel in response"))?;
|
|
|
|
let channel_id = channel.id;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
let task = this.update_channels(
|
|
proto::UpdateChannels {
|
|
channels: vec![channel],
|
|
..Default::default()
|
|
},
|
|
cx,
|
|
);
|
|
assert!(task.is_none());
|
|
|
|
// This event is emitted because the collab panel wants to clear the pending edit state
|
|
// before this frame is rendered. But we can't guarantee that the collab panel's future
|
|
// will resolve before this flush_effects finishes. Synchronously emitting this event
|
|
// ensures that the collab panel will observe this creation before the frame completes
|
|
cx.emit(ChannelEvent::ChannelCreated(channel_id));
|
|
});
|
|
|
|
Ok(channel_id)
|
|
})
|
|
}
|
|
|
|
pub fn link_channel(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
to: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
let client = self.client.clone();
|
|
cx.spawn(|_, _| async move {
|
|
let _ = client
|
|
.request(proto::LinkChannel { channel_id, to })
|
|
.await?;
|
|
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn unlink_channel(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
from: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
let client = self.client.clone();
|
|
cx.spawn(|_, _| async move {
|
|
let _ = client
|
|
.request(proto::UnlinkChannel { channel_id, from })
|
|
.await?;
|
|
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn move_channel(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
from: ChannelId,
|
|
to: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
let client = self.client.clone();
|
|
cx.spawn(|_, _| async move {
|
|
let _ = client
|
|
.request(proto::MoveChannel {
|
|
channel_id,
|
|
from,
|
|
to,
|
|
})
|
|
.await?;
|
|
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn invite_member(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
user_id: UserId,
|
|
admin: bool,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
if !self.outgoing_invites.insert((channel_id, user_id)) {
|
|
return Task::ready(Err(anyhow!("invite request already in progress")));
|
|
}
|
|
|
|
cx.notify();
|
|
let client = self.client.clone();
|
|
cx.spawn(|this, mut cx| async move {
|
|
let result = client
|
|
.request(proto::InviteChannelMember {
|
|
channel_id,
|
|
user_id,
|
|
admin,
|
|
})
|
|
.await;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
this.outgoing_invites.remove(&(channel_id, user_id));
|
|
cx.notify();
|
|
});
|
|
|
|
result?;
|
|
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn remove_member(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
user_id: u64,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
if !self.outgoing_invites.insert((channel_id, user_id)) {
|
|
return Task::ready(Err(anyhow!("invite request already in progress")));
|
|
}
|
|
|
|
cx.notify();
|
|
let client = self.client.clone();
|
|
cx.spawn(|this, mut cx| async move {
|
|
let result = client
|
|
.request(proto::RemoveChannelMember {
|
|
channel_id,
|
|
user_id,
|
|
})
|
|
.await;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
this.outgoing_invites.remove(&(channel_id, user_id));
|
|
cx.notify();
|
|
});
|
|
result?;
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn set_member_admin(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
user_id: UserId,
|
|
admin: bool,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
if !self.outgoing_invites.insert((channel_id, user_id)) {
|
|
return Task::ready(Err(anyhow!("member request already in progress")));
|
|
}
|
|
|
|
cx.notify();
|
|
let client = self.client.clone();
|
|
cx.spawn(|this, mut cx| async move {
|
|
let result = client
|
|
.request(proto::SetChannelMemberAdmin {
|
|
channel_id,
|
|
user_id,
|
|
admin,
|
|
})
|
|
.await;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
this.outgoing_invites.remove(&(channel_id, user_id));
|
|
cx.notify();
|
|
});
|
|
|
|
result?;
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn rename(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
new_name: &str,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<()>> {
|
|
let client = self.client.clone();
|
|
let name = new_name.to_string();
|
|
cx.spawn(|this, mut cx| async move {
|
|
let channel = client
|
|
.request(proto::RenameChannel { channel_id, name })
|
|
.await?
|
|
.channel
|
|
.ok_or_else(|| anyhow!("missing channel in response"))?;
|
|
this.update(&mut cx, |this, cx| {
|
|
let task = this.update_channels(
|
|
proto::UpdateChannels {
|
|
channels: vec![channel],
|
|
..Default::default()
|
|
},
|
|
cx,
|
|
);
|
|
assert!(task.is_none());
|
|
|
|
// This event is emitted because the collab panel wants to clear the pending edit state
|
|
// before this frame is rendered. But we can't guarantee that the collab panel's future
|
|
// will resolve before this flush_effects finishes. Synchronously emitting this event
|
|
// ensures that the collab panel will observe this creation before the frame complete
|
|
cx.emit(ChannelEvent::ChannelRenamed(channel_id))
|
|
});
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
pub fn respond_to_channel_invite(
|
|
&mut self,
|
|
channel_id: ChannelId,
|
|
accept: bool,
|
|
) -> impl Future<Output = Result<()>> {
|
|
let client = self.client.clone();
|
|
async move {
|
|
client
|
|
.request(proto::RespondToChannelInvite { channel_id, accept })
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
pub fn get_channel_member_details(
|
|
&self,
|
|
channel_id: ChannelId,
|
|
cx: &mut ModelContext<Self>,
|
|
) -> Task<Result<Vec<ChannelMembership>>> {
|
|
let client = self.client.clone();
|
|
let user_store = self.user_store.downgrade();
|
|
cx.spawn(|_, mut cx| async move {
|
|
let response = client
|
|
.request(proto::GetChannelMembers { channel_id })
|
|
.await?;
|
|
|
|
let user_ids = response.members.iter().map(|m| m.user_id).collect();
|
|
let user_store = user_store
|
|
.upgrade(&cx)
|
|
.ok_or_else(|| anyhow!("user store dropped"))?;
|
|
let users = user_store
|
|
.update(&mut cx, |user_store, cx| user_store.get_users(user_ids, cx))
|
|
.await?;
|
|
|
|
Ok(users
|
|
.into_iter()
|
|
.zip(response.members)
|
|
.filter_map(|(user, member)| {
|
|
Some(ChannelMembership {
|
|
user,
|
|
admin: member.admin,
|
|
kind: proto::channel_member::Kind::from_i32(member.kind)?,
|
|
})
|
|
})
|
|
.collect())
|
|
})
|
|
}
|
|
|
|
pub fn remove_channel(&self, channel_id: ChannelId) -> impl Future<Output = Result<()>> {
|
|
let client = self.client.clone();
|
|
async move {
|
|
client.request(proto::DeleteChannel { channel_id }).await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
pub fn has_pending_channel_invite_response(&self, _: &Arc<Channel>) -> bool {
|
|
false
|
|
}
|
|
|
|
pub fn has_pending_channel_invite(&self, channel_id: ChannelId, user_id: UserId) -> bool {
|
|
self.outgoing_invites.contains(&(channel_id, user_id))
|
|
}
|
|
|
|
async fn handle_update_channels(
|
|
this: ModelHandle<Self>,
|
|
message: TypedEnvelope<proto::UpdateChannels>,
|
|
_: Arc<Client>,
|
|
mut cx: AsyncAppContext,
|
|
) -> Result<()> {
|
|
this.update(&mut cx, |this, _| {
|
|
this.update_channels_tx
|
|
.unbounded_send(message.payload)
|
|
.unwrap();
|
|
});
|
|
Ok(())
|
|
}
|
|
|
|
fn handle_connect(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
|
|
self.disconnect_channel_buffers_task.take();
|
|
|
|
for chat in self.opened_chats.values() {
|
|
if let OpenedModelHandle::Open(chat) = chat {
|
|
if let Some(chat) = chat.upgrade(cx) {
|
|
chat.update(cx, |chat, cx| {
|
|
chat.rejoin(cx);
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut buffer_versions = Vec::new();
|
|
for buffer in self.opened_buffers.values() {
|
|
if let OpenedModelHandle::Open(buffer) = buffer {
|
|
if let Some(buffer) = buffer.upgrade(cx) {
|
|
let channel_buffer = buffer.read(cx);
|
|
let buffer = channel_buffer.buffer().read(cx);
|
|
buffer_versions.push(proto::ChannelBufferVersion {
|
|
channel_id: channel_buffer.channel().id,
|
|
epoch: channel_buffer.epoch(),
|
|
version: language::proto::serialize_version(&buffer.version()),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
if buffer_versions.is_empty() {
|
|
return Task::ready(Ok(()));
|
|
}
|
|
|
|
let response = self.client.request(proto::RejoinChannelBuffers {
|
|
buffers: buffer_versions,
|
|
});
|
|
|
|
cx.spawn(|this, mut cx| async move {
|
|
let mut response = response.await?;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
this.opened_buffers.retain(|_, buffer| match buffer {
|
|
OpenedModelHandle::Open(channel_buffer) => {
|
|
let Some(channel_buffer) = channel_buffer.upgrade(cx) else {
|
|
return false;
|
|
};
|
|
|
|
channel_buffer.update(cx, |channel_buffer, cx| {
|
|
let channel_id = channel_buffer.channel().id;
|
|
if let Some(remote_buffer) = response
|
|
.buffers
|
|
.iter_mut()
|
|
.find(|buffer| buffer.channel_id == channel_id)
|
|
{
|
|
let channel_id = channel_buffer.channel().id;
|
|
let remote_version =
|
|
language::proto::deserialize_version(&remote_buffer.version);
|
|
|
|
channel_buffer.replace_collaborators(
|
|
mem::take(&mut remote_buffer.collaborators),
|
|
cx,
|
|
);
|
|
|
|
let operations = channel_buffer
|
|
.buffer()
|
|
.update(cx, |buffer, cx| {
|
|
let outgoing_operations =
|
|
buffer.serialize_ops(Some(remote_version), cx);
|
|
let incoming_operations =
|
|
mem::take(&mut remote_buffer.operations)
|
|
.into_iter()
|
|
.map(language::proto::deserialize_operation)
|
|
.collect::<Result<Vec<_>>>()?;
|
|
buffer.apply_ops(incoming_operations, cx)?;
|
|
anyhow::Ok(outgoing_operations)
|
|
})
|
|
.log_err();
|
|
|
|
if let Some(operations) = operations {
|
|
let client = this.client.clone();
|
|
cx.background()
|
|
.spawn(async move {
|
|
let operations = operations.await;
|
|
for chunk in
|
|
language::proto::split_operations(operations)
|
|
{
|
|
client
|
|
.send(proto::UpdateChannelBuffer {
|
|
channel_id,
|
|
operations: chunk,
|
|
})
|
|
.ok();
|
|
}
|
|
})
|
|
.detach();
|
|
return true;
|
|
}
|
|
}
|
|
|
|
channel_buffer.disconnect(cx);
|
|
false
|
|
})
|
|
}
|
|
OpenedModelHandle::Loading(_) => true,
|
|
});
|
|
});
|
|
anyhow::Ok(())
|
|
})
|
|
}
|
|
|
|
fn handle_disconnect(&mut self, cx: &mut ModelContext<Self>) {
|
|
self.channel_index.clear();
|
|
self.channel_invitations.clear();
|
|
self.channel_participants.clear();
|
|
self.channels_with_admin_privileges.clear();
|
|
self.channel_index.clear();
|
|
self.outgoing_invites.clear();
|
|
cx.notify();
|
|
|
|
self.disconnect_channel_buffers_task.get_or_insert_with(|| {
|
|
cx.spawn_weak(|this, mut cx| async move {
|
|
cx.background().timer(RECONNECT_TIMEOUT).await;
|
|
if let Some(this) = this.upgrade(&cx) {
|
|
this.update(&mut cx, |this, cx| {
|
|
for (_, buffer) in this.opened_buffers.drain() {
|
|
if let OpenedModelHandle::Open(buffer) = buffer {
|
|
if let Some(buffer) = buffer.upgrade(cx) {
|
|
buffer.update(cx, |buffer, cx| buffer.disconnect(cx));
|
|
}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
})
|
|
});
|
|
}
|
|
|
|
pub(crate) fn update_channels(
|
|
&mut self,
|
|
payload: proto::UpdateChannels,
|
|
cx: &mut ModelContext<ChannelStore>,
|
|
) -> Option<Task<Result<()>>> {
|
|
if !payload.remove_channel_invitations.is_empty() {
|
|
self.channel_invitations
|
|
.retain(|channel| !payload.remove_channel_invitations.contains(&channel.id));
|
|
}
|
|
for channel in payload.channel_invitations {
|
|
match self
|
|
.channel_invitations
|
|
.binary_search_by_key(&channel.id, |c| c.id)
|
|
{
|
|
Ok(ix) => Arc::make_mut(&mut self.channel_invitations[ix]).name = channel.name,
|
|
Err(ix) => self.channel_invitations.insert(
|
|
ix,
|
|
Arc::new(Channel {
|
|
id: channel.id,
|
|
name: channel.name,
|
|
}),
|
|
),
|
|
}
|
|
}
|
|
|
|
let channels_changed = !payload.channels.is_empty() || !payload.delete_channels.is_empty() || !payload.delete_edge.is_empty();
|
|
if channels_changed {
|
|
if !payload.delete_channels.is_empty() {
|
|
self.channel_index.delete_channels(&payload.delete_channels);
|
|
self.channel_participants
|
|
.retain(|channel_id, _| !payload.delete_channels.contains(channel_id));
|
|
self.channels_with_admin_privileges
|
|
.retain(|channel_id| !payload.delete_channels.contains(channel_id));
|
|
|
|
for channel_id in &payload.delete_channels {
|
|
let channel_id = *channel_id;
|
|
if let Some(OpenedModelHandle::Open(buffer)) =
|
|
self.opened_buffers.remove(&channel_id)
|
|
{
|
|
if let Some(buffer) = buffer.upgrade(cx) {
|
|
buffer.update(cx, ChannelBuffer::disconnect);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut index_edit = self.channel_index.bulk_edit();
|
|
|
|
for channel in payload.channels {
|
|
index_edit.upsert(channel)
|
|
}
|
|
|
|
for edge in payload.delete_edge {
|
|
index_edit
|
|
.delete_edge(edge.parent_id, edge.channel_id);
|
|
}
|
|
}
|
|
|
|
|
|
|
|
for permission in payload.channel_permissions {
|
|
if permission.is_admin {
|
|
self.channels_with_admin_privileges
|
|
.insert(permission.channel_id);
|
|
} else {
|
|
self.channels_with_admin_privileges
|
|
.remove(&permission.channel_id);
|
|
}
|
|
}
|
|
|
|
cx.notify();
|
|
if payload.channel_participants.is_empty() {
|
|
return None;
|
|
}
|
|
|
|
let mut all_user_ids = Vec::new();
|
|
let channel_participants = payload.channel_participants;
|
|
for entry in &channel_participants {
|
|
for user_id in entry.participant_user_ids.iter() {
|
|
if let Err(ix) = all_user_ids.binary_search(user_id) {
|
|
all_user_ids.insert(ix, *user_id);
|
|
}
|
|
}
|
|
}
|
|
|
|
let users = self
|
|
.user_store
|
|
.update(cx, |user_store, cx| user_store.get_users(all_user_ids, cx));
|
|
Some(cx.spawn(|this, mut cx| async move {
|
|
let users = users.await?;
|
|
|
|
this.update(&mut cx, |this, cx| {
|
|
for entry in &channel_participants {
|
|
let mut participants: Vec<_> = entry
|
|
.participant_user_ids
|
|
.iter()
|
|
.filter_map(|user_id| {
|
|
users
|
|
.binary_search_by_key(&user_id, |user| &user.id)
|
|
.ok()
|
|
.map(|ix| users[ix].clone())
|
|
})
|
|
.collect();
|
|
|
|
participants.sort_by_key(|u| u.id);
|
|
|
|
this.channel_participants
|
|
.insert(entry.channel_id, participants);
|
|
}
|
|
|
|
cx.notify();
|
|
});
|
|
anyhow::Ok(())
|
|
}))
|
|
}
|
|
}
|