perspective_client/virtual_server/
server.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;
14
15use indexmap::{IndexMap, IndexSet};
16use prost::Message as ProstMessage;
17use prost::bytes::{Bytes, BytesMut};
18
19use super::data::RowPathStyle;
20use super::error::VirtualServerError;
21use super::generic_sql_model::{column_path_source, sort_column_paths};
22use super::handler::VirtualServerHandler;
23use crate::config::{GroupRollupMode, ViewConfig, ViewConfigUpdate};
24use crate::proto::response::ClientResp;
25use crate::proto::{
26    ColumnPathLevel, ColumnType, GetFeaturesResp, GetHostedTablesResp, HostedTable, MakeTableResp,
27    Request, Response, Scalar, ServerError, StatusCode, TableDeleteResp, TableDescribeResp,
28    TableMakePortResp, TableMakeViewResp, TableOnDeleteResp, TableRemoveDeleteResp,
29    TableSchemaResp, TableSizeResp, ViewCollapseResp, ViewColumnPathsResp, ViewDeleteResp,
30    ViewDescription, ViewDimensionsResp, ViewExpandResp, ViewExpressionSchemaResp,
31    ViewGetConfigResp, ViewGetMinMaxResp, ViewOnDeleteResp, ViewOnRemoveResp, ViewOnUpdateResp,
32    ViewRemoveDeleteResp, ViewRemoveOnRemoveResp, ViewRemoveOnUpdateResp, ViewSchemaResp,
33    ViewSetDepthResp, ViewToArrowResp, ViewToColumnsStringResp, ViewToCsvResp,
34    ViewToNdjsonStringResp, ViewToRowsStringResp, scalar, table_describe_resp,
35};
36use crate::table::{DescribeError, Description};
37
38macro_rules! respond {
39    ($msg:ident, $name:ident { $($rest:tt)* }) => {{
40        let mut resp = BytesMut::new();
41        let resp2 = ClientResp::$name($name {
42            $($rest)*
43        });
44
45        Response {
46            msg_id: $msg.msg_id,
47            entity_id: $msg.entity_id,
48            client_resp: Some(resp2),
49        }.encode(&mut resp).map_err(VirtualServerError::EncodeError)?;
50
51        resp.freeze()
52    }};
53}
54
55/// A virtual server that processes Perspective protocol messages.
56///
57/// `VirtualServer` acts as a bridge between the Perspective protocol and a
58/// custom data backend. It handles protocol decoding/encoding and delegates
59/// actual data operations to the provided [`VirtualServerHandler`].
60pub struct VirtualServer<T: VirtualServerHandler> {
61    handler: T,
62    view_to_table: IndexMap<String, String>,
63    view_configs: IndexMap<String, ViewConfig>,
64    view_schemas: IndexMap<String, IndexMap<String, ColumnType>>,
65
66    /// The tables made from views, which this server created and may drop.
67    view_tables: IndexSet<String>,
68
69    /// Per-view `table_describe` answers, computed LAZILY on the first
70    /// `ViewExpressionSchemaReq` for that view — never on view creation.
71    view_descriptions: IndexMap<String, Description>,
72}
73
74impl<T: VirtualServerHandler> VirtualServer<T> {
75    /// Creates a new virtual server with the given handler.
76    pub fn new(handler: T) -> Self {
77        Self {
78            handler,
79            view_configs: IndexMap::default(),
80            view_to_table: IndexMap::default(),
81            view_schemas: IndexMap::default(),
82            view_tables: IndexSet::default(),
83            view_descriptions: IndexMap::default(),
84        }
85    }
86
87    /// Processes a Perspective protocol request and returns the response.
88    ///
89    /// Decodes the incoming protobuf message, dispatches to the appropriate
90    /// handler method, and encodes the response.
91    pub async fn handle_request(
92        &mut self,
93        bytes: Bytes,
94    ) -> Result<Bytes, VirtualServerError<T::Error>> {
95        let msg = Request::decode(bytes).map_err(VirtualServerError::DecodeError)?;
96        tracing::debug!(
97            "Handling request: entity_id={}, req={:?}",
98            msg.entity_id,
99            msg.client_req
100        );
101
102        let was_view = self.view_to_table.contains_key(&msg.entity_id);
103        match self.internal_handle_request(msg.clone()).await {
104            Ok(resp) => Ok(resp),
105            Err(err) if was_view && !self.view_to_table.contains_key(&msg.entity_id) => {
106                tracing::debug!("View '{}' deleted mid-request: {}", msg.entity_id, err);
107                Ok(respond!(msg, ServerError {
108                    message: VirtualServerError::<T::Error>::UnknownViewId(msg.entity_id.clone())
109                        .to_string(),
110                    status_code: StatusCode::ViewNotFound as i32
111                }))
112            },
113            Err(err @ VirtualServerError::UnknownViewId(_)) => {
114                tracing::debug!("{}", err);
115                Ok(respond!(msg, ServerError {
116                    message: err.to_string(),
117                    status_code: StatusCode::ViewNotFound as i32
118                }))
119            },
120            Err(err) => {
121                tracing::error!("{}", err);
122                Ok(respond!(msg, ServerError {
123                    message: err.to_string(),
124                    status_code: StatusCode::ServerError as i32
125                }))
126            },
127        }
128    }
129
130    /// The config of a live view, cloned because a `ViewDeleteReq` answered
131    /// during any `await` of a re-entrant caller may remove the entry.
132    fn view_config(&self, view_id: &str) -> Result<ViewConfig, VirtualServerError<T::Error>> {
133        self.view_configs
134            .get(view_id)
135            .cloned()
136            .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.to_string()))
137    }
138
139    /// The table a live view reads, cloned for the same reason as
140    /// [`Self::view_config`].
141    fn view_table(&self, view_id: &str) -> Result<String, VirtualServerError<T::Error>> {
142        self.view_to_table
143            .get(view_id)
144            .cloned()
145            .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.to_string()))
146    }
147
148    /// Deletes every view this server still holds, as closing a native
149    /// `Session` does, returning the first handler error after attempting all.
150    pub async fn close(&mut self) -> Result<(), VirtualServerError<T::Error>> {
151        let mut result = Ok(());
152        for view_id in self.view_configs.keys().cloned().collect::<Vec<_>>() {
153            if !self.view_configs.contains_key(&view_id) {
154                continue;
155            }
156
157            if let Err(err) = self.handler.view_delete(&view_id).await
158                && result.is_ok()
159            {
160                result = Err(err.into());
161            }
162
163            self.view_to_table.shift_remove(&view_id);
164            self.view_configs.shift_remove(&view_id);
165            self.view_schemas.shift_remove(&view_id);
166            self.view_descriptions.shift_remove(&view_id);
167        }
168
169        result
170    }
171
172    async fn get_cached_view_schema(
173        &mut self,
174        entity_id: &str,
175        to_psp_format: bool,
176    ) -> Result<IndexMap<String, ColumnType>, VirtualServerError<T::Error>> {
177        if !self.view_schemas.contains_key(entity_id) {
178            let config = self.view_config(entity_id)?;
179            let schema = self.handler.view_schema(entity_id, &config).await?;
180            self.view_config(entity_id)?;
181            self.view_schemas.insert(entity_id.to_string(), schema);
182        }
183
184        if to_psp_format {
185            // `view.schema()` is keyed by *source* column name, matching the
186            // native engine, while the cached schema is keyed by the view's
187            // actual (possibly pivoted-path) SQL column names.
188            let config = &self.view_config(entity_id)?;
189            Ok(self
190                .view_schemas
191                .get(entity_id)
192                .ok_or_else(|| VirtualServerError::UnknownViewId(entity_id.to_string()))?
193                .iter()
194                .map(|(k, v)| {
195                    let name = column_path_source(k, config)
196                        .map(|(_, col)| col.to_string())
197                        .unwrap_or_else(|| k.clone());
198
199                    (name, *v)
200                })
201                .collect())
202        } else {
203            self.view_schemas
204                .get(entity_id)
205                .cloned()
206                .ok_or_else(|| VirtualServerError::UnknownViewId(entity_id.to_string()))
207        }
208    }
209
210    /// Whether `view_id` has an expandable row tree, or an error if this
211    /// server cannot expand one.
212    async fn is_expandable(&self, view_id: &str) -> Result<bool, VirtualServerError<T::Error>> {
213        if !self.handler.get_features().await?.expand_collapse {
214            return Err(VirtualServerError::Other(
215                "This data source does not support `expand`, `collapse` or `set_depth`".to_string(),
216            ));
217        }
218
219        let config = self.view_config(view_id)?;
220        Ok(!config.group_by.is_empty() && config.group_rollup_mode == GroupRollupMode::Rollup)
221    }
222
223    async fn internal_handle_request(
224        &mut self,
225        msg: Request,
226    ) -> Result<Bytes, VirtualServerError<T::Error>> {
227        use crate::proto::request::ClientReq::*;
228        let resp = match msg.client_req.unwrap() {
229            GetFeaturesReq(_) => {
230                let features = self.handler.get_features().await?;
231                respond!(msg, GetFeaturesResp { ..features.into() })
232            },
233            GetHostedTablesReq(_) => {
234                let mut table_infos = self.handler.get_hosted_tables().await?;
235                for name in &self.view_tables {
236                    if !table_infos.iter().any(|x| &x.entity_id == name) {
237                        table_infos.push(HostedTable {
238                            entity_id: name.clone(),
239                            index: Vec::new(),
240                            limit: None,
241                        });
242                    }
243                }
244
245                respond!(msg, GetHostedTablesResp { table_infos })
246            },
247            TableSchemaReq(_) => {
248                respond!(msg, TableSchemaResp {
249                    schema: Some(crate::proto::Schema {
250                        schema: self
251                            .handler
252                            .table_schema(msg.entity_id.as_str())
253                            .await?
254                            .iter()
255                            .map(|x| crate::proto::schema::KeyTypePair {
256                                name: x.0.to_string(),
257                                r#type: *x.1 as i32,
258                            })
259                            .collect()
260                    })
261                })
262            },
263            TableMakePortReq(req) => {
264                respond!(msg, TableMakePortResp {
265                    port_id: self.handler.table_make_port(&req).await?
266                })
267            },
268            TableMakeViewReq(req) => {
269                let mut config: ViewConfigUpdate = req.config.clone().unwrap_or_default().into();
270                if let Some(group_by) = &config.group_by
271                    && let Some(duplicate) = group_by
272                        .iter()
273                        .enumerate()
274                        .find(|(idx, col)| group_by[..*idx].contains(*col))
275                {
276                    return Err(VirtualServerError::Other(format!(
277                        "Duplicate column `{}` in `group_by`",
278                        duplicate.1
279                    )));
280                }
281
282                // An UNORDERED store has no natural row order to fall back
283                // on, so every window must carry an explicit `order_by`.
284                if let Some(windows) = &config.windows
285                    && windows.values().any(|w| w.order_by.is_none())
286                    && self.handler.get_features().await?.unordered
287                {
288                    return Err(VirtualServerError::Other(
289                        "This data store is unordered - windows require an explicit `order_by`"
290                            .to_string(),
291                    ));
292                }
293
294                let view_id = msg.entity_id.clone();
295                let bytes = respond!(msg, TableMakeViewResp {
296                    view_id: self
297                        .handler
298                        .table_make_view(view_id.as_str(), req.view_id.as_str(), &mut config)
299                        .await?
300                });
301
302                self.view_to_table
303                    .insert(req.view_id.clone(), view_id.clone());
304                self.view_configs.insert(req.view_id.clone(), config.into());
305                bytes
306            },
307            TableSizeReq(_) => {
308                respond!(msg, TableSizeResp {
309                    size: self.handler.table_size(msg.entity_id.as_str()).await?
310                })
311            },
312            TableDescribeReq(req) => {
313                let config: ViewConfig = req.config.unwrap_or_default().into();
314                let verdict = self
315                    .handler
316                    .table_describe(msg.entity_id.as_str(), &config)
317                    .await?;
318
319                let (expression_schema, expression_errors, result) = match verdict {
320                    Ok(d) => (
321                        d.expression_schema,
322                        HashMap::new(),
323                        Some(table_describe_resp::Result::View(ViewDescription {
324                            schema: d
325                                .view_schema
326                                .into_iter()
327                                .map(|(x, y)| (x, y as i32))
328                                .collect(),
329                        })),
330                    ),
331                    Err(DescribeError::Expressions {
332                        expression_schema,
333                        errors,
334                    }) => (expression_schema, errors, None),
335                    Err(DescribeError::Config(msg)) => (
336                        HashMap::new(),
337                        HashMap::new(),
338                        Some(table_describe_resp::Result::ConfigError(msg)),
339                    ),
340                };
341
342                respond!(msg, TableDescribeResp {
343                    expression_schema: expression_schema
344                        .into_iter()
345                        .map(|(x, y)| (x, y as i32))
346                        .collect(),
347                    expression_errors,
348                    result,
349                })
350            },
351            ViewSchemaReq(_) => {
352                respond!(msg, ViewSchemaResp {
353                    schema: self
354                        .get_cached_view_schema(&msg.entity_id, true)
355                        .await?
356                        .into_iter()
357                        .map(|(x, y)| (x.to_string(), y as i32))
358                        .collect()
359                })
360            },
361            ViewDimensionsReq(_) => {
362                let view_id = &msg.entity_id;
363                let table_id = self.view_table(view_id)?;
364                let num_table_rows = self.handler.table_size(&table_id).await?;
365                let num_table_columns = self.handler.table_column_size(&table_id).await? as u32;
366                let config = self.view_config(view_id)?;
367                let num_view_columns =
368                    self.handler.view_column_size(view_id, &config).await? as u32;
369                let num_view_rows = self.handler.view_size(view_id).await?;
370                let resp = ViewDimensionsResp {
371                    num_table_columns,
372                    num_table_rows,
373                    num_view_columns,
374                    num_view_rows,
375                };
376
377                respond!(msg, ViewDimensionsResp { ..resp })
378            },
379            ViewGetConfigReq(_) => {
380                respond!(msg, ViewGetConfigResp {
381                    config: Some(ViewConfigUpdate::from(self.view_config(&msg.entity_id)?).into())
382                })
383            },
384            ViewExpressionSchemaReq(_) => {
385                let view_id = msg.entity_id.clone();
386                if !self.view_descriptions.contains_key(&view_id) {
387                    let table_id = self.view_table(&view_id)?;
388                    let config = self.view_config(&view_id)?;
389                    let description = self
390                        .handler
391                        .table_describe(&table_id, &config)
392                        .await?
393                        .map_err(|e| VirtualServerError::Other(e.to_string()))?;
394
395                    self.view_config(&view_id)?;
396                    self.view_descriptions.insert(view_id.clone(), description);
397                }
398
399                let schema = self
400                    .view_descriptions
401                    .get(&view_id)
402                    .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.clone()))?
403                    .expression_schema
404                    .iter()
405                    .map(|(name, ty)| (name.clone(), *ty as i32))
406                    .collect();
407
408                respond!(msg, ViewExpressionSchemaResp { schema })
409            },
410            ViewColumnPathsReq(view_column_paths_req) => {
411                let config = self.view_config(&msg.entity_id)?;
412                let mut paths: Vec<String> = self
413                    .handler
414                    .view_schema(msg.entity_id.as_str(), &config)
415                    .await?
416                    .keys()
417                    .cloned()
418                    .collect();
419
420                if !config.split_by.is_empty() {
421                    sort_column_paths(&mut paths, &config);
422                }
423
424                let start = view_column_paths_req.start_col.unwrap_or(0) as usize;
425                let end = view_column_paths_req
426                    .end_col
427                    .map_or(paths.len(), |x| x as usize);
428
429                let paths = paths
430                    .into_iter()
431                    .skip(start)
432                    .take(end.saturating_sub(start))
433                    .collect::<Vec<_>>();
434
435                let depth = config.split_by.len() + 1;
436                let mut area = vec![
437                    ColumnPathLevel {
438                        values: Vec::with_capacity(paths.len())
439                    };
440                    depth
441                ];
442
443                for path in paths {
444                    let mut split = path.splitn(depth, '|');
445                    let mut levels = Vec::with_capacity(depth);
446                    for _ in 0..depth {
447                        match split.next() {
448                            Some(x) => levels.push(Some(x.to_owned())),
449                            None => levels.push(None),
450                        }
451                    }
452
453                    let name = levels
454                        .iter()
455                        .rposition(|x| x.is_some())
456                        .map(|i| levels[i].take().unwrap())
457                        .unwrap_or_default();
458
459                    for (level, value) in area.iter_mut().zip(levels).take(depth - 1) {
460                        level.values.push(Scalar {
461                            scalar: value.map(scalar::Scalar::String),
462                        });
463                    }
464
465                    area[depth - 1].values.push(Scalar {
466                        scalar: Some(scalar::Scalar::String(name)),
467                    });
468                }
469
470                respond!(msg, ViewColumnPathsResp { area })
471            },
472            ViewToArrowReq(view_to_arrow_req) => {
473                let viewport = view_to_arrow_req.viewport.unwrap();
474                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
475                let config = self.view_config(&msg.entity_id)?;
476                let mut cols = self
477                    .handler
478                    .view_get_data(msg.entity_id.as_str(), &config, &schema, &viewport)
479                    .await?;
480
481                let arrow = cols
482                    .render_to_arrow_ipc()
483                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
484
485                respond!(msg, ViewToArrowResp { arrow })
486            },
487            ViewToCsvReq(view_to_csv_req) => {
488                let viewport = view_to_csv_req.viewport.unwrap();
489                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
490                let config = self.view_config(&msg.entity_id)?;
491                let mut cols = self
492                    .handler
493                    .view_get_data(msg.entity_id.as_str(), &config, &schema, &viewport)
494                    .await?;
495
496                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
497                let mut csv = String::new();
498                if let Some(first_row) = rows.first() {
499                    let headers: Vec<&str> = first_row.keys().map(|k| k.as_str()).collect();
500                    csv.push_str(&headers.join(","));
501                    csv.push('\n');
502                }
503
504                for row in &rows {
505                    let values: Vec<String> = row
506                        .values()
507                        .map(|cell| serde_json::to_string(cell).unwrap_or_default())
508                        .collect();
509                    csv.push_str(&values.join(","));
510                    csv.push('\n');
511                }
512
513                respond!(msg, ViewToCsvResp { csv })
514            },
515            ViewToNdjsonStringReq(view_to_ndjson_req) => {
516                let viewport = view_to_ndjson_req.viewport.unwrap();
517                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
518                let config = self.view_config(&msg.entity_id)?;
519                let mut cols = self
520                    .handler
521                    .view_get_data(msg.entity_id.as_str(), &config, &schema, &viewport)
522                    .await?;
523
524                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
525                let ndjson_string = rows
526                    .iter()
527                    .map(serde_json::to_string)
528                    .collect::<Result<Vec<_>, _>>()
529                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?
530                    .join("\n");
531
532                respond!(msg, ViewToNdjsonStringResp { ndjson_string })
533            },
534            ViewToRowsStringReq(view_to_rows_string_req) => {
535                let viewport = view_to_rows_string_req.viewport.unwrap();
536                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
537                let config = self.view_config(&msg.entity_id)?;
538                let mut cols = self
539                    .handler
540                    .view_get_data(msg.entity_id.as_str(), &config, &schema, &viewport)
541                    .await?;
542
543                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
544                let json_string = serde_json::to_string(&rows)
545                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?;
546
547                respond!(msg, ViewToRowsStringResp { json_string })
548            },
549            ViewToColumnsStringReq(view_to_columns_string_req) => {
550                let viewport = view_to_columns_string_req.viewport.unwrap();
551                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
552                let config = self.view_config(&msg.entity_id)?;
553                let mut cols = self
554                    .handler
555                    .view_get_data(msg.entity_id.as_str(), &config, &schema, &viewport)
556                    .await?;
557
558                let json_string = cols
559                    .render_to_columns_json(
560                        RowPathStyle::Sidecar,
561                        view_to_columns_string_req.id.unwrap_or_default(),
562                    )
563                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
564
565                respond!(msg, ViewToColumnsStringResp { json_string })
566            },
567            ViewDeleteReq(_) => {
568                self.view_table(&msg.entity_id)?;
569                self.handler.view_delete(msg.entity_id.as_str()).await?;
570                self.view_to_table.shift_remove(&msg.entity_id);
571                self.view_configs.shift_remove(&msg.entity_id);
572                self.view_schemas.shift_remove(&msg.entity_id);
573                self.view_descriptions.shift_remove(&msg.entity_id);
574                respond!(msg, ViewDeleteResp {})
575            },
576            MakeTableReq(req) => {
577                let data = req.data.as_ref().unwrap();
578                if let Some(crate::proto::make_table_data::Data::FromView(view_id)) = &data.data {
579                    if !self.handler.get_features().await?.view_derivations {
580                        return Err(VirtualServerError::Other(
581                            "This data source cannot derive a Table from a View".to_string(),
582                        ));
583                    }
584
585                    let options = req.options.clone().unwrap_or_default();
586                    if options.make_table_type.is_some() {
587                        return Err(VirtualServerError::Other(
588                            "`index` and `limit` cannot be set on a Table made from a View"
589                                .to_string(),
590                        ));
591                    }
592
593                    let config = self.view_config(view_id)?;
594
595                    let schema = options
596                        .view_schema
597                        .map(|schema| {
598                            schema
599                                .schema
600                                .into_iter()
601                                .map(|x| Ok((x.name, ColumnType::try_from(x.r#type)?)))
602                                .collect::<Result<IndexMap<_, _>, prost::DecodeError>>()
603                        })
604                        .transpose()
605                        .map_err(|e| VirtualServerError::Other(e.to_string()))?;
606
607                    self.handler
608                        .view_make_table(view_id, &msg.entity_id, &config, schema.as_ref())
609                        .await?;
610
611                    self.view_tables.insert(msg.entity_id.clone());
612                } else {
613                    self.handler.make_table(&msg.entity_id, data).await?;
614                }
615
616                respond!(msg, MakeTableResp {})
617            },
618            ViewGetMinMaxReq(req) => {
619                let config = self.view_config(&msg.entity_id)?;
620                let (min, max) = self
621                    .handler
622                    .view_get_min_max(&msg.entity_id, &req.column_name, &config)
623                    .await?;
624                respond!(msg, ViewGetMinMaxResp {
625                    min: Some(min.into()),
626                    max: Some(max.into()),
627                })
628            },
629
630            ViewExpandReq(req) => {
631                let num_changed = if self.is_expandable(&msg.entity_id).await? {
632                    let config = self.view_config(&msg.entity_id)?;
633                    self.handler
634                        .view_expand(&msg.entity_id, &config, req.row_index)
635                        .await?
636                } else {
637                    0
638                };
639
640                respond!(msg, ViewExpandResp { num_changed })
641            },
642            ViewCollapseReq(req) => {
643                let num_changed = if self.is_expandable(&msg.entity_id).await? {
644                    let config = self.view_config(&msg.entity_id)?;
645                    self.handler
646                        .view_collapse(&msg.entity_id, &config, req.row_index)
647                        .await?
648                } else {
649                    0
650                };
651
652                respond!(msg, ViewCollapseResp { num_changed })
653            },
654            ViewSetDepthReq(req) => {
655                if self.is_expandable(&msg.entity_id).await? {
656                    let config = self.view_config(&msg.entity_id)?;
657                    if req.depth as usize <= config.group_by.len() {
658                        self.handler
659                            .view_set_depth(&msg.entity_id, &config, req.depth)
660                            .await?;
661                    }
662                }
663
664                respond!(msg, ViewSetDepthResp {})
665            },
666
667            TableDeleteReq(_) if self.view_tables.contains(&msg.entity_id) => {
668                self.handler.table_delete(msg.entity_id.as_str()).await?;
669                self.view_tables.shift_remove(&msg.entity_id);
670                respond!(msg, TableDeleteResp {})
671            },
672
673            // Stub implementations for callback/update requests that VirtualServer doesn't support
674            TableOnDeleteReq(_) => {
675                respond!(msg, TableOnDeleteResp {})
676            },
677            ViewOnRemoveReq(_) => {
678                respond!(msg, ViewOnRemoveResp {
679                    indices: None,
680                    port_id: 0,
681                    cause: 0
682                })
683            },
684            ViewRemoveOnRemoveReq(_) => {
685                respond!(msg, ViewRemoveOnRemoveResp {})
686            },
687            ViewOnUpdateReq(_) => {
688                respond!(msg, ViewOnUpdateResp {
689                    delta: None,
690                    port_id: 0,
691                    reset: false
692                })
693            },
694            ViewOnDeleteReq(_) => {
695                respond!(msg, ViewOnDeleteResp {})
696            },
697            ViewRemoveOnUpdateReq(_) => {
698                respond!(msg, ViewRemoveOnUpdateResp {})
699            },
700            TableRemoveDeleteReq(_) => {
701                respond!(msg, TableRemoveDeleteResp {})
702            },
703            ViewRemoveDeleteReq(_) => {
704                respond!(msg, ViewRemoveDeleteResp {})
705            },
706            x => {
707                // Return an error response instead of empty bytes
708                return Err(VirtualServerError::Other(format!(
709                    "Unhandled request: {:?}",
710                    x
711                )));
712            },
713        };
714
715        Ok(resp)
716    }
717}
perspective_client/virtual_server/server.rs

perspective_client/virtual_server/
server.rs
