Skip to content

Commit 65bd9a8

Browse files
texodustexodus
authored
Merge pull request #2948 from finos/wupdates
Add `Client::on_hosted_tables_update`
2 parents a450820 + 878e256 commit 65bd9a8

File tree

27 files changed

+451
-141
lines changed

27 files changed

+451
-141
lines changed

‎Cargo.lock

+24
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎cpp/perspective/src/cpp/server.cpp

+119-36
Original file line numberDiff line numberDiff line change
@@ -608,6 +608,11 @@ ServerResources::remove_view_on_delete_sub(
608608
}
609609
}
610610

611+
void
612+
ServerResources::drop_view_on_delete_sub(const t_id& view_id) {
613+
m_view_on_delete_subs.erase(view_id);
614+
}
615+
611616
void
612617
ServerResources::create_view_on_update_sub(
613618
const t_id& view_id, Subscription sub
@@ -636,6 +641,53 @@ ServerResources::drop_view_on_update_sub(const t_id& view_id) {
636641
m_view_on_update_subs.erase(view_id);
637642
}
638643

644+
void
645+
ServerResources::remove_view_on_update_sub(
646+
const t_id& view_id, std::uint32_t sub_id, std::uint32_t client_id
647+
) {
648+
if (m_view_on_update_subs.find(view_id) != m_view_on_update_subs.end()) {
649+
auto& subs = m_view_on_update_subs[view_id];
650+
subs.erase(
651+
std::remove_if(
652+
subs.begin(),
653+
subs.end(),
654+
[sub_id, client_id](const Subscription& sub) {
655+
return sub.id == sub_id && sub.client_id == client_id;
656+
}
657+
),
658+
subs.end()
659+
);
660+
}
661+
}
662+
663+
void
664+
ServerResources::create_on_hosted_tables_update_sub(Subscription sub) {
665+
PSP_WRITE_LOCK(m_write_lock);
666+
m_on_hosted_tables_update_subs.push_back(sub);
667+
}
668+
669+
std::vector<Subscription>
670+
ServerResources::get_on_hosted_tables_update_sub() {
671+
PSP_READ_LOCK(m_write_lock);
672+
return m_on_hosted_tables_update_subs;
673+
}
674+
675+
void
676+
ServerResources::remove_on_hosted_tables_update_sub(
677+
std::uint32_t sub_id, std::uint32_t client_id
678+
) {
679+
m_on_hosted_tables_update_subs.erase(
680+
std::remove_if(
681+
m_on_hosted_tables_update_subs.begin(),
682+
m_on_hosted_tables_update_subs.end(),
683+
[sub_id, client_id](const Subscription& sub) {
684+
return sub.id == sub_id && sub.client_id == client_id;
685+
}
686+
),
687+
m_on_hosted_tables_update_subs.end()
688+
);
689+
}
690+
639691
std::vector<std::pair<std::shared_ptr<Table>, const ServerResources::t_id>>
640692
ServerResources::get_dirty_tables() {
641693
PSP_READ_LOCK(m_write_lock);
@@ -662,6 +714,18 @@ ServerResources::drop_client(const std::uint32_t client_id) {
662714
delete_view(client_id, view_id);
663715
}
664716
}
717+
718+
std::vector<Subscription> subs;
719+
std::remove_copy_if(
720+
m_on_hosted_tables_update_subs.begin(),
721+
m_on_hosted_tables_update_subs.end(),
722+
std::back_inserter(subs),
723+
[&client_id](const Subscription& item) {
724+
return item.client_id == client_id;
725+
}
726+
);
727+
728+
m_on_hosted_tables_update_subs = subs;
665729
}
666730

667731
std::uint32_t
@@ -876,6 +940,7 @@ needs_poll(const proto::Request::ClientReqCase proto_case) {
876940
case ReqCase::kTableUpdateReq:
877941
case ReqCase::kTableRemoveDeleteReq:
878942
case ReqCase::kGetHostedTablesReq:
943+
case ReqCase::kRemoveHostedTablesUpdateReq:
879944
case ReqCase::kTableReplaceReq:
880945
case ReqCase::kTableDeleteReq:
881946
case ReqCase::kViewGetConfigReq:
@@ -932,6 +997,7 @@ entity_type_is_table(const proto::Request::ClientReqCase proto_case) {
932997
case ReqCase::kViewDeleteReq:
933998
case ReqCase::kViewExpressionSchemaReq:
934999
case ReqCase::kViewRemoveOnUpdateReq:
1000+
case ReqCase::kRemoveHostedTablesUpdateReq:
9351001
return false;
9361002
case proto::Request::CLIENT_REQ_NOT_SET:
9371003
throw std::runtime_error("Unhandled request type 2");
@@ -1228,24 +1294,41 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) {
12281294
break;
12291295
}
12301296
case proto::Request::kGetHostedTablesReq: {
1231-
proto::Response resp;
1232-
const auto& tables = resp.mutable_get_hosted_tables_resp();
1233-
const auto& infos = tables->mutable_table_infos();
1234-
for (const auto& name : m_resources.get_table_ids()) {
1235-
const auto& v = infos->Add();
1297+
const auto& r = req.get_hosted_tables_req();
1298+
if (!r.subscribe()) {
1299+
proto::Response resp;
1300+
const auto& tables = resp.mutable_get_hosted_tables_resp();
1301+
const auto& infos = tables->mutable_table_infos();
1302+
for (const auto& name : m_resources.get_table_ids()) {
1303+
const auto& v = infos->Add();
12361304

1237-
v->set_entity_id(name);
1238-
const auto tbl = m_resources.get_table(name);
1305+
v->set_entity_id(name);
1306+
const auto tbl = m_resources.get_table(name);
12391307

1240-
if (!tbl->get_index().empty()) {
1241-
v->set_index(tbl->get_index());
1242-
}
1308+
if (!tbl->get_index().empty()) {
1309+
v->set_index(tbl->get_index());
1310+
}
12431311

1244-
if (tbl->get_limit() != std::numeric_limits<int>::max()) {
1245-
v->set_limit(tbl->get_limit());
1312+
if (tbl->get_limit() != std::numeric_limits<int>::max()) {
1313+
v->set_limit(tbl->get_limit());
1314+
}
12461315
}
1316+
1317+
push_resp(std::move(resp));
1318+
} else {
1319+
Subscription sub_info;
1320+
sub_info.id = req.msg_id();
1321+
sub_info.client_id = client_id;
1322+
m_resources.create_on_hosted_tables_update_sub(sub_info);
12471323
}
12481324

1325+
break;
1326+
}
1327+
case proto::Request::kRemoveHostedTablesUpdateReq: {
1328+
auto sub_id = req.remove_hosted_tables_update_req().id();
1329+
m_resources.remove_on_hosted_tables_update_sub(sub_id, client_id);
1330+
proto::Response resp;
1331+
resp.mutable_remove_hosted_tables_update_resp();
12491332
push_resp(std::move(resp));
12501333
break;
12511334
}
@@ -1348,6 +1431,18 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) {
13481431
proto::Response resp;
13491432
resp.mutable_make_table_resp();
13501433
push_resp(std::move(resp));
1434+
1435+
// Notify `on_thsoted_tables_update` listeners
1436+
auto subscriptions = m_resources.get_on_hosted_tables_update_sub();
1437+
for (auto& subscription : subscriptions) {
1438+
Response out;
1439+
out.set_msg_id(subscription.id);
1440+
ProtoServerResp<ProtoServer::Response> resp2;
1441+
resp2.data = std::move(out);
1442+
resp2.client_id = subscription.client_id;
1443+
proto_resp.emplace_back(std::move(resp2));
1444+
}
1445+
13511446
break;
13521447
}
13531448
case proto::Request::kTableSizeReq: {
@@ -2272,6 +2367,18 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) {
22722367
proto::Response resp;
22732368
resp.mutable_table_delete_resp();
22742369
push_resp(std::move(resp));
2370+
2371+
// notify `on_hosted_tables_update` listeners
2372+
auto subscriptions = m_resources.get_on_hosted_tables_update_sub();
2373+
for (auto& subscription : subscriptions) {
2374+
Response out;
2375+
out.set_msg_id(subscription.id);
2376+
ProtoServerResp<ProtoServer::Response> resp2;
2377+
resp2.data = std::move(out);
2378+
resp2.client_id = subscription.client_id;
2379+
proto_resp.emplace_back(std::move(resp2));
2380+
}
2381+
22752382
break;
22762383
}
22772384
case proto::Request::kViewDeleteReq: {
@@ -2509,30 +2616,6 @@ ProtoServer::_process_table(
25092616
m_resources.mark_table_clean(table_id);
25102617
}
25112618

2512-
void
2513-
ServerResources::remove_view_on_update_sub(
2514-
const t_id& view_id, std::uint32_t sub_id, std::uint32_t client_id
2515-
) {
2516-
if (m_view_on_update_subs.find(view_id) != m_view_on_update_subs.end()) {
2517-
auto& subs = m_view_on_update_subs[view_id];
2518-
subs.erase(
2519-
std::remove_if(
2520-
subs.begin(),
2521-
subs.end(),
2522-
[sub_id, client_id](const Subscription& sub) {
2523-
return sub.id == sub_id && sub.client_id == client_id;
2524-
}
2525-
),
2526-
subs.end()
2527-
);
2528-
}
2529-
}
2530-
2531-
void
2532-
ServerResources::drop_view_on_delete_sub(const t_id& view_id) {
2533-
m_view_on_delete_subs.erase(view_id);
2534-
}
2535-
25362619
} // namespace perspective::server
25372620

25382621
const char*

‎cpp/perspective/src/include/perspective/server.h

+9
Original file line numberDiff line numberDiff line change
@@ -569,6 +569,13 @@ namespace server {
569569
);
570570
void drop_view_on_delete_sub(const t_id& view_id);
571571

572+
// `on_hosted_tables_update()`
573+
void create_on_hosted_tables_update_sub(Subscription sub);
574+
std::vector<Subscription> get_on_hosted_tables_update_sub();
575+
void remove_on_hosted_tables_update_sub(
576+
std::uint32_t sub_id, std::uint32_t client_id
577+
);
578+
572579
void mark_table_dirty(const t_id& id);
573580
void mark_table_clean(const t_id& id);
574581
void mark_all_tables_clean();
@@ -594,6 +601,8 @@ namespace server {
594601
tsl::hopscotch_map<t_id, std::vector<Subscription>>
595602
m_table_on_delete_subs;
596603

604+
std::vector<Subscription> m_on_hosted_tables_update_subs;
605+
597606
tsl::hopscotch_set<t_id> m_dirty_tables;
598607

599608
#ifdef PSP_PARALLEL_FOR

‎cpp/protos/perspective.proto

+11-1
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,7 @@ message Request {
118118
// Minimum Virtual API (theoretical).
119119
GetFeaturesReq get_features_req = 3;
120120
GetHostedTablesReq get_hosted_tables_req = 4;
121+
RemoveHostedTablesUpdateReq remove_hosted_tables_update_req = 37;
121122
TableMakePortReq table_make_port_req = 5;
122123
TableMakeViewReq table_make_view_req = 6;
123124
TableSchemaReq table_schema_req = 7;
@@ -164,6 +165,7 @@ message Response {
164165
oneof client_resp {
165166
GetFeaturesResp get_features_resp = 3;
166167
GetHostedTablesResp get_hosted_tables_resp = 4;
168+
RemoveHostedTablesUpdateResp remove_hosted_tables_update_resp = 37;
167169
TableMakePortResp table_make_port_resp = 5;
168170
TableMakeViewResp table_make_view_resp = 6;
169171
TableSchemaResp table_schema_resp = 7;
@@ -219,7 +221,10 @@ message GetFeaturesResp {
219221
}
220222

221223
// `Client::get_hosted_tables`
222-
message GetHostedTablesReq {}
224+
message GetHostedTablesReq {
225+
bool subscribe = 1;
226+
}
227+
223228
message GetHostedTablesResp {
224229
repeated HostedTable table_infos = 1;
225230
}
@@ -230,6 +235,11 @@ message HostedTable {
230235
optional uint32 limit = 3;
231236
}
232237

238+
message RemoveHostedTablesUpdateReq {
239+
uint32 id = 1;
240+
}
241+
message RemoveHostedTablesUpdateResp {}
242+
233243
// `Table::size`
234244
message TableSizeReq {}
235245
message TableSizeResp {

‎examples/blocks/src/editable/README.md

+4-7
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,4 @@
1-
A simple example of [Perspective](https://github.com/finos/perspective) with
2-
superstore data, and editability enabled (not default but easy to toggle at
3-
runtime).
4-
5-
An explcit `SharedWorker` is used to create the Perspective engine, allowing
6-
edits to be shared live when this example is opened across multiple browser
7-
tabs.
1+
A simple example of [Perspective](https://github.com/finos/perspective) with superstore
2+
data, and editability enabled (not default but easy to toggle at runtime). This
3+
example has no server component, and the edits occur only within the browser session;
4+
refreshing the page will forget any edits and revert to the original dataset.

‎examples/blocks/src/editable/index.html

+6-16
Original file line numberDiff line numberDiff line change
@@ -13,23 +13,13 @@
1313

1414
import perspective from "/node_modules/@finos/perspective/dist/cdn/perspective.js";
1515

16+
const worker = await perspective.worker();
17+
const resp = await fetch("/node_modules/superstore-arrow/superstore.lz4.arrow");
18+
const arrow = await resp.arrayBuffer();
1619
const viewer = document.getElementsByTagName("perspective-viewer")[0];
17-
18-
const worker = new SharedWorker("/node_modules/@finos/perspective/dist/cdn/perspective-server.worker.js");
19-
const client = await perspective.worker(worker);
20-
const tables = await client.get_hosted_table_names();
21-
22-
if (tables.length > 0) {
23-
const table = client.open_table(tables[0]);
24-
viewer.load(table);
25-
} else {
26-
const resp = await fetch("/node_modules/superstore-arrow/superstore.lz4.arrow");
27-
const arrow = await resp.arrayBuffer();
28-
const table = client.table(arrow);
29-
viewer.load(table);
30-
}
31-
32-
viewer.restore({ plugin_config: { edit_mode: "EDIT" }, settings: true });
20+
const table = worker.table(arrow);
21+
viewer.load(table);
22+
viewer.restore({ settings: true, plugin_config: { edit_mode: "EDIT" } });
3323
</script>
3424
<style>
3525
perspective-viewer {

‎packages/perspective-workspace/src/ts/workspace/commands.ts

+2-2
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ export const createCommands = (
3939
args.widget_name as string
4040
)!;
4141

42-
menu.unsafe_set_model(await widget.viewer.unsafe_get_model());
42+
menu.set_model(widget.viewer.get_model());
4343
menu.open(indicator);
4444
workspace.get_context_menu()?.init_overlay?.();
4545
menu.addEventListener("blur", () => {
@@ -80,7 +80,7 @@ export const createCommands = (
8080
const widget = workspace.getWidgetByName(
8181
args.widget_name as string
8282
)!;
83-
menu.unsafe_set_model(await widget.viewer.unsafe_get_model());
83+
menu.set_model(widget.viewer.get_model());
8484

8585
menu.open(indicator);
8686
workspace.get_context_menu()?.init_overlay?.();

0 commit comments

Comments
 (0)