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

perspective_client/virtual_server/
server.rs
