perspective_client/view.rs
1// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
2// ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃
3// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃
4// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃
5// ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃
6// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
7// ┃ Copyright (c) 2017, the Perspective Authors. ┃
8// ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
9// ┃ This file is part of the Perspective library, distributed under the terms ┃
10// ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
11// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
12
13use std::collections::HashMap;
14use std::ops::Deref;
15use std::str::FromStr;
16use std::sync::Arc;
17
18use futures::Future;
19use prost::bytes::Bytes;
20use serde::{Deserialize, Serialize};
21use ts_rs::TS;
22
23use self::view_on_update_req::Mode;
24use crate::assert_view_api;
25use crate::client::Client;
26use crate::proto::request::ClientReq;
27use crate::proto::response::ClientResp;
28use crate::proto::*;
29#[cfg(doc)]
30use crate::table::Table;
31pub use crate::utils::*;
32
33/// Options for [`View::on_update`].
34#[derive(Default, Debug, Deserialize, TS)]
35pub struct OnUpdateOptions {
36 pub mode: Option<OnUpdateMode>,
37}
38
39/// The update mode for [`View::on_update`].
40///
41/// `Row` mode calculates and provides the update batch new rows/columns as an
42/// Apache Arrow to the callback provided to [`View::on_update`]. This allows
43/// incremental updates if your callbakc can read this format, but should be
44/// disabled otherwise.
45#[derive(Default, Debug, Deserialize, TS)]
46pub enum OnUpdateMode {
47 #[default]
48 #[serde(rename = "row")]
49 Row,
50}
51
52impl FromStr for OnUpdateMode {
53 type Err = ClientError;
54
55 fn from_str(s: &str) -> Result<Self, Self::Err> {
56 if s == "row" {
57 Ok(OnUpdateMode::Row)
58 } else {
59 Err(ClientError::Option)
60 }
61 }
62}
63
64#[derive(Clone, Debug, Default, Deserialize, Serialize, TS, PartialEq)]
65pub struct ColumnWindow {
66 #[serde(skip_serializing_if = "Option::is_none")]
67 pub start_col: Option<f32>,
68
69 #[serde(skip_serializing_if = "Option::is_none")]
70 pub end_col: Option<f32>,
71}
72
73/// Options for serializing a window of data from a [`View`].
74///
75/// Some fields of [`ViewWindow`] are only applicable to specific methods of
76/// [`View`].
77#[derive(Clone, Debug, Default, Deserialize, Serialize, TS, PartialEq)]
78pub struct ViewWindow {
79 #[ts(optional)]
80 #[serde(skip_serializing_if = "Option::is_none")]
81 pub start_row: Option<f64>,
82
83 #[ts(optional)]
84 #[serde(skip_serializing_if = "Option::is_none")]
85 pub start_col: Option<f64>,
86
87 #[ts(optional)]
88 #[serde(skip_serializing_if = "Option::is_none")]
89 pub end_row: Option<f64>,
90
91 #[ts(optional)]
92 #[serde(skip_serializing_if = "Option::is_none")]
93 pub end_col: Option<f64>,
94
95 #[ts(optional)]
96 #[serde(skip_serializing_if = "Option::is_none")]
97 pub id: Option<bool>,
98
99 #[ts(optional)]
100 #[serde(skip_serializing_if = "Option::is_none")]
101 pub index: Option<bool>,
102
103 /// Only impacts [`View::to_csv`]
104 #[ts(optional)]
105 #[serde(skip_serializing_if = "Option::is_none")]
106 pub formatted: Option<bool>,
107
108 /// Arrow IPC body compression for [`View::to_arrow`], `"lz4"` or `"zstd"`
109 /// (uncompressed when omitted).
110 #[ts(optional)]
111 #[ts(type = "\"lz4\" | \"zstd\"")]
112 #[serde(skip_serializing_if = "Option::is_none")]
113 pub compression: Option<String>,
114
115 /// When `true`, [`View::to_arrow`] names `group_by` key columns
116 /// `__ROW_PATH_N__` and never qualifies an aggregate, instead of the
117 /// human-readable names a key column and its aggregate would otherwise
118 /// take.
119 #[ts(optional)]
120 #[serde(skip_serializing_if = "Option::is_none")]
121 pub machine_column_names: Option<bool>,
122}
123
124impl From<ViewWindow> for ViewPort {
125 fn from(window: ViewWindow) -> Self {
126 ViewPort {
127 start_row: window.start_row.map(|x| x.floor() as u32),
128 start_col: window.start_col.map(|x| x.floor() as u32),
129 end_row: window.end_row.map(|x| x.ceil() as u32),
130 end_col: window.end_col.map(|x| x.ceil() as u32),
131 machine_column_names: window.machine_column_names,
132 }
133 }
134}
135
136impl From<ViewPort> for ViewWindow {
137 fn from(window: ViewPort) -> Self {
138 ViewWindow {
139 start_row: window.start_row.map(|x| x as f64),
140 start_col: window.start_col.map(|x| x as f64),
141 end_row: window.end_row.map(|x| x as f64),
142 end_col: window.end_col.map(|x| x as f64),
143 machine_column_names: window.machine_column_names,
144 ..ViewWindow::default()
145 }
146 }
147}
148
149/// Rows updated and port ID corresponding to an update batch, provided to the
150/// callback argument to [`View::on_update`] with the "rows" mode.
151#[derive(TS)]
152pub struct OnUpdateData(crate::proto::ViewOnUpdateResp);
153
154impl Deref for OnUpdateData {
155 type Target = crate::proto::ViewOnUpdateResp;
156
157 fn deref(&self) -> &Self::Target {
158 &self.0
159 }
160}
161
162/// Removed index values and port ID corresponding to a remove batch, provided
163/// to the callback argument to [`View::on_remove`].
164#[derive(TS)]
165pub struct OnRemoveData(crate::proto::ViewOnRemoveResp);
166
167impl Deref for OnRemoveData {
168 type Target = crate::proto::ViewOnRemoveResp;
169
170 fn deref(&self) -> &Self::Target {
171 &self.0
172 }
173}
174/// The [`View`] struct is Perspective's query and serialization interface. It
175/// represents a query on the `Table`'s dataset and is always created from an
176/// existing `Table` instance via the [`Table::view`] method.
177///
178/// [`View`]s are immutable with respect to the arguments provided to the
179/// [`Table::view`] method; to change these parameters, you must create a new
180/// [`View`] on the same [`Table`]. However, each [`View`] is _live_ with
181/// respect to the [`Table`]'s data, and will (within a conflation window)
182/// update with the latest state as its parent [`Table`] updates, including
183/// incrementally recalculating all aggregates, pivots, filters, etc. [`View`]
184/// query parameters are composable, in that each parameter works independently
185/// _and_ in conjunction with each other, and there is no limit to the number of
186/// pivots, filters, etc. which can be applied.
187///
188/// To construct a [`View`], call the [`Table::view`] factory method. A
189/// [`Table`] can have as many [`View`]s associated with it as you need -
190/// Perspective conserves memory by relying on a single [`Table`] to power
191/// multiple [`View`]s concurrently.
192///
193/// # Examples
194///
195/// ```no_run
196/// # use perspective_client::{Client, TableData, TableInitOptions, UpdateData, ViewWindow};
197/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
198/// # let client: Client = todo!();
199/// let opts = TableInitOptions::default();
200/// let data = TableData::Update(UpdateData::Csv("x,y\n1,2\n3,4".into()));
201/// let table = client.table(data, opts).await?;
202///
203/// let view = table.view(None).await?;
204/// let arrow = view.to_arrow(ViewWindow::default()).await?;
205/// view.delete().await?;
206/// # Ok(()) }
207/// ```
208///
209/// ```no_run
210/// # use std::collections::HashMap;
211/// # use perspective_client::Table;
212/// # use perspective_client::config::*;
213/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
214/// # let table: Table = todo!();
215/// let view = table
216/// .view(Some(ViewConfigUpdate {
217/// columns: Some(vec![Some("Sales".into())]),
218/// aggregates: Some(HashMap::from_iter(vec![("Sales".into(), "sum".into())])),
219/// group_by: Some(vec!["Region".into(), "Country".into()]),
220/// filter: Some(vec![Filter::new("Category", "in", &[
221/// "Furniture",
222/// "Technology",
223/// ])]),
224/// ..ViewConfigUpdate::default()
225/// }))
226/// .await?;
227/// # Ok(()) }
228/// ```
229///
230/// Group By
231///
232/// ```no_run
233/// # use perspective_client::Table;
234/// # use perspective_client::config::*;
235/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
236/// # let table: Table = todo!();
237/// let view = table
238/// .view(Some(ViewConfigUpdate {
239/// group_by: Some(vec!["a".into(), "c".into()]),
240/// ..ViewConfigUpdate::default()
241/// }))
242/// .await?;
243/// # Ok(()) }
244/// ```
245///
246/// Split By
247///
248/// ```no_run
249/// # use perspective_client::Table;
250/// # use perspective_client::config::*;
251/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
252/// # let table: Table = todo!();
253/// let view = table
254/// .view(Some(ViewConfigUpdate {
255/// split_by: Some(vec!["a".into(), "c".into()]),
256/// ..ViewConfigUpdate::default()
257/// }))
258/// .await?;
259/// # Ok(()) }
260/// ```
261///
262/// In Javascript, a [`Table`] can be constructed on a [`Table::view`] instance,
263/// which will return a new [`Table`] based on the [`Table::view`]'s dataset,
264/// and all future updates that affect the [`Table::view`] will be forwarded to
265/// the new [`Table`]. This is particularly useful for implementing a
266/// [Client/Server Replicated](server.md#clientserver-replicated) design, by
267/// serializing the `View` to an arrow and setting up an `on_update` callback.
268///
269/// ```no_run
270/// # use perspective_client::{Client, TableData, TableInitOptions, UpdateData, UpdateOptions};
271/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
272/// # let client: Client = todo!();
273/// let opts = TableInitOptions::default();
274/// let data = TableData::Update(UpdateData::Csv("x,y\n1,2\n3,4".into()));
275/// let table = client.table(data, opts.clone()).await?;
276/// let view = table.view(None).await?;
277/// let table2 = client.table(TableData::View(view), opts).await?;
278/// let more = UpdateData::Csv("x,y\n5,6".into());
279/// table.update(more, UpdateOptions::default()).await?;
280/// # Ok(()) }
281/// ```
282#[derive(Clone, Debug)]
283pub struct View {
284 pub name: String,
285 client: Client,
286 pub(crate) source: Option<ViewSource>,
287}
288
289/// The options of the [`Table`] a [`View`] was created from, as known to the
290/// [`Client`] which created it; a [`View`] opened by name alone has no source.
291#[derive(Clone, Debug)]
292pub(crate) struct ViewSource {
293 pub options: crate::table::TableOptions,
294}
295
296assert_view_api!(View);
297
298impl View {
299 pub(crate) fn client(&self) -> &Client {
300 &self.client
301 }
302
303 pub fn new(name: String, client: Client) -> Self {
304 View {
305 name,
306 client,
307 source: None,
308 }
309 }
310
311 pub(crate) fn new_with_source(name: String, client: Client, source: ViewSource) -> Self {
312 View {
313 name,
314 client,
315 source: Some(source),
316 }
317 }
318
319 fn client_message(&self, req: ClientReq) -> Request {
320 crate::proto::Request {
321 msg_id: self.client.gen_id(),
322 entity_id: self.name.clone(),
323 client_req: Some(req),
324 }
325 }
326
327 /// Returns this [`View`]'s column header area for `window`, transposed as
328 /// `area[level][column]`.
329 ///
330 /// A column path shows the columns that a given cell belongs to after
331 /// pivots are applied. The area is rectangular with one level per
332 /// `split_by` plus one: a column's split values occupy the leading levels
333 /// and its name always occupies the last, so a subtotal or grand total
334 /// under `split_rollup_mode: "rollup"` reads [`Scalar::Null`] at the levels
335 /// it does not pivot on. Values keep their column's type - a `datetime`
336 /// split value is epoch milliseconds, not text - so formatting them is the
337 /// caller's choice.
338 ///
339 /// `window` slices the *column* axis; the number of levels does not depend
340 /// on it.
341 pub async fn column_paths(
342 &self,
343 window: ColumnWindow,
344 ) -> ClientResult<Vec<Vec<crate::config::Scalar>>> {
345 let msg = self.client_message(ClientReq::ViewColumnPathsReq(ViewColumnPathsReq {
346 start_col: window.start_col.map(|x| x as u32),
347 end_col: window.end_col.map(|x| x as u32),
348 }));
349
350 match self.client.oneshot(&msg).await? {
351 ClientResp::ViewColumnPathsResp(ViewColumnPathsResp { area }) => Ok(area
352 .into_iter()
353 .map(|level| {
354 level
355 .values
356 .into_iter()
357 .map(crate::config::Scalar::from)
358 .collect()
359 })
360 .collect()),
361 resp => Err(resp.into()),
362 }
363 }
364
365 /// Returns this [`View`]'s _dimensions_, row and column count, as well as
366 /// those of the [`crate::Table`] from which it was derived.
367 ///
368 /// - `num_table_rows` - The number of rows in the underlying
369 /// [`crate::Table`].
370 /// - `num_table_columns` - The number of columns in the underlying
371 /// [`crate::Table`] (including the `index` column if this
372 /// [`crate::Table`] was constructed with one).
373 /// - `num_view_rows` - The number of rows in this [`View`]. If this
374 /// [`View`] has a `group_by` clause, `num_view_rows` will also include
375 /// aggregated rows.
376 /// - `num_view_columns` - The number of columns in this [`View`]. If this
377 /// [`View`] has a `split_by` clause, `num_view_columns` will include all
378 /// _column paths_, e.g. the number of `columns` clause times the number
379 /// of `split_by` groups.
380 pub async fn dimensions(&self) -> ClientResult<ViewDimensionsResp> {
381 let msg = self.client_message(ClientReq::ViewDimensionsReq(ViewDimensionsReq {}));
382 match self.client.oneshot(&msg).await? {
383 ClientResp::ViewDimensionsResp(resp) => Ok(resp),
384 resp => Err(resp.into()),
385 }
386 }
387
388 /// The expression schema of this [`View`], which contains only the
389 /// expressions created on this [`View`]. See [`View::schema`] for
390 /// details.
391 pub async fn expression_schema(&self) -> ClientResult<HashMap<String, ColumnType>> {
392 if self.client.get_features().await?.expressions {
393 let msg = self.client_message(ClientReq::ViewExpressionSchemaReq(
394 ViewExpressionSchemaReq {},
395 ));
396 match self.client.oneshot(&msg).await? {
397 ClientResp::ViewExpressionSchemaResp(ViewExpressionSchemaResp { schema }) => {
398 Ok(schema
399 .into_iter()
400 .map(|(x, y)| (x, ColumnType::try_from(y).unwrap()))
401 .collect())
402 },
403 resp => Err(resp.into()),
404 }
405 } else {
406 Ok([].into_iter().collect())
407 }
408 }
409
410 /// A copy of the [`ViewConfig`] object passed to the [`Table::view`] method
411 /// which created this [`View`].
412 pub async fn get_config(&self) -> ClientResult<crate::config::ViewConfig> {
413 let msg = self.client_message(ClientReq::ViewGetConfigReq(ViewGetConfigReq {}));
414 match self.client.oneshot(&msg).await? {
415 ClientResp::ViewGetConfigResp(ViewGetConfigResp {
416 config: Some(config),
417 }) => Ok(config.into()),
418 resp => Err(resp.into()),
419 }
420 }
421
422 /// The number of aggregated rows in this [`View`]. This is affected by the
423 /// "group_by" configuration parameter supplied to this view's contructor.
424 ///
425 /// # Returns
426 ///
427 /// The number of aggregated rows.
428 pub async fn num_rows(&self) -> ClientResult<u32> {
429 Ok(self.dimensions().await?.num_view_rows)
430 }
431
432 /// The schema of this [`View`].
433 ///
434 /// The [`View`] schema differs from the `schema` returned by
435 /// [`Table::schema`]; it may have different column names due to
436 /// `expressions` or `columns` configs, or it maye have _different
437 /// column types_ due to the application og `group_by` and `aggregates`
438 /// config. You can think of [`Table::schema`] as the _input_ schema and
439 /// [`View::schema`] as the _output_ schema of a Perspective pipeline.
440 pub async fn schema(&self) -> ClientResult<HashMap<String, ColumnType>> {
441 let msg = self.client_message(ClientReq::ViewSchemaReq(ViewSchemaReq {}));
442 match self.client.oneshot(&msg).await? {
443 ClientResp::ViewSchemaResp(ViewSchemaResp { schema }) => Ok(schema
444 .into_iter()
445 .map(|(x, y)| (x, ColumnType::try_from(y).unwrap()))
446 .collect()),
447 resp => Err(resp.into()),
448 }
449 }
450
451 /// Serializes a [`View`] to the Apache Arrow data format, with IPC body
452 /// compression per [`ViewWindow::compression`].
453 pub async fn to_arrow(&self, window: ViewWindow) -> ClientResult<Bytes> {
454 let msg = self.client_message(ClientReq::ViewToArrowReq(ViewToArrowReq {
455 viewport: Some(window.clone().into()),
456 compression: window.compression,
457 }));
458
459 match self.client.oneshot(&msg).await? {
460 ClientResp::ViewToArrowResp(ViewToArrowResp { arrow }) => Ok(arrow.into()),
461 resp => Err(resp.into()),
462 }
463 }
464
465 /// Serializes this [`View`] to a string of JSON data. Useful if you want to
466 /// save additional round trip serialize/deserialize cycles.
467 pub async fn to_columns_string(&self, window: ViewWindow) -> ClientResult<String> {
468 let msg = self.client_message(ClientReq::ViewToColumnsStringReq(ViewToColumnsStringReq {
469 viewport: Some(window.clone().into()),
470 id: window.id,
471 index: window.index,
472 formatted: window.formatted,
473 }));
474
475 match self.client.oneshot(&msg).await? {
476 ClientResp::ViewToColumnsStringResp(ViewToColumnsStringResp { json_string }) => {
477 Ok(json_string)
478 },
479 resp => Err(resp.into()),
480 }
481 }
482
483 /// Render this `View` as a JSON string.
484 pub async fn to_json_string(&self, window: ViewWindow) -> ClientResult<String> {
485 let viewport = ViewPort::from(window.clone());
486 let msg = self.client_message(ClientReq::ViewToRowsStringReq(ViewToRowsStringReq {
487 viewport: Some(viewport),
488 id: window.id,
489 index: window.index,
490 formatted: window.formatted,
491 }));
492
493 match self.client.oneshot(&msg).await? {
494 ClientResp::ViewToRowsStringResp(ViewToRowsStringResp { json_string }) => {
495 Ok(json_string)
496 },
497 resp => Err(resp.into()),
498 }
499 }
500
501 /// Renders this [`View`] as an [NDJSON](https://github.com/ndjson/ndjson-spec)
502 /// formatted [`String`].
503 pub async fn to_ndjson(&self, window: ViewWindow) -> ClientResult<String> {
504 let viewport = ViewPort::from(window.clone());
505 let msg = self.client_message(ClientReq::ViewToNdjsonStringReq(ViewToNdjsonStringReq {
506 viewport: Some(viewport),
507 id: window.id,
508 index: window.index,
509 formatted: window.formatted,
510 }));
511
512 match self.client.oneshot(&msg).await? {
513 ClientResp::ViewToNdjsonStringResp(ViewToNdjsonStringResp { ndjson_string }) => {
514 Ok(ndjson_string)
515 },
516 resp => Err(resp.into()),
517 }
518 }
519
520 /// Serializes this [`View`] to CSV data in a standard format.
521 pub async fn to_csv(&self, window: ViewWindow) -> ClientResult<String> {
522 let msg = self.client_message(ClientReq::ViewToCsvReq(ViewToCsvReq {
523 viewport: Some(window.into()),
524 }));
525
526 match self.client.oneshot(&msg).await? {
527 ClientResp::ViewToCsvResp(ViewToCsvResp { csv }) => Ok(csv),
528 resp => Err(resp.into()),
529 }
530 }
531
532 /// Delete this [`View`] and clean up all resources associated with it.
533 /// [`View`] objects do not stop consuming resources or processing
534 /// updates when they are garbage collected - you must call this method
535 /// to reclaim these.
536 pub async fn delete(&self) -> ClientResult<()> {
537 let msg = self.client_message(ClientReq::ViewDeleteReq(ViewDeleteReq {}));
538 match self.client.oneshot(&msg).await? {
539 ClientResp::ViewDeleteResp(_) => Ok(()),
540 resp => Err(resp.into()),
541 }
542 }
543
544 /// Calculates the [min, max] of the leaf nodes of a column `column_name`.
545 ///
546 /// # Returns
547 ///
548 /// A tuple of [min, max], whose types are column and aggregate dependent.
549 pub async fn get_min_max(
550 &self,
551 column_name: String,
552 ) -> ClientResult<(crate::config::Scalar, crate::config::Scalar)> {
553 let msg = self.client_message(ClientReq::ViewGetMinMaxReq(ViewGetMinMaxReq {
554 column_name,
555 }));
556
557 match self.client.oneshot(&msg).await? {
558 ClientResp::ViewGetMinMaxResp(ViewGetMinMaxResp { min, max }) => {
559 let min = min.map(crate::config::Scalar::from).unwrap_or_default();
560 let max = max.map(crate::config::Scalar::from).unwrap_or_default();
561 Ok((min, max))
562 },
563 resp => Err(resp.into()),
564 }
565 }
566
567 /// Register a callback with this [`View`]. Whenever the view's underlying
568 /// table emits an update, this callback will be invoked with an object
569 /// containing `port_id`, indicating which port the update fired on, and
570 /// optionally `delta`, which is the new data that was updated for each
571 /// cell or each row.
572 ///
573 /// # Arguments
574 ///
575 /// - `on_update` - A callback function invoked on update, which receives an
576 /// object with two keys: `port_id`, indicating which port the update was
577 /// triggered on, and `delta`, whose value is dependent on the mode
578 /// parameter.
579 /// - `options` - If this is provided as `OnUpdateOptions { mode:
580 /// Some(OnUpdateMode::Row) }`, then `delta` is an Arrow of the updated
581 /// rows. Otherwise `delta` will be [`Option::None`].
582 pub async fn on_update<T, U>(&self, on_update: T, options: OnUpdateOptions) -> ClientResult<u32>
583 where
584 T: Fn(OnUpdateData) -> U + Send + Sync + 'static,
585 U: Future<Output = ()> + Send + 'static,
586 {
587 let on_update = Arc::new(on_update);
588 let callback = move |resp: Response| {
589 let on_update = on_update.clone();
590 async move {
591 match resp.client_resp {
592 Some(ClientResp::ViewOnUpdateResp(resp)) => {
593 on_update(OnUpdateData(resp)).await;
594 Ok(())
595 },
596 resp => Err(resp.into()),
597 }
598 }
599 };
600
601 let msg = self.client_message(ClientReq::ViewOnUpdateReq(ViewOnUpdateReq {
602 mode: options.mode.map(|OnUpdateMode::Row| Mode::Row as i32),
603 }));
604
605 self.client.subscribe(&msg, callback).await?;
606 Ok(msg.msg_id)
607 }
608
609 /// Unregister a previously registered update callback with this [`View`].
610 ///
611 /// # Arguments
612 ///
613 /// - `id` - A callback `id` as returned by a recipricol call to
614 /// [`View::on_update`].
615 ///
616 /// # Examples
617 ///
618 /// ```no_run
619 /// # use perspective_client::{OnUpdateOptions, View};
620 /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
621 /// # let view: View = todo!();
622 /// let callback = |_| async { print!("Updated!") };
623 /// let cid = view.on_update(callback, OnUpdateOptions::default()).await?;
624 /// view.remove_update(cid).await?;
625 /// # Ok(()) }
626 /// ```
627 pub async fn remove_update(&self, update_id: u32) -> ClientResult<()> {
628 let msg = self.client_message(ClientReq::ViewRemoveOnUpdateReq(ViewRemoveOnUpdateReq {
629 id: update_id,
630 }));
631
632 self.client.unsubscribe(update_id).await?;
633 match self.client.oneshot(&msg).await? {
634 ClientResp::ViewRemoveOnUpdateResp(_) => Ok(()),
635 resp => Err(resp.into()),
636 }
637 }
638
639 /// Register a callback which is invoked whenever rows are removed from
640 /// this [`View`]'s [`Table`] by [`Table::remove`], with the removed `index`
641 /// column values as an Apache Arrow of one column named after the index.
642 ///
643 /// [`Table::replace`] reports the keys it does not re-supply and
644 /// [`Table::clear`] reports every key. It never fires for a
645 /// [`Table`] without an `index`.
646 ///
647 /// # Examples
648 ///
649 /// ```no_run
650 /// # use perspective_client::{View, OnRemoveData};
651 /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
652 /// # let view: View = todo!();
653 /// let callback = |removed: OnRemoveData| async move { println!("{:?}", removed.port_id) };
654 /// let cid = view.on_remove(callback).await?;
655 /// view.remove_remove(cid).await?;
656 /// # Ok(()) }
657 /// ```
658 pub async fn on_remove<T, U>(&self, on_remove: T) -> ClientResult<u32>
659 where
660 T: Fn(OnRemoveData) -> U + Send + Sync + 'static,
661 U: Future<Output = ()> + Send + 'static,
662 {
663 let on_remove = Arc::new(on_remove);
664 let callback = move |resp: Response| {
665 let on_remove = on_remove.clone();
666 async move {
667 match resp.client_resp {
668 Some(ClientResp::ViewOnRemoveResp(resp)) => {
669 on_remove(OnRemoveData(resp)).await;
670 Ok(())
671 },
672 resp => Err(resp.into()),
673 }
674 }
675 };
676
677 let msg = self.client_message(ClientReq::ViewOnRemoveReq(ViewOnRemoveReq {}));
678 self.client.subscribe(&msg, callback).await?;
679 Ok(msg.msg_id)
680 }
681
682 /// Unregister a previously registered [`View::on_remove`] callback.
683 ///
684 /// # Arguments
685 ///
686 /// - `callback_id` - A callback `id` as returned by a reciprocal call to
687 /// [`View::on_remove`].
688 pub async fn remove_remove(&self, callback_id: u32) -> ClientResult<()> {
689 let msg = self.client_message(ClientReq::ViewRemoveOnRemoveReq(ViewRemoveOnRemoveReq {
690 id: callback_id,
691 }));
692
693 self.client.unsubscribe(callback_id).await?;
694 match self.client.oneshot(&msg).await? {
695 ClientResp::ViewRemoveOnRemoveResp(_) => Ok(()),
696 resp => Err(resp.into()),
697 }
698 }
699
700 /// Register a callback with this [`View`]. Whenever the [`View`] is
701 /// deleted, this callback will be invoked.
702 pub async fn on_delete(
703 &self,
704 on_delete: Box<dyn Fn() + Send + Sync + 'static>,
705 ) -> ClientResult<u32> {
706 let callback = move |resp: Response| match resp.client_resp.unwrap() {
707 ClientResp::ViewOnDeleteResp(_) => {
708 on_delete();
709 Ok(())
710 },
711 resp => Err(resp.into()),
712 };
713
714 let msg = self.client_message(ClientReq::ViewOnDeleteReq(ViewOnDeleteReq {}));
715 self.client.subscribe_once(&msg, Box::new(callback)).await?;
716 Ok(msg.msg_id)
717 }
718
719 /// Unregister a previously registered [`View::on_delete`] callback.
720 pub async fn remove_delete(&self, callback_id: u32) -> ClientResult<()> {
721 let msg = self.client_message(ClientReq::ViewRemoveDeleteReq(ViewRemoveDeleteReq {
722 id: callback_id,
723 }));
724
725 match self.client.oneshot(&msg).await? {
726 ClientResp::ViewRemoveDeleteResp(ViewRemoveDeleteResp {}) => Ok(()),
727 resp => Err(resp.into()),
728 }
729 }
730
731 /// Collapses the `group_by` row at `row_index`.
732 pub async fn collapse(&self, row_index: u32) -> ClientResult<u32> {
733 let msg = self.client_message(ClientReq::ViewCollapseReq(ViewCollapseReq { row_index }));
734 match self.client.oneshot(&msg).await? {
735 ClientResp::ViewCollapseResp(ViewCollapseResp { num_changed }) => Ok(num_changed),
736 resp => Err(resp.into()),
737 }
738 }
739
740 /// Expand the `group_by` row at `row_index`.
741 pub async fn expand(&self, row_index: u32) -> ClientResult<u32> {
742 let msg = self.client_message(ClientReq::ViewExpandReq(ViewExpandReq { row_index }));
743 match self.client.oneshot(&msg).await? {
744 ClientResp::ViewExpandResp(ViewExpandResp { num_changed }) => Ok(num_changed),
745 resp => Err(resp.into()),
746 }
747 }
748
749 /// Set expansion `depth` of the `group_by` tree.
750 pub async fn set_depth(&self, depth: u32) -> ClientResult<()> {
751 let msg = self.client_message(ClientReq::ViewSetDepthReq(ViewSetDepthReq { depth }));
752 match self.client.oneshot(&msg).await? {
753 ClientResp::ViewSetDepthResp(_) => Ok(()),
754 resp => Err(resp.into()),
755 }
756 }
757}