grafana/public/app/features/live/centrifuge/transferHandlers.ts
Artur Wierzbicki f45eb309ef
Live: move centrifuge service to a web worker (#41090)
* Fix: make webpack pickup workers written in TS

* Add comlink to dependencies

* Temporary fix: copy paste `toDataQueryError` from @grafana/runtime to avoid web dependencies

* Implemented comlink-based centrifuge worker & worker proxy

* Temporary fix: implement comlink transferHandlers for subscriptions and streamingdataframes

* Move liveTimer filtering from CentrifugeService into GrafanaLiveService

* Switch from CentrifugeService to CentrifugeServiceWorkerProxy in GrafanaLive

* Naming fix

* Refactor: move liveTimer-based data filtering from GrafanaLiveService to CentrifugeServiceWorker

* observe dataStream on an async scheduler

* Fix: - Unsubscribe is now propagated from the main thread to the worker, - improve worker&workerProxy types

* Fix: Prettify types

* Fix: Add error & complete observers

* Docs: Add comment explaining the `subscriberTransferHandler`

* Fix: Replace `StreamingDataFrameHandler` with explicitly converting StreamingDataFrame to a DataFrameDTO

* Refactor: move liveTimer filtering to service.ts to make it easy to implement a `live-service-web-worker` feature flag

* Feat: add `live-service-web-worker` feature flag

* Fix: extract toDataQueryError.ts to a separate file within `@grafana-runtime` to avoid having a dependency from webworker to the whole package (@grafana-runtime/index.ts)

* Update public/app/features/dashboard/dashgrid/liveTimer.ts

Co-authored-by: Leon Sorokin <leeoniya@gmail.com>

* Fix: fixed default import class in worker file

* Fix: cast worker as Endpoint

* Migrate from worker-loader to webpack native worker support v1 - broken prod build

* Fix: Use custom path in HtmlWebpackPlugin

* Fix: Loading workers from CDNs

* Fix: Avoid issues with jest ESM support by mocking `createWorker` files

* Fix: move the custom mockWorker rendering layout to `test/mocks`

Co-authored-by: Leon Sorokin <leeoniya@gmail.com>
2021-11-09 21:05:01 +04:00

28 lines
923 B
TypeScript

import * as comlink from 'comlink';
import { Subscriber } from 'rxjs';
// Observers, ie. functions passed to `observable.subscribe(...)`, are converted to a subclass of `Subscriber` before they are sent to the source Observable.
// The conversion happens internally in the RxJS library - this transfer handler is catches them and wraps them with a proxy
const subscriberTransferHandler: any = {
canHandle(value: any): boolean {
return value && value instanceof Subscriber;
},
serialize(value: Function): [MessagePort, Transferable[]] {
const obj = comlink.proxy(value);
const { port1, port2 } = new MessageChannel();
comlink.expose(obj, port1);
return [port2, [port2]];
},
deserialize(value: MessagePort): comlink.Remote<MessagePort> {
value.start();
return comlink.wrap<MessagePort>(value);
},
};
comlink.transferHandlers.set('SubscriberHandler', subscriberTransferHandler);