snapshot of https://github.com/dream-num/univer.git (dev, 2026-09-16; blobs > 131072B omitted)

packages/rpc/src/config/config.ts

@@ -0,0 +1,34 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export const PLUGIN_CONFIG_KEY_MAIN_THREAD = 'rpc.main-thread.config';
+
+export const configSymbolMainThread = Symbol(PLUGIN_CONFIG_KEY_MAIN_THREAD);
+
+export interface IUniverRPCMainThreadConfig {
+    workerURL?: string | URL | Worker;
+}
+
+export const defaultPluginMainThreadConfig: IUniverRPCMainThreadConfig = {};
+
+export const PLUGIN_CONFIG_KEY_WORKER_THREAD = 'rpc.worker-thread.config';
+
+export const configSymbolWorkerThread = Symbol(PLUGIN_CONFIG_KEY_WORKER_THREAD);
+
+export interface IUniverRPCWorkerThreadConfig {
+}
+
+export const defaultPluginWorkerThreadConfig: IUniverRPCWorkerThreadConfig = {};

packages/rpc/src/controllers/data-sync/__tests__/data-sync.controller.spec.ts

@@ -0,0 +1,392 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { ICommandService, Injector, IUniverInstanceService } from '@univerjs/core';
+import type { IRPCChannelService } from '../../../services/rpc/channel.service';
+import type { IChannel } from '../../../services/rpc/rpc.service';
+import { CommandType, UniverInstanceType } from '@univerjs/core';
+import { Subject } from 'rxjs';
+import { describe, expect, it, vi } from 'vitest';
+import { IRemoteInstanceService, IRemoteSyncService, RemoteInstanceServiceName, RemoteSyncServiceName } from '../../../services/remote-instance/remote-instance.service';
+import { DataSyncPrimaryController } from '../data-sync-primary.controller';
+import { DataSyncReplicaController } from '../data-sync-replica.controller';
+
+interface IDependencyFactory {
+    useFactory: () => unknown;
+}
+
+function buildRemoteChannel(target: Record<string, (...args: unknown[]) => unknown>): IChannel {
+    return {
+        call: (async (method: string, args?: unknown[]) => {
+            const fn = target[method];
+            if (!fn) {
+                throw new Error(`missing method: ${method}`);
+            }
+            return args ? fn(...args) : fn();
+        }) as IChannel['call'],
+        subscribe: (() => new Subject<unknown>().asObservable()) as IChannel['subscribe'],
+    };
+}
+
+async function flushPromises(times = 6): Promise<void> {
+    for (let index = 0; index < times; index++) {
+        await Promise.resolve();
+    }
+}
+
+describe('data-sync controllers', () => {
+    it('DataSyncPrimaryController should sync add/dispose/mutation with branch guards', async () => {
+        const added$ = new Subject<{ unit: { getUnitId(): string; getSnapshot(): unknown } }>();
+        const disposed$ = new Subject<{ getUnitId(): string }>();
+        let commandCallback: ((commandInfo: unknown, options?: unknown) => void) | undefined;
+        const dependencies: Array<[unknown, IDependencyFactory]> = [];
+        let remoteProxy: unknown;
+
+        const remoteInstanceImpl = {
+            createInstance: vi.fn(async () => true),
+            disposeInstance: vi.fn(async () => true),
+            syncMutation: vi.fn(async () => true),
+            whenReady: vi.fn(async () => true as const),
+        };
+
+        const remoteChannel = buildRemoteChannel(remoteInstanceImpl);
+
+        const rpcChannelService: IRPCChannelService = {
+            requestChannel: vi.fn(() => remoteChannel),
+            registerChannel: vi.fn(),
+        };
+
+        const injector = {
+            add: vi.fn((dependency: [unknown, IDependencyFactory]) => {
+                dependencies.push(dependency);
+            }),
+            get: vi.fn((token: unknown) => {
+                if (token === IRemoteInstanceService) {
+                    if (!remoteProxy) {
+                        const record = dependencies.find(([depToken]) => depToken === IRemoteInstanceService);
+                        remoteProxy = record?.[1].useFactory();
+                    }
+                    return remoteProxy;
+                }
+                return undefined;
+            }),
+        } as unknown as Injector;
+
+        const commandService = {
+            onCommandExecuted: vi.fn((callback: (commandInfo: unknown, options?: unknown) => void) => {
+                commandCallback = callback;
+                return { dispose: vi.fn() };
+            }),
+        } as unknown as ICommandService;
+
+        const univerInstanceService = {
+            getTypeOfUnitAdded$: vi.fn(() => added$.asObservable()),
+            getTypeOfUnitDisposed$: vi.fn(() => disposed$.asObservable()),
+            getUnit: vi.fn((unitId: string) => {
+                if (unitId !== 'unit-2') {
+                    return null;
+                }
+                return {
+                    type: UniverInstanceType.UNIVER_BASE,
+                    getUnitId: () => 'unit-2',
+                    getSnapshot: () => ({ base: true }),
+                };
+            }),
+        } as unknown as IUniverInstanceService;
+
+        const remoteSyncService = {
+            syncMutation: vi.fn(async () => true),
+        };
+
+        const controller = new DataSyncPrimaryController(
+            injector,
+            commandService,
+            univerInstanceService,
+            rpcChannelService,
+            remoteSyncService as never
+        );
+
+        expect(rpcChannelService.registerChannel).toHaveBeenCalledWith(RemoteSyncServiceName, expect.any(Object));
+        expect(rpcChannelService.requestChannel).toHaveBeenCalledWith(RemoteInstanceServiceName);
+
+        const workbook = {
+            getUnitId: () => 'unit-1',
+            getSnapshot: () => ({ snapshot: true }),
+        };
+        added$.next({ unit: workbook });
+        await Promise.resolve();
+        expect(remoteInstanceImpl.createInstance).toHaveBeenCalledWith({
+            unitID: 'unit-1',
+            type: UniverInstanceType.UNIVER_SHEET,
+            snapshot: { snapshot: true },
+        });
+
+        disposed$.next({ getUnitId: () => 'unit-1' });
+        await Promise.resolve();
+        expect(remoteInstanceImpl.disposeInstance).toHaveBeenCalledWith({ unitID: 'unit-1' });
+
+        controller.registerSyncingMutations({ id: 'm-sync' } as never);
+
+        const unit2Disposable = controller.syncUnit('unit-2');
+        expect(remoteInstanceImpl.createInstance).toHaveBeenCalledWith({
+            unitID: 'unit-2',
+            type: UniverInstanceType.UNIVER_BASE,
+            snapshot: { base: true },
+        });
+
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-2' },
+        }, {});
+        await flushPromises();
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(1);
+
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.COMMAND,
+            params: { unitId: 'unit-2' },
+        }, {});
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: { unitId: 'other' },
+        }, {});
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-2' },
+        }, { fromSync: true });
+        commandCallback?.({
+            id: 'm-ignore',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-2' },
+        }, {});
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: {},
+        }, {});
+        await flushPromises();
+
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(2);
+
+        unit2Disposable.dispose();
+        expect(remoteInstanceImpl.disposeInstance).toHaveBeenCalledWith({ unitID: 'unit-2' });
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-2' },
+        }, {});
+        await flushPromises();
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(2);
+
+        const unit3Disposable = controller.syncUnitMutations('unit-3');
+        commandCallback?.({
+            id: 'm-sync',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-3' },
+        }, {});
+        await flushPromises();
+        expect(remoteInstanceImpl.createInstance).toHaveBeenCalledTimes(2);
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(3);
+        unit3Disposable.dispose();
+
+        controller.dispose();
+    });
+
+    it('DataSyncPrimaryController should serialize remote mutation sync calls', async () => {
+        const added$ = new Subject<{ unit: { getUnitId(): string; getSnapshot(): unknown } }>();
+        const disposed$ = new Subject<{ getUnitId(): string }>();
+        let commandCallback: ((commandInfo: unknown, options?: unknown) => void) | undefined;
+        const dependencies: Array<[unknown, IDependencyFactory]> = [];
+        let remoteProxy: unknown;
+        const resolvers: Array<(value: boolean) => void> = [];
+
+        const remoteInstanceImpl = {
+            createInstance: vi.fn(async () => true),
+            disposeInstance: vi.fn(async () => true),
+            syncMutation: vi.fn(() => new Promise<boolean>((resolve) => resolvers.push(resolve))),
+            whenReady: vi.fn(async () => true as const),
+        };
+
+        const remoteChannel = buildRemoteChannel(remoteInstanceImpl);
+
+        const rpcChannelService: IRPCChannelService = {
+            requestChannel: vi.fn(() => remoteChannel),
+            registerChannel: vi.fn(),
+        };
+
+        const injector = {
+            add: vi.fn((dependency: [unknown, IDependencyFactory]) => {
+                dependencies.push(dependency);
+            }),
+            get: vi.fn((token: unknown) => {
+                if (token === IRemoteInstanceService) {
+                    if (!remoteProxy) {
+                        const record = dependencies.find(([depToken]) => depToken === IRemoteInstanceService);
+                        remoteProxy = record?.[1].useFactory();
+                    }
+                    return remoteProxy;
+                }
+                return undefined;
+            }),
+        } as unknown as Injector;
+
+        const commandService = {
+            onCommandExecuted: vi.fn((callback: (commandInfo: unknown, options?: unknown) => void) => {
+                commandCallback = callback;
+                return { dispose: vi.fn() };
+            }),
+        } as unknown as ICommandService;
+
+        const univerInstanceService = {
+            getTypeOfUnitAdded$: vi.fn(() => added$.asObservable()),
+            getTypeOfUnitDisposed$: vi.fn(() => disposed$.asObservable()),
+            getUnit: vi.fn(() => null),
+        } as unknown as IUniverInstanceService;
+
+        const remoteSyncService = {
+            syncMutation: vi.fn(async () => true),
+        };
+
+        const controller = new DataSyncPrimaryController(
+            injector,
+            commandService,
+            univerInstanceService,
+            rpcChannelService,
+            remoteSyncService as never
+        );
+
+        controller.registerSyncingMutations({ id: 'm-source' } as never);
+        controller.registerSyncingMutations({ id: 'm-calculate' } as never);
+        controller.syncUnit('unit-1');
+
+        commandCallback?.({
+            id: 'm-source',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-1' },
+        }, {});
+        commandCallback?.({
+            id: 'm-calculate',
+            type: CommandType.MUTATION,
+            params: { unitId: 'unit-1' },
+        }, {});
+
+        await flushPromises();
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(1);
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenNthCalledWith(1, {
+            mutationInfo: expect.objectContaining({ id: 'm-source' }),
+        }, {});
+
+        resolvers[0](true);
+        await flushPromises();
+
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenCalledTimes(2);
+        expect(remoteInstanceImpl.syncMutation).toHaveBeenNthCalledWith(2, {
+            mutationInfo: expect.objectContaining({ id: 'm-calculate' }),
+        }, {});
+
+        resolvers[1](true);
+        await flushPromises();
+        controller.dispose();
+    });
+
+    it('DataSyncReplicaController should sync only mutation not fromSync', async () => {
+        let commandCallback: ((commandInfo: unknown, options?: unknown) => void) | undefined;
+        const dependencies: Array<[unknown, IDependencyFactory]> = [];
+        let remoteSyncProxy: unknown;
+
+        const remoteSyncImpl = {
+            syncMutation: vi.fn(async () => true),
+        };
+
+        const remoteChannel = buildRemoteChannel(remoteSyncImpl);
+
+        const rpcChannelService: IRPCChannelService = {
+            requestChannel: vi.fn(() => remoteChannel),
+            registerChannel: vi.fn(),
+        };
+
+        const injector = {
+            add: vi.fn((dependency: [unknown, IDependencyFactory]) => {
+                dependencies.push(dependency);
+            }),
+            get: vi.fn((token: unknown) => {
+                if (token === IRemoteSyncService) {
+                    if (!remoteSyncProxy) {
+                        const record = dependencies.find(([depToken]) => depToken === IRemoteSyncService);
+                        remoteSyncProxy = record?.[1].useFactory();
+                    }
+                    return remoteSyncProxy;
+                }
+                return undefined;
+            }),
+        } as unknown as Injector;
+
+        const remoteInstanceService = {
+            createInstance: vi.fn(async () => true),
+            disposeInstance: vi.fn(async () => true),
+            syncMutation: vi.fn(async () => true),
+            whenReady: vi.fn(async () => true),
+        };
+
+        const commandService = {
+            onCommandExecuted: vi.fn((callback: (commandInfo: unknown, options?: unknown) => void) => {
+                commandCallback = callback;
+                return { dispose: vi.fn() };
+            }),
+        } as unknown as ICommandService;
+
+        const controller = new DataSyncReplicaController(
+            injector,
+            remoteInstanceService as never,
+            commandService,
+            rpcChannelService
+        );
+
+        expect(rpcChannelService.registerChannel).toHaveBeenCalledWith(RemoteInstanceServiceName, expect.any(Object));
+        expect(rpcChannelService.requestChannel).toHaveBeenCalledWith(RemoteSyncServiceName);
+
+        commandCallback?.({
+            id: 'm1',
+            type: CommandType.MUTATION,
+            params: { a: 1 },
+        }, {});
+        commandCallback?.({
+            id: 'm2',
+            type: CommandType.MUTATION,
+            params: { a: 2 },
+        }, { fromSync: true });
+        commandCallback?.({
+            id: 'c1',
+            type: CommandType.COMMAND,
+            params: { a: 3 },
+        }, {});
+
+        await Promise.resolve();
+        expect(remoteSyncImpl.syncMutation).toHaveBeenCalledTimes(1);
+        expect(remoteSyncImpl.syncMutation).toHaveBeenCalledWith({
+            mutationInfo: {
+                id: 'm1',
+                type: CommandType.MUTATION,
+                params: { a: 1 },
+            },
+        }, {});
+
+        controller.dispose();
+    });
+});

packages/rpc/src/controllers/data-sync/data-sync-primary.controller.ts

@@ -0,0 +1,177 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { BaseDataModel, IDisposable, IMutation, IMutationInfo, Workbook } from '@univerjs/core';
+import type { IRemoteSyncMutationOptions } from '../../services/remote-instance/remote-instance.service';
+import {
+    CommandType,
+    ICommandService,
+    Inject,
+    Injector,
+    IUniverInstanceService,
+    RxDisposable,
+    toDisposable,
+    UniverInstanceType,
+} from '@univerjs/core';
+import { takeUntil } from 'rxjs/operators';
+import {
+    IRemoteInstanceService,
+    IRemoteSyncService,
+    RemoteInstanceServiceName,
+    RemoteSyncServiceName,
+} from '../../services/remote-instance/remote-instance.service';
+import { IRPCChannelService } from '../../services/rpc/channel.service';
+import { fromModule, toModule } from '../../services/rpc/rpc.service';
+
+/**
+ * This controller is responsible for syncing data from the primary thread to
+ * the worker thread.
+ *
+ * Note that only spreadsheets will be synced to the remote calculation instance by default.
+ */
+export class DataSyncPrimaryController extends RxDisposable {
+    private _remoteInstanceService!: IRemoteInstanceService;
+
+    private readonly _syncingUnits = new Set<string>();
+
+    private readonly _syncingMutations = new Set<string>();
+
+    private _syncMutationQueue: Promise<unknown> = Promise.resolve();
+
+    private _remoteReady: Promise<true> = Promise.resolve(true);
+
+    constructor(
+        @Inject(Injector) private readonly _injector: Injector,
+        @ICommandService private readonly _commandService: ICommandService,
+        @IUniverInstanceService private readonly _univerInstanceService: IUniverInstanceService,
+        @IRPCChannelService private readonly _rpcChannelService: IRPCChannelService,
+        @IRemoteSyncService private readonly _remoteSyncService: IRemoteSyncService
+    ) {
+        super();
+
+        this._initRPCChannels();
+        this._init();
+    }
+
+    registerSyncingMutations(mutation: IMutation<object>): void {
+        this._syncingMutations.add(mutation.id);
+    }
+
+    /**
+     * Only spreadsheets would be synced to the web worker in normal situations. If you would like to
+     * sync other types of documents, you should manually call this method with that document's id.
+     */
+    syncUnit(unitId: string): IDisposable {
+        const alreadySyncing = this._syncingUnits.has(unitId);
+        this._syncingUnits.add(unitId);
+        const unit = this._univerInstanceService.getUnit<Workbook>(unitId, UniverInstanceType.UNIVER_SHEET)
+            ?? this._univerInstanceService.getUnit<BaseDataModel>(unitId, UniverInstanceType.UNIVER_BASE);
+        if (!alreadySyncing && unit) {
+            this._remoteInstanceService.createInstance({
+                unitID: unit.getUnitId(),
+                type: unit.type,
+                snapshot: unit.getSnapshot(),
+            });
+        }
+
+        return toDisposable(() => {
+            if (!alreadySyncing) {
+                this._syncingUnits.delete(unitId);
+            }
+            if (!alreadySyncing && unit) {
+                this._remoteInstanceService.disposeInstance({
+                    unitID: unit.getUnitId(),
+                });
+            }
+        });
+    }
+
+    /** Sync registered mutations for a unit without creating a replica unit. */
+    syncUnitMutations(unitId: string): IDisposable {
+        const alreadySyncing = this._syncingUnits.has(unitId);
+        this._syncingUnits.add(unitId);
+        return toDisposable(() => {
+            if (!alreadySyncing) {
+                this._syncingUnits.delete(unitId);
+            }
+        });
+    }
+
+    async syncMutation(commandInfo: IMutationInfo, options?: IRemoteSyncMutationOptions): Promise<boolean> {
+        const sync = () => this._remoteReady.then(() =>
+            this._remoteInstanceService.syncMutation({ mutationInfo: commandInfo }, options)
+        );
+        const result = this._syncMutationQueue.then(sync, sync);
+        this._syncMutationQueue = result.catch(() => false);
+        return result;
+    }
+
+    async waitForPendingMutations(): Promise<void> {
+        await this._syncMutationQueue.catch(() => false);
+    }
+
+    private _initRPCChannels(): void {
+        // for the worker to call
+        this._rpcChannelService.registerChannel(RemoteSyncServiceName, fromModule(this._remoteSyncService));
+
+        // to call the worker
+        this._injector.add([
+            IRemoteInstanceService,
+            // eslint-disable-next-line react/no-unnecessary-use-prefix
+            { useFactory: () => toModule<IRemoteInstanceService>(this._rpcChannelService.requestChannel(RemoteInstanceServiceName)) },
+        ]);
+        this._remoteInstanceService = this._injector.get(IRemoteInstanceService);
+        this._remoteReady = this._remoteInstanceService.whenReady();
+    }
+
+    private _init(): void {
+        this._univerInstanceService.getTypeOfUnitAdded$<Workbook>(UniverInstanceType.UNIVER_SHEET).pipe(takeUntil(this.dispose$)).subscribe((event) => {
+            const { unit: sheet } = event;
+            this._syncingUnits.add(sheet.getUnitId());
+
+            // If a sheet is created, it should sync the data to the worker thread.
+            this._remoteInstanceService.createInstance({
+                unitID: sheet.getUnitId(),
+                type: UniverInstanceType.UNIVER_SHEET,
+                snapshot: sheet.getSnapshot(),
+            });
+        });
+
+        this._univerInstanceService.getTypeOfUnitDisposed$<Workbook>(UniverInstanceType.UNIVER_SHEET).pipe(takeUntil(this.dispose$)).subscribe((workbook) => {
+            this._syncingUnits.delete(workbook.getUnitId());
+            // If a sheet is disposed, it should sync the data to the worker thread.
+            this._remoteInstanceService.disposeInstance({
+                unitID: workbook.getUnitId(),
+            });
+        });
+
+        // Mutations executed on the main thread should be synced to the worker thread.
+        this.disposeWithMe(this._commandService.onCommandExecuted((commandInfo, options) => {
+            const { type, params, id } = commandInfo;
+            const unitId = (params as { unitId: string })?.unitId || '';
+
+            if (type === CommandType.MUTATION &&
+                // only sync mutations to the worker thread
+                (!unitId || this._syncingUnits.has(unitId)) &&
+                // do not sync mutations from the web worker back to the web worker
+                !(options as IRemoteSyncMutationOptions)?.fromSync &&
+                // do not sync mutations those are not meant to be synced
+                this._syncingMutations.has(id)) {
+                void this.syncMutation(commandInfo as IMutationInfo, options);
+            }
+        }));
+    }
+}

packages/rpc/src/controllers/data-sync/data-sync-replica.controller.ts

@@ -0,0 +1,71 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IMutationInfo } from '@univerjs/core';
+import type { IRemoteSyncMutationOptions } from '../../services/remote-instance/remote-instance.service';
+import { CommandType, Disposable, ICommandService, Inject, Injector } from '@univerjs/core';
+import {
+    IRemoteInstanceService,
+    IRemoteSyncService,
+    RemoteInstanceServiceName,
+    RemoteSyncServiceName,
+} from '../../services/remote-instance/remote-instance.service';
+import { IRPCChannelService } from '../../services/rpc/channel.service';
+import { fromModule, toModule } from '../../services/rpc/rpc.service';
+
+/**
+ * This controller is responsible for syncing data from the worker thread to
+ * the primary thread.
+ */
+export class DataSyncReplicaController extends Disposable {
+    private _remoteSyncService!: IRemoteSyncService;
+
+    constructor(
+        @Inject(Injector) private readonly _injector: Injector,
+        @IRemoteInstanceService private readonly _remoteInstanceService: IRemoteInstanceService,
+        @ICommandService private readonly _commandService: ICommandService,
+        @IRPCChannelService private readonly _rpcChannelService: IRPCChannelService
+    ) {
+        super();
+
+        this._initRPCChannels();
+        this._init();
+    }
+
+    private _initRPCChannels(): void {
+        this._rpcChannelService.registerChannel(RemoteInstanceServiceName, fromModule(this._remoteInstanceService));
+
+        this._injector.add([
+            IRemoteSyncService,
+            // eslint-disable-next-line react/no-unnecessary-use-prefix
+            { useFactory: () => toModule<IRemoteSyncService>(this._rpcChannelService.requestChannel(RemoteSyncServiceName)) },
+        ]);
+        this._remoteSyncService = this._injector.get(IRemoteSyncService);
+    }
+
+    private _init(): void {
+        this.disposeWithMe(
+            // Mutations executed on the main thread should be synced to the worker thread.
+            this._commandService.onCommandExecuted((commandInfo, options) => {
+                if (commandInfo.type === CommandType.MUTATION && !(options as IRemoteSyncMutationOptions)?.fromSync) {
+                    this._remoteSyncService.syncMutation({
+                        mutationInfo: commandInfo as IMutationInfo,
+                    }, options);
+                }
+            })
+        );
+    }
+}

packages/rpc/src/index.ts

@@ -0,0 +1,35 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export type { IUniverRPCMainThreadConfig, IUniverRPCWorkerThreadConfig } from './config/config';
+export { PLUGIN_CONFIG_KEY_MAIN_THREAD, PLUGIN_CONFIG_KEY_WORKER_THREAD } from './config/config';
+export { DataSyncPrimaryController } from './controllers/data-sync/data-sync-primary.controller';
+export { DataSyncReplicaController } from './controllers/data-sync/data-sync-replica.controller';
+export {
+    UniverRPCMainThreadPlugin,
+    UniverRPCWorkerThreadPlugin,
+} from './plugin';
+export {
+    IRemoteInstanceService,
+    RemoteInstanceServiceName,
+    RemoteSyncPrimaryService,
+    RemoteSyncServiceName,
+    WebWorkerRemoteInstanceService,
+} from './services/remote-instance/remote-instance.service';
+export { IRemoteSyncService } from './services/remote-instance/remote-instance.service';
+export { ChannelService, IRPCChannelService } from './services/rpc/channel.service';
+export { ChannelClient, ChannelServer, fromModule, toModule } from './services/rpc/rpc.service';
+export type { IMessageProtocol } from './services/rpc/rpc.service';

packages/rpc/src/plugin.ts

@@ -0,0 +1,150 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { Dependency } from '@univerjs/core';
+import type {
+    IUniverRPCMainThreadConfig,
+    IUniverRPCWorkerThreadConfig,
+} from './config/config';
+import { IConfigService, Inject, Injector, merge, Plugin } from '@univerjs/core';
+import pkg from '../package.json';
+import {
+    defaultPluginMainThreadConfig,
+    defaultPluginWorkerThreadConfig,
+    PLUGIN_CONFIG_KEY_MAIN_THREAD,
+    PLUGIN_CONFIG_KEY_WORKER_THREAD,
+} from './config/config';
+import { DataSyncPrimaryController } from './controllers/data-sync/data-sync-primary.controller';
+import { DataSyncReplicaController } from './controllers/data-sync/data-sync-replica.controller';
+import {
+    IRemoteInstanceService,
+    IRemoteSyncService,
+    RemoteSyncPrimaryService,
+    WebWorkerRemoteInstanceService,
+} from './services/remote-instance/remote-instance.service';
+import { ChannelService, IRPCChannelService } from './services/rpc/channel.service';
+import {
+    createWebWorkerMessagePortOnMain,
+    createWebWorkerMessagePortOnWorker,
+} from './services/rpc/implementations/web-worker-rpc.service';
+
+/**
+ * This plugin is used to register the RPC services on the main thread. It
+ * is also responsible for booting up the Web Worker instance of Univer.
+ */
+export class UniverRPCMainThreadPlugin extends Plugin {
+    static override pluginName = 'UNIVER_RPC_MAIN_THREAD_PLUGIN';
+    static override packageName = pkg.name;
+    static override version = pkg.version;
+
+    private _internalWorker: Worker | null = null;
+
+    constructor(
+        private readonly _config: Partial<IUniverRPCMainThreadConfig> = defaultPluginMainThreadConfig,
+        @Inject(Injector) protected readonly _injector: Injector,
+        @IConfigService private readonly _configService: IConfigService
+    ) {
+        super();
+
+        // Manage the plugin configuration.
+        const { ...rest } = merge(
+            {},
+            defaultPluginMainThreadConfig,
+            this._config
+        );
+        this._configService.setConfig(PLUGIN_CONFIG_KEY_MAIN_THREAD, rest);
+    }
+
+    override dispose(): void {
+        super.dispose();
+
+        if (this._internalWorker) {
+            this._internalWorker.terminate();
+            this._internalWorker = null;
+        }
+    }
+
+    override onStarting(): void {
+        const { workerURL } = this._config;
+        if (!workerURL) {
+            throw new Error('[UniverRPCMainThreadPlugin]: The workerURL is required for the RPC main thread plugin.');
+        }
+
+        const worker = workerURL instanceof Worker ? workerURL : new Worker(workerURL);
+        this._internalWorker = workerURL instanceof Worker ? null : worker;
+
+        const messageProtocol = createWebWorkerMessagePortOnMain(worker);
+        const dependencies: Dependency[] = [
+            [
+                IRPCChannelService,
+                {
+                    // eslint-disable-next-line react/no-unnecessary-use-prefix
+                    useFactory: () => new ChannelService(messageProtocol),
+                },
+            ],
+            [DataSyncPrimaryController],
+            [IRemoteSyncService, { useClass: RemoteSyncPrimaryService }],
+        ];
+
+        dependencies.forEach((dependency) => this._injector.add(dependency));
+
+        // let DataSyncPrimaryController to be initialized and registering other modules
+        this._injector.get(DataSyncPrimaryController);
+    }
+}
+
+/**
+ * This plugin is used to register the RPC services on the worker thread.
+ */
+export class UniverRPCWorkerThreadPlugin extends Plugin {
+    static override pluginName = 'UNIVER_RPC_WORKER_THREAD_PLUGIN';
+    static override packageName = pkg.name;
+    static override version = pkg.version;
+
+    constructor(
+        private readonly _config: Partial<IUniverRPCWorkerThreadConfig> = defaultPluginWorkerThreadConfig,
+        @Inject(Injector) protected readonly _injector: Injector,
+        @IConfigService private readonly _configService: IConfigService
+    ) {
+        super();
+
+        // Manage the plugin configuration.
+        const { ...rest } = merge(
+            {},
+            defaultPluginWorkerThreadConfig,
+            this._config
+        );
+        this._configService.setConfig(PLUGIN_CONFIG_KEY_WORKER_THREAD, rest);
+    }
+
+    override onStarting(): void {
+        (
+            [
+                [DataSyncReplicaController],
+                [
+                    IRPCChannelService,
+                    {
+                        // eslint-disable-next-line react/no-unnecessary-use-prefix
+                        useFactory: () => new ChannelService(createWebWorkerMessagePortOnWorker()),
+                    },
+                ],
+                [IRemoteInstanceService, { useClass: WebWorkerRemoteInstanceService }],
+            ] as Dependency[]
+        ).forEach((dependency) => this._injector.add(dependency));
+
+        this._injector.get(DataSyncReplicaController);
+    }
+}

packages/rpc/src/services/remote-instance/__tests__/remote-instance.service.spec.ts

@@ -0,0 +1,170 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ICommandService, ILogService, Injector, IUniverInstanceService, LifecycleService, LifecycleStages, UniverInstanceType } from '@univerjs/core';
+import { describe, expect, it, vi } from 'vitest';
+import { IRemoteInstanceService, IRemoteSyncService, RemoteSyncPrimaryService, WebWorkerRemoteInstanceService } from '../remote-instance.service';
+
+describe('remote-instance.service', () => {
+    it('RemoteSyncPrimaryService should sync mutation with normalized options', async () => {
+        const syncExecuteCommand = vi.fn(async () => true);
+        class TestCommandService {
+            syncExecuteCommand = syncExecuteCommand;
+        }
+
+        const injector = new Injector();
+        injector.add([ICommandService, { useClass: TestCommandService as never }]);
+        injector.add([IRemoteSyncService, { useClass: RemoteSyncPrimaryService }]);
+        const service = injector.get(IRemoteSyncService);
+        await expect(service.syncMutation(
+            {
+                mutationInfo: {
+                    id: 'mutation.id',
+                    params: { a: 1 },
+                } as never,
+            },
+            {
+                fromCollab: true,
+                source: 'test',
+            } as never
+        )).resolves.toBe(true);
+
+        expect(syncExecuteCommand).toHaveBeenCalledWith('mutation.id', { a: 1 }, {
+            source: 'test',
+            onlyLocal: true,
+            fromSync: true,
+        });
+
+        await expect(service.syncMutation({
+            mutationInfo: {
+                id: 'mutation.id.2',
+                params: { b: 2 },
+            } as never,
+        })).resolves.toBe(true);
+        expect(syncExecuteCommand).toHaveBeenCalledWith('mutation.id.2', { b: 2 }, {
+            onlyLocal: true,
+            fromSync: true,
+        });
+    });
+
+    it('WebWorkerRemoteInstanceService should cover create/sync/dispose branches', async () => {
+        const createUnit = vi.fn();
+        const disposeUnit = vi.fn(() => true);
+        const syncExecuteCommand = vi.fn(() => true);
+        const debug = vi.fn();
+
+        class TestUniverInstanceService {
+            createUnit = createUnit;
+            disposeUnit = disposeUnit;
+        }
+
+        class TestCommandService {
+            syncExecuteCommand = syncExecuteCommand;
+        }
+
+        class TestLogService {
+            debug = debug;
+        }
+
+        const injector = new Injector();
+        injector.add([IUniverInstanceService, { useClass: TestUniverInstanceService as never }]);
+        injector.add([ICommandService, { useClass: TestCommandService as never }]);
+        injector.add([ILogService, { useClass: TestLogService as never }]);
+        injector.add([LifecycleService]);
+        injector.add([IRemoteInstanceService, { useClass: WebWorkerRemoteInstanceService }]);
+        const service = injector.get(IRemoteInstanceService);
+
+        const ready = service.whenReady();
+        let readyResolved = false;
+        ready.then(() => readyResolved = true);
+        await Promise.resolve();
+        expect(readyResolved).toBe(false);
+        injector.get(LifecycleService).stage = LifecycleStages.Ready;
+        await expect(ready).resolves.toBe(true);
+
+        await expect(service.syncMutation(
+            {
+                mutationInfo: {
+                    id: 'm1',
+                    params: { x: 1 },
+                } as never,
+            },
+            {
+                fromCollab: true,
+                source: 'sync',
+            } as never
+        )).resolves.toBe(true);
+        expect(syncExecuteCommand).toHaveBeenCalledWith('m1', { x: 1 }, {
+            source: 'sync',
+            onlyLocal: true,
+            fromSync: true,
+        });
+
+        await expect(service.syncMutation({
+            mutationInfo: {
+                id: 'm1b',
+                params: { y: 2 },
+            } as never,
+        })).resolves.toBe(true);
+        expect(syncExecuteCommand).toHaveBeenCalledWith('m1b', { y: 2 }, {
+            onlyLocal: true,
+            fromSync: true,
+        });
+
+        await expect(service.createInstance({
+            unitID: 'sheet-1',
+            type: UniverInstanceType.UNIVER_SHEET,
+            snapshot: { id: 'snapshot' } as never,
+        })).resolves.toBe(true);
+        expect(createUnit).toHaveBeenCalledWith(UniverInstanceType.UNIVER_SHEET, { id: 'snapshot' });
+
+        await expect(service.createInstance({
+            unitID: 'base-1',
+            type: UniverInstanceType.UNIVER_BASE,
+            snapshot: { id: 'base-snapshot' } as never,
+        })).resolves.toBe(true);
+        expect(createUnit).toHaveBeenCalledWith(UniverInstanceType.UNIVER_BASE, { id: 'base-snapshot' });
+
+        await expect(service.disposeInstance({ unitID: 'sheet-1' })).resolves.toBe(true);
+        expect(disposeUnit).toHaveBeenCalledWith('sheet-1');
+        expect(debug).toHaveBeenCalled();
+
+        await expect(service.createInstance({
+            unitID: 'doc-1',
+            type: UniverInstanceType.UNIVER_DOC,
+            snapshot: {} as never,
+        })).rejects.toThrow(`[WebWorkerRemoteInstanceService]: cannot create replica for document type: ${UniverInstanceType.UNIVER_DOC}.`);
+
+        createUnit.mockImplementationOnce(() => {
+            throw new Error('create error');
+        });
+        await expect(service.createInstance({
+            unitID: 'sheet-2',
+            type: UniverInstanceType.UNIVER_SHEET,
+            snapshot: {} as never,
+        })).rejects.toThrow('create error');
+
+        createUnit.mockImplementationOnce(() => {
+            const nonError: unknown = 'bad-value';
+            throw nonError;
+        });
+        await expect(service.createInstance({
+            unitID: 'sheet-3',
+            type: UniverInstanceType.UNIVER_SHEET,
+            snapshot: {} as never,
+        })).rejects.toThrow('bad-value');
+    });
+});

packages/rpc/src/services/remote-instance/remote-instance.service.ts

@@ -0,0 +1,130 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IBaseSnapshot, IExecutionOptions, IMutationInfo, IWorkbookData } from '@univerjs/core';
+import { createIdentifier, ICommandService, ILogService, Inject, IUniverInstanceService, LifecycleService, LifecycleStages, UniverInstanceType } from '@univerjs/core';
+
+export interface IRemoteSyncMutationOptions extends IExecutionOptions {
+    /** If this mutation is executed after it was sent from the peer univer instance (e.g. in a web worker). */
+    fromSync?: boolean;
+}
+
+export const RemoteSyncServiceName = 'rpc.remote-sync.service';
+/**
+ * This service is provided by the primary Univer.
+ *
+ * Replica Univer could call this service to update mutations back to the primary Univer.
+ */
+export const IRemoteSyncService = createIdentifier<IRemoteSyncService>(RemoteSyncServiceName);
+export interface IRemoteSyncService {
+    syncMutation(params: { mutationInfo: IMutationInfo }, options?: IExecutionOptions): Promise<boolean>;
+}
+export class RemoteSyncPrimaryService implements IRemoteSyncService {
+    constructor(@ICommandService private readonly _commandService: ICommandService) {
+        // empty
+    }
+
+    async syncMutation(params: { mutationInfo: IMutationInfo }, options?: IExecutionOptions): Promise<boolean> {
+        const { fromCollab, ...restOptions } = options || {};
+        return this._commandService.syncExecuteCommand(params.mutationInfo.id, params.mutationInfo.params, {
+            ...restOptions,
+            onlyLocal: true,
+            fromSync: true,
+        });
+    }
+}
+
+export const RemoteInstanceServiceName = 'univer.remote-instance-service';
+
+/**
+ * This service is provided by the replica Univer.
+ *
+ * Primary univer could call this service to init and dispose univer business instances
+ * and sync mutations to replica univer.
+ */
+export const IRemoteInstanceService = createIdentifier<IRemoteInstanceService>(RemoteInstanceServiceName);
+export interface IRemoteInstanceService {
+    /** Tell other modules if the `IRemoteInstanceService` is ready to load files. */
+    whenReady(): Promise<true>;
+
+    createInstance(params: { unitID: string; type: UniverInstanceType; snapshot: IWorkbookData | IBaseSnapshot }): Promise<boolean>;
+    disposeInstance(params: { unitID: string }): Promise<boolean>;
+    syncMutation(params: { mutationInfo: IMutationInfo }, options?: IExecutionOptions): Promise<boolean>;
+}
+
+export class WebWorkerRemoteInstanceService implements IRemoteInstanceService {
+    constructor(
+        @IUniverInstanceService protected readonly _univerInstanceService: IUniverInstanceService,
+        @ICommandService protected readonly _commandService: ICommandService,
+        @ILogService protected readonly _logService: ILogService,
+        @Inject(LifecycleService) protected readonly _lifecycleService: LifecycleService
+    ) {
+        // empty
+    }
+
+    whenReady(): Promise<true> {
+        if (this._lifecycleService.stage >= LifecycleStages.Ready) return Promise.resolve(true);
+        return this._lifecycleService.onStage(LifecycleStages.Ready).then(() => true);
+    }
+
+    async syncMutation(params: { mutationInfo: IMutationInfo }, options?: IExecutionOptions): Promise<boolean> {
+        return this._applyMutation(params.mutationInfo, options);
+    }
+
+    async createInstance(params: {
+        unitID: string;
+        type: UniverInstanceType;
+        snapshot: IWorkbookData | IBaseSnapshot;
+    }): Promise<boolean> {
+        this._logService.debug(`[WebWorkerRemoteInstanceService]: Creating instance with id ${params.unitID}`);
+        const { type, snapshot } = params;
+        try {
+            switch (type) {
+                case UniverInstanceType.UNIVER_SHEET:
+                    this._univerInstanceService.createUnit(UniverInstanceType.UNIVER_SHEET, snapshot);
+                    return true;
+                case UniverInstanceType.UNIVER_BASE:
+                    this._univerInstanceService.createUnit(UniverInstanceType.UNIVER_BASE, snapshot);
+                    return true;
+                default:
+                    throw new Error(
+                        `[WebWorkerRemoteInstanceService]: cannot create replica for document type: ${type}.`
+                    );
+            }
+        } catch (err: unknown) {
+            if (err instanceof Error) {
+                throw err;
+            } else {
+                throw new TypeError(`${err}`);
+            }
+        }
+    }
+
+    async disposeInstance(params: { unitID: string }): Promise<boolean> {
+        this._logService.debug(`[WebWorkerRemoteInstanceService]: Disposing instance with id ${params.unitID}`);
+        return this._univerInstanceService.disposeUnit(params.unitID);
+    }
+
+    protected _applyMutation(mutationInfo: IMutationInfo, options?: IExecutionOptions): boolean {
+        const { id, params: mutationParams } = mutationInfo;
+        const { fromCollab, ...restOptions } = options || {};
+        return this._commandService.syncExecuteCommand(id, mutationParams, {
+            ...restOptions,
+            onlyLocal: true,
+            fromSync: true,
+        });
+    }
+}

packages/rpc/src/services/rpc/__tests__/channel.service.spec.ts

@@ -0,0 +1,59 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IMessageProtocol } from '../rpc.service';
+import { awaitTime, Injector } from '@univerjs/core';
+import { Observable, Subject } from 'rxjs';
+import { describe, expect, it } from 'vitest';
+import { ChannelService } from '../channel.service';
+import { fromModule } from '../rpc.service';
+
+describe('ChannelService', () => {
+    it('should register channel, request channel, and dispose', async () => {
+        const message$ = new Subject<unknown>();
+        const protocol: IMessageProtocol = {
+            send(message: unknown): void {
+                message$.next(message);
+            },
+            onMessage: message$.asObservable(),
+        };
+
+        const injector = new Injector();
+        const service = injector.createInstance(ChannelService, protocol);
+
+        service.registerChannel('math', fromModule({
+            add(a: number, b: number) {
+                return a + b;
+            },
+            values$() {
+                return new Observable<number>((subscriber) => {
+                    subscriber.next(1);
+                    subscriber.complete();
+                });
+            },
+        }));
+
+        const channel = service.requestChannel('math');
+        await expect(channel.call<number>('add', [2, 3])).resolves.toBe(5);
+
+        const values: number[] = [];
+        channel.subscribe<number>('values$').subscribe((value) => values.push(value));
+        await awaitTime(0);
+        expect(values).toEqual([1]);
+
+        service.dispose();
+    });
+});

packages/rpc/src/services/rpc/__tests__/rpc.service.edge.spec.ts

@@ -0,0 +1,287 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IChannel, IMessageProtocol } from '../rpc.service';
+import { awaitTime } from '@univerjs/core';
+import { Observable, Subject } from 'rxjs';
+import { describe, expect, it, vi } from 'vitest';
+import { ChannelClient, ChannelServer, fromModule, toModule } from '../rpc.service';
+
+const REQUEST_INITIALIZATION = 50;
+const CALL = 100;
+const SUBSCRIBE = 101;
+const UNSUBSCRIBE = 102;
+
+const INITIALIZE = 0;
+const CALL_SUCCESS = 201;
+const CALL_FAILURE = 202;
+const SUBSCRIBE_NEXT = 300;
+const SUBSCRIBE_ERROR = 301;
+const SUBSCRIBE_COMPLETE = 302;
+
+function getType(value: unknown): number | undefined {
+    if (typeof value !== 'object' || value === null) {
+        return undefined;
+    }
+    return (value as { type?: number }).type;
+}
+
+function getSeq(value: unknown): number | undefined {
+    if (typeof value !== 'object' || value === null) {
+        return undefined;
+    }
+    return (value as { seq?: number }).seq;
+}
+
+class TestMessageProtocol {
+    readonly sent: unknown[] = [];
+    private readonly _message$ = new Subject<unknown>();
+    readonly onMessage = this._message$.asObservable();
+
+    send(message: unknown): void {
+        this.sent.push(message);
+    }
+
+    emit(message: unknown): void {
+        this._message$.next(message);
+    }
+}
+
+describe('rpc.service edge cases', () => {
+    it('fromModule should cover args/no-args, non-observable and missing methods', async () => {
+        const channel = fromModule({
+            sum(a: number, b: number) {
+                return a + b;
+            },
+            ping() {
+                return 'pong';
+            },
+            value$() {
+                return 7;
+            },
+        });
+
+        await expect(channel.call<number>('sum', [1, 2])).resolves.toBe(3);
+        await expect(channel.call<string>('ping')).resolves.toBe('pong');
+
+        const values: number[] = [];
+        channel.subscribe<number>('value$').subscribe((value) => values.push(value));
+        expect(values).toEqual([7]);
+
+        expect(() => channel.call('missing')).toThrow('[RPC]: method not found for missing!');
+        expect(() => channel.subscribe('missing$')).toThrow('[RPC]: observable method not found for missing$!');
+    });
+
+    it('toModule should route call/subscribe and hide dispose', async () => {
+        const call = vi.fn(async () => 'ok') as IChannel['call'];
+        const subscribe = vi.fn(() => new Observable<number>((subscriber) => {
+            subscriber.next(1);
+            subscriber.complete();
+        })) as IChannel['subscribe'];
+
+        const channel: IChannel = { call, subscribe };
+        const module = toModule<{
+            dispose?: () => void;
+            fetch(value: number): Promise<string>;
+            values$(): Observable<number>;
+        }>(channel);
+
+        expect(module.dispose).toBeUndefined();
+        await expect(module.fetch(123)).resolves.toBe('ok');
+
+        const values: number[] = [];
+        module.values$().subscribe((value) => values.push(value));
+        expect(values).toEqual([1]);
+
+        expect(call).toHaveBeenCalledWith('fetch', [123]);
+        expect(subscribe).toHaveBeenCalledWith('values$', []);
+    });
+
+    it('ChannelClient should wait initialization, support disposal, and hit unknown response branches', async () => {
+        const protocol = new TestMessageProtocol();
+        const client = new ChannelClient(protocol as IMessageProtocol);
+        const channel = client.getChannel<IChannel>('demo');
+
+        const callPromise = channel.call<string>('get', ['value']);
+        expect(getType(protocol.sent[0])).toBe(REQUEST_INITIALIZATION);
+        expect(protocol.sent.some((msg) => getType(msg) === CALL)).toBe(false);
+
+        const lazySubscription = channel.subscribe<number>('count$').subscribe(() => undefined);
+        lazySubscription.unsubscribe();
+
+        protocol.emit({ seq: -1, type: INITIALIZE });
+        await awaitTime(0);
+
+        const callRequest = protocol.sent.find((msg) => getType(msg) === CALL);
+        expect(callRequest).toBeDefined();
+        const callSeq = getSeq(callRequest);
+        expect(callSeq).toBeTypeOf('number');
+
+        protocol.emit({ seq: callSeq, type: CALL_SUCCESS, data: 'done' });
+        await expect(callPromise).resolves.toBe('done');
+
+        const failurePromise = channel.call<string>('fail', []);
+        const failureRequest = [...protocol.sent].reverse().find((msg) => getType(msg) === CALL);
+        protocol.emit({ seq: getSeq(failureRequest), type: CALL_FAILURE, data: 'boom' });
+        await expect(failurePromise).rejects.toBe('boom');
+
+        const unknownCallPromise = channel.call<string>('unknown', []);
+        const unknownCallRequest = [...protocol.sent].reverse().find((msg) => getType(msg) === CALL);
+        const callPending = (client as unknown as {
+            _pendingRequests: Map<number, { handle(response: { seq: number; type: number; data?: unknown }): void }>;
+        })._pendingRequests;
+        expect(() =>
+            callPending.get(getSeq(unknownCallRequest) as number)?.handle({
+                seq: getSeq(unknownCallRequest) as number,
+                type: SUBSCRIBE_NEXT,
+                data: 'x',
+            })
+        ).toThrow('[ChannelClient]: unknown response type!');
+        callPending.delete(getSeq(unknownCallRequest) as number);
+        unknownCallPromise.catch(() => undefined);
+
+        const values: number[] = [];
+        const stream = channel.subscribe<number>('stream$').subscribe((value) => values.push(value));
+        const subscribeRequest = [...protocol.sent].reverse().find((msg) => getType(msg) === SUBSCRIBE);
+        const subscribeSeq = getSeq(subscribeRequest);
+
+        protocol.emit({ seq: subscribeSeq, type: SUBSCRIBE_NEXT, data: 1 });
+        expect(values).toEqual([1]);
+
+        const subscribePending = (client as unknown as {
+            _pendingRequests: Map<number, { handle(response: { seq: number; type: number; data?: unknown }): void }>;
+        })._pendingRequests;
+        expect(() =>
+            subscribePending.get(subscribeSeq as number)?.handle({
+                seq: subscribeSeq as number,
+                type: CALL_SUCCESS,
+                data: 'x',
+            })
+        ).toThrow('[ChannelClient]: unknown response type!');
+
+        protocol.emit({ seq: subscribeSeq, type: SUBSCRIBE_COMPLETE });
+        stream.unsubscribe();
+
+        const pendingCall = channel.call('pending');
+        const subscriptionError = vi.fn();
+        channel.subscribe('pending$').subscribe({ error: subscriptionError });
+        client.dispose();
+
+        await expect(pendingCall).rejects.toThrow('[ChannelClient]: client is disposed!');
+        expect(subscriptionError).toHaveBeenCalledWith(expect.objectContaining({
+            message: '[ChannelClient]: client is disposed!',
+        }));
+        await expect(channel.call('disposed')).rejects.toThrow('[ChannelClient]: client is disposed!');
+        expect(() => channel.subscribe('disposed$')).toThrow('[ChannelClient]: client is disposed!');
+        client.dispose();
+
+        const uninitializedClient = new ChannelClient(new TestMessageProtocol() as IMessageProtocol);
+        const waitingCall = uninitializedClient.getChannel('waiting').call('pending');
+        uninitializedClient.dispose();
+        await expect(waitingCall).rejects.toThrow('[ChannelClient]: client is disposed!');
+    });
+
+    it('ChannelServer should handle request branches, call/subscribe failures and unsubscription', async () => {
+        const protocol = new TestMessageProtocol();
+        const server = new ChannelServer(protocol as IMessageProtocol);
+
+        expect(getType(protocol.sent[0])).toBe(INITIALIZE);
+
+        protocol.emit({ type: REQUEST_INITIALIZATION, seq: 1, channelName: '', method: '' });
+        expect(protocol.sent.filter((msg) => getType(msg) === INITIALIZE).length).toBe(2);
+
+        protocol.emit({ type: 999, seq: 1, channelName: '', method: '' });
+
+        protocol.emit({ type: CALL, seq: 10, channelName: 'missing', method: 'm' });
+        await awaitTime(0);
+        expect(protocol.sent).toContainEqual({
+            seq: 10,
+            type: CALL_FAILURE,
+            data: '[ChannelServer]: Channel missing not found!',
+        });
+
+        server.registerChannel('bad-call', {
+            call: () => {
+                const nonError: unknown = 'text-error';
+                return Promise.reject(nonError);
+            },
+            subscribe: () => new Observable(),
+        } as IChannel);
+        protocol.emit({ type: CALL, seq: 11, channelName: 'bad-call', method: 'm', args: [] });
+        await awaitTime(0);
+        expect(protocol.sent).toContainEqual({
+            seq: 11,
+            type: CALL_FAILURE,
+            data: 'text-error',
+        });
+
+        server.registerChannel('no-args', {
+            call: async () => 'ok',
+            subscribe: () => new Observable(),
+        } as IChannel);
+        protocol.emit({ type: CALL, seq: 12, channelName: 'no-args', method: 'm' });
+        await awaitTime(0);
+        expect(protocol.sent).toContainEqual({
+            seq: 12,
+            type: CALL_SUCCESS,
+            data: 'ok',
+        });
+
+        protocol.emit({ type: SUBSCRIBE, seq: 20, channelName: 'missing-sub', method: 'stream$' });
+        expect(protocol.sent).toContainEqual({
+            seq: 20,
+            type: SUBSCRIBE_ERROR,
+            data: '[ChannelServer]: Channel missing-sub not found!',
+        });
+
+        server.registerChannel('bad-sub', {
+            call: async () => true,
+            subscribe: () => {
+                const nonError: unknown = 'subscribe-error';
+                throw nonError;
+            },
+        } as IChannel);
+        protocol.emit({ type: SUBSCRIBE, seq: 21, channelName: 'bad-sub', method: 'stream$' });
+        expect(protocol.sent).toContainEqual({
+            seq: 21,
+            type: SUBSCRIBE_ERROR,
+            data: 'subscribe-error',
+        });
+
+        let unsubscribed = false;
+        server.registerChannel('stream', {
+            call: async () => true,
+            subscribe: () => new Observable<string>(() => () => {
+                unsubscribed = true;
+            }),
+        } as IChannel);
+        protocol.emit({ type: SUBSCRIBE, seq: 30, channelName: 'stream', method: 'stream$', args: [] });
+        protocol.emit({ type: UNSUBSCRIBE, seq: 999, channelName: 'stream', method: 'stream$' });
+        protocol.emit({ type: UNSUBSCRIBE, seq: 30, channelName: 'stream', method: 'stream$' });
+        expect(unsubscribed).toBe(true);
+
+        let disposedSubscription = false;
+        server.registerChannel('dispose-stream', {
+            call: async () => true,
+            subscribe: () => new Observable<string>(() => () => {
+                disposedSubscription = true;
+            }),
+        } as IChannel);
+        protocol.emit({ type: SUBSCRIBE, seq: 31, channelName: 'dispose-stream', method: 'stream$' });
+        server.dispose();
+        expect(disposedSubscription).toBe(true);
+    });
+});

packages/rpc/src/services/rpc/__tests__/rpc.service.spec.ts

@@ -0,0 +1,201 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { Observable } from 'rxjs';
+import type { IMessageProtocol } from '../rpc.service';
+import { awaitTime } from '@univerjs/core';
+import { BehaviorSubject, of, Subject } from 'rxjs';
+import { beforeEach, describe, expect, it } from 'vitest';
+import { ChannelClient, ChannelServer, fromModule, toModule } from '../rpc.service';
+
+async function wait() {
+    return awaitTime(1);
+}
+
+describe('test ChannelClient & ChannelServer', () => {
+    let clientProtocol: TestMessageProtocolForClient;
+    let serverProtocol: TestMessageProtocolForServer;
+
+    let client: ChannelClient;
+    let server: ChannelServer;
+
+    // Here we mock an in-memory message protocol for client and server.
+
+    class TestMessageProtocolForClient implements IMessageProtocol {
+        private readonly _message$ = new Subject<any>();
+        onMessage = this._message$.asObservable();
+
+        send(message: any): void {
+            serverProtocol.mockSendMessage(message);
+        }
+
+        mockSendMessage(message: any): void {
+            this._message$.next(message);
+        }
+    }
+
+    class TestMessageProtocolForServer implements IMessageProtocol {
+        private readonly _message$ = new Subject<any>();
+        onMessage = this._message$.asObservable();
+
+        send(message: any): void {
+            clientProtocol.mockSendMessage(message);
+        }
+
+        mockSendMessage(message: any): void {
+            this._message$.next(message);
+        }
+    }
+
+    beforeEach(() => {
+        clientProtocol = new TestMessageProtocolForClient();
+        serverProtocol = new TestMessageProtocolForServer();
+
+        client = new ChannelClient(clientProtocol);
+        server = new ChannelServer(serverProtocol);
+    });
+
+    describe('test fromModule and toModule', () => {
+        it('should remote call work', async () => {
+            interface INameService {
+                getName(): Promise<string>;
+                getCount$(): Observable<number>;
+            }
+
+            const clientService = toModule<INameService>(client.getChannel('test'));
+            server.registerChannel(
+                'test',
+                fromModule({
+                    async getName(): Promise<string> {
+                        return 'service';
+                    },
+                    getCount$(): Observable<number> {
+                        return of(1, 2, 3);
+                    },
+                } as INameService)
+            );
+
+            const name = await clientService.getName();
+            expect(name).toBe('service');
+
+            // Should not emit any value when the returned observable is not subscribe.
+            const values: number[] = [];
+            const observable = clientService.getCount$();
+            await wait();
+            expect(values).toEqual([]);
+
+            // Should send request only after the observable is subscribed.
+            const subscription = observable.subscribe((value) => values.push(value));
+            await wait();
+            expect(values).toEqual([1, 2, 3]);
+            expect(subscription.closed).toBe(true);
+        });
+
+        it('should remote subscription work', async () => {
+            interface ISignalService {
+                getSignal$(): Observable<number>;
+            }
+
+            const signalSubject$ = new BehaviorSubject<number>(0);
+
+            const clientService = toModule<ISignalService>(client.getChannel('test'));
+            server.registerChannel(
+                'test',
+                fromModule({
+                    getSignal$(): Observable<number> {
+                        return signalSubject$.asObservable();
+                    },
+                })
+            );
+
+            let err: string = '';
+            let completed = false;
+
+            const values: number[] = [];
+            const subscription = clientService.getSignal$()
+                .subscribe({
+                    next(value) {
+                        values.push(value);
+                    },
+                    error(e) {
+                        err = e;
+                    },
+                    complete() {
+                        completed = true;
+                    },
+                });
+
+            await wait();
+            expect(values).toEqual([0]);
+
+            signalSubject$.next(1);
+            await wait();
+            expect(values).toEqual([0, 1]);
+
+            signalSubject$.error(new Error('mock error'));
+            await wait();
+            expect(err).toEqual('mock error');
+
+            subscription.unsubscribe();
+            signalSubject$.next(2);
+            await wait();
+            expect(values).toEqual([0, 1]);
+
+            signalSubject$.complete();
+            await wait();
+            expect(completed).toBe(false);
+        });
+
+        it('should remote completion work', async () => {
+            interface ISignalService {
+                getSignal$(): Observable<number>;
+            }
+
+            const signalSubject$ = new BehaviorSubject<number>(0);
+
+            const clientService = toModule<ISignalService>(client.getChannel('test'));
+            server.registerChannel(
+                'test',
+                fromModule({
+                    getSignal$(): Observable<number> {
+                        return signalSubject$.asObservable();
+                    },
+                })
+            );
+
+            let completed = false;
+
+            const values: number[] = [];
+            const subscription = clientService.getSignal$()
+                .subscribe({
+                    next(value) {
+                        values.push(value);
+                    },
+                    complete() {
+                        completed = true;
+                    },
+                });
+
+            await wait();
+            expect(values).toEqual([0]);
+
+            signalSubject$.complete();
+            await wait();
+            expect(completed).toBe(true);
+            expect(subscription.closed).toBe(true);
+        });
+    });
+});

packages/rpc/src/services/rpc/channel.service.ts

@@ -0,0 +1,53 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IDisposable } from '@univerjs/core';
+import type { IChannel, IMessageProtocol } from './rpc.service';
+import { createIdentifier } from '@univerjs/core';
+import { ChannelClient, ChannelServer } from './rpc.service';
+
+export interface IRPCChannelService {
+    requestChannel(name: string): IChannel;
+    registerChannel(name: string, channel: IChannel): void;
+}
+
+export const IRPCChannelService = createIdentifier<IRPCChannelService>('IRPCChannelService');
+
+/**
+ * This service is responsible for managing the RPC channels.
+ */
+export class ChannelService implements IDisposable {
+    private readonly _client: ChannelClient;
+    private readonly _server: ChannelServer;
+
+    constructor(_messageProtocol: IMessageProtocol) {
+        this._client = new ChannelClient(_messageProtocol);
+        this._server = new ChannelServer(_messageProtocol);
+    }
+
+    dispose(): void {
+        this._client.dispose();
+        this._server.dispose();
+    }
+
+    requestChannel(name: string): IChannel {
+        return this._client.getChannel(name);
+    }
+
+    registerChannel(name: string, channel: IChannel): void {
+        this._server.registerChannel(name, channel);
+    }
+}

packages/rpc/src/services/rpc/implementations/__tests__/web-worker-rpc.service.spec.ts

@@ -0,0 +1,78 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { createWebWorkerMessagePortOnMain, createWebWorkerMessagePortOnWorker } from '../web-worker-rpc.service';
+
+vi.mock('rxjs', async () => {
+    const actual = await vi.importActual<typeof import('rxjs')>('rxjs');
+    return {
+        ...actual,
+        shareReplay: () => (source: unknown) => source,
+    };
+});
+
+describe('web-worker-rpc.service', () => {
+    afterEach(() => {
+        vi.restoreAllMocks();
+    });
+
+    it('should create worker-side message protocol', () => {
+        const postMessageSpy = vi.fn();
+        const addEventListenerSpy = vi.fn();
+        const removeEventListenerSpy = vi.fn();
+
+        vi.stubGlobal('postMessage', postMessageSpy);
+        vi.stubGlobal('addEventListener', addEventListenerSpy);
+        vi.stubGlobal('removeEventListener', removeEventListenerSpy);
+
+        const protocol = createWebWorkerMessagePortOnWorker();
+        protocol.send('worker-message');
+
+        const received: unknown[] = [];
+        const subscription = protocol.onMessage.subscribe((message) => received.push(message));
+        const handler = addEventListenerSpy.mock.calls[0][1] as (event: MessageEvent) => void;
+        handler({ data: 'from-main' } as MessageEvent);
+
+        expect(postMessageSpy).toHaveBeenCalledWith('worker-message');
+        expect(received).toEqual(['from-main']);
+
+        subscription.unsubscribe();
+        expect(removeEventListenerSpy).toHaveBeenCalledWith('message', handler);
+    });
+
+    it('should create main-side message protocol', () => {
+        const worker = {
+            postMessage: vi.fn(),
+            addEventListener: vi.fn(),
+            removeEventListener: vi.fn(),
+        };
+
+        const protocol = createWebWorkerMessagePortOnMain(worker as unknown as Worker);
+        protocol.send('main-message');
+
+        const received: unknown[] = [];
+        const subscription = protocol.onMessage.subscribe((message) => received.push(message));
+        const handler = worker.addEventListener.mock.calls[0][1] as (event: MessageEvent) => void;
+        handler({ data: 'from-worker' } as MessageEvent);
+
+        expect(worker.postMessage).toHaveBeenCalledWith('main-message');
+        expect(received).toEqual(['from-worker']);
+
+        subscription.unsubscribe();
+        expect(worker.removeEventListener).toHaveBeenCalledWith('message', handler);
+    });
+});

packages/rpc/src/services/rpc/implementations/web-worker-rpc.service.ts

@@ -0,0 +1,57 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { IMessageProtocol } from '../rpc.service';
+import { Observable, shareReplay } from 'rxjs';
+
+/**
+ * Generate an `IMessageProtocol` on the web worker.
+ * @returns A protocol wrapper around worker global messaging APIs.
+ */
+export function createWebWorkerMessagePortOnWorker(): IMessageProtocol {
+    return {
+        send(message: unknown): void {
+            postMessage(message);
+        },
+        onMessage: new Observable<unknown>((subscriber) => {
+            const handler = (event: MessageEvent) => {
+                subscriber.next(event.data);
+            };
+            addEventListener('message', handler);
+            return () => removeEventListener('message', handler);
+        }).pipe(shareReplay({ bufferSize: 1, refCount: true })),
+    };
+}
+
+/**
+ * Generate an `IMessageProtocol` on the main thread side.
+ * @param worker The Web Worker object
+ * @returns A protocol wrapper around the given worker messaging APIs.
+ */
+export function createWebWorkerMessagePortOnMain(worker: Worker): IMessageProtocol {
+    return {
+        send(message) {
+            worker.postMessage(message);
+        },
+        onMessage: new Observable<unknown>((subscriber) => {
+            const handler = (event: MessageEvent) => {
+                subscriber.next(event.data);
+            };
+            worker.addEventListener('message', handler);
+            return () => worker.removeEventListener('message', handler);
+        }).pipe(shareReplay({ bufferSize: 1, refCount: true })),
+    };
+}

packages/rpc/src/services/rpc/rpc.service.ts

@@ -0,0 +1,489 @@
+/**
+ * Copyright 2023-present DreamNum Co., Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/* eslint-disable ts/no-explicit-any */
+
+import type { Subscription } from 'rxjs';
+import { RxDisposable } from '@univerjs/core';
+import { BehaviorSubject, firstValueFrom, isObservable, Observable, of } from 'rxjs';
+import { filter, take, takeUntil } from 'rxjs/operators';
+
+// TODO: change this parameter type to `Serializable`.
+
+/** This protocol is for transferring data from the two peer univer instance running in different locations. */
+export interface IMessageProtocol {
+    send(message: any): void;
+    onMessage: Observable<any>;
+}
+
+// TODO: change this parameter type to `Serializable`.
+
+/**
+ * Channel is a combination of methods and event sources. These methods and
+ * event sources are usually provided by the same service or controller.
+ */
+export interface IChannel {
+    call<T>(method: string, args?: any): Promise<T>;
+    subscribe<T>(event: string, args?: any): Observable<T>;
+}
+
+/**
+ * Wrapper a service or a controller into a channel so it could be invoked by
+ * a remote client. When the protocol is called, it would forward to the
+ * underlying service or controller.
+ *
+ * @param module the wrapper service or controller
+ * @returns the channel instance
+ */
+export function fromModule(module: unknown): IChannel {
+    const handler = module as { [key: string]: unknown };
+
+    // Iterate over the module and find all the methods and event sources.
+    // const observables = new Map<string, Observable<any>>();
+
+    return new (class implements IChannel {
+        call<T>(method: string, args?: any): Promise<T> {
+            const target = handler[method];
+            if (typeof target === 'function') {
+                let res = args ? target.apply(handler, args) : target.call(handler);
+                if (!(res instanceof Promise)) {
+                    res = Promise.resolve(res);
+                }
+                return res;
+            }
+
+            throw new Error(`[RPC]: method not found for ${method}!`);
+        }
+
+        subscribe<T>(eventMethod: string, args?: any): Observable<T> {
+            const target = handler[eventMethod];
+            if (typeof target === 'function') {
+                const res = args ? target.apply(handler, args) : target.call(handler);
+                if (!isObservable(res)) {
+                    return of(res);
+                }
+                return res as Observable<T>;
+            }
+
+            throw new Error(`[RPC]: observable method not found for ${eventMethod}!`);
+        }
+    })();
+}
+
+/**
+ * Wrap a channel into a service or a controller so it could be invoked by
+ * the upper layer modules. When the service or controller is called, it would
+ * request the remote server by calling the channel.
+ *
+ * @param channel
+ * @returns A proxy object that forwards calls/subscriptions to the remote channel.
+ */
+export function toModule<T extends object>(channel: IChannel): T {
+    return new Proxy({} as T, {
+        get(_: T, propKey: string) {
+            // a remote module is not disposable to the current instance
+            if (propKey === 'dispose') {
+                return undefined;
+            }
+
+            return function (...args: any[]) {
+                const isObservable = propertyIsEventSource(propKey);
+                if (isObservable) {
+                    const observable = channel.subscribe(propKey, args);
+                    return observable;
+                }
+
+                return channel.call(propKey, args);
+            };
+        },
+    });
+}
+
+function propertyIsEventSource(name: string): boolean {
+    return name.endsWith('$');
+}
+
+export interface IChannelClient {
+    getChannel<T extends IChannel>(channelName: string): T;
+}
+
+export interface IChannelServer {
+    registerChannel<T extends IChannel>(channelName: string, channel: T): void;
+}
+
+enum RequestType {
+    /**
+     * In Univer, we cannot make sure that when IPCServer constructs, the process (or thread)
+     * where the corresponding IPCClient residents has bootstrapped and been ready to recieve messages.
+     * This may result in the IPCClient hanging there, waiting for the `INITIALIZE` message that it has
+     * already missed. So the client should send a REQUEST_INITIALIZATION in case of that.
+     *
+     * Later, we may want a more sophisticated RPC system where the server can serve more than
+     * one clients, and this event may be removed.
+     */
+    REQUEST_INITIALIZATION = 50,
+
+    /** A simple remote calling wrapper in a Promise. */
+    CALL = 100,
+
+    /** Subscribe to a remote Observable. */
+    SUBSCRIBE = 101,
+
+    /** Cancel subscription to that remove Observable. */
+    UNSUBSCRIBE = 102,
+}
+
+interface IRPCRequest {
+    seq: number;
+    type: RequestType;
+    channelName: string;
+    method: string;
+    args?: any[];
+}
+
+enum ResponseType {
+    /**
+     * When underlying protocol is established. The server should send the
+     * client an `INITIALIZE` response to indicate that the server is up
+     * and ready to provide services.
+     */
+    INITIALIZE = 0,
+
+    CALL_SUCCESS = 201,
+    CALL_FAILURE = 202,
+
+    SUBSCRIBE_NEXT = 300,
+    SUBSCRIBE_ERROR = 301,
+    SUBSCRIBE_COMPLETE = 302,
+}
+
+interface IRPCResponse {
+    /** It should be the same as its corresponding requests' `seq`. */
+    seq: number;
+    type: ResponseType;
+
+    data?: any; // TODO: replace it with ISerializable.
+}
+
+interface IResponseHandler {
+    handle(response: IRPCResponse): void;
+    dispose(error: Error): void;
+}
+
+/**
+ * This method provides implementation for `IChannel` and is responsible for
+ * transforming a local calling to a RPC calling.
+ */
+export class ChannelClient extends RxDisposable implements IChannelClient {
+    private _initialized = new BehaviorSubject<boolean>(false);
+    private _lastRequestCounter = 0;
+    private _pendingRequests = new Map<number, IResponseHandler>();
+
+    constructor(private readonly _protocol: IMessageProtocol) {
+        super();
+
+        this._protocol.send({ type: RequestType.REQUEST_INITIALIZATION });
+        this._protocol.onMessage.pipe(takeUntil(this.dispose$)).subscribe((message) => this._onMessage(message));
+    }
+
+    override dispose(): void {
+        if (this._disposed) {
+            return;
+        }
+
+        const error = new Error('[ChannelClient]: client is disposed!');
+        this._initialized.error(error);
+        for (const responseHandler of this._pendingRequests.values()) {
+            responseHandler.dispose(error);
+        }
+        this._pendingRequests.clear();
+        super.dispose();
+    }
+
+    getChannel<T extends IChannel>(channelName: string): T {
+        return {
+            call: (method: string, args?: any) => {
+                if (this._disposed) {
+                    return Promise.reject(new Error('[ChannelClient]: client is disposed!'));
+                }
+
+                return this._remoteCall(channelName, method, args);
+            },
+            subscribe: (eventMethod: string, args?: any) => {
+                if (this._disposed) {
+                    throw new Error('[ChannelClient]: client is disposed!');
+                }
+
+                return this._remoteSubscribe(channelName, eventMethod, args);
+            },
+        } as T;
+    }
+
+    private _whenReady(): Promise<boolean> {
+        return firstValueFrom(
+            this._initialized.pipe(
+                filter((v) => v),
+                take(1)
+            )
+        );
+    }
+
+    private _remoteCall(channelName: string, method: string, args?: any): Promise<any> {
+        // Fast path: if the channel is already initialized, execute synchronously
+        // to avoid yielding to the microtask queue. This prevents race conditions
+        // where mutable args (e.g. snapshots) could be modified before serialization.
+        if (this._initialized.getValue()) {
+            return this._doRemoteCall(channelName, method, args);
+        }
+
+        return this._whenReady().then(() => this._doRemoteCall(channelName, method, args));
+    }
+
+    private _doRemoteCall(channelName: string, method: string, args?: any): Promise<any> {
+        const sequence = ++this._lastRequestCounter;
+        const type = RequestType.CALL;
+        const request: IRPCRequest = { seq: sequence, type, channelName, method, args };
+        const pendingRequests = this._pendingRequests;
+
+        return new Promise((resolve, reject) => {
+            const responseHandler: IResponseHandler = {
+                handle(response: IRPCResponse) {
+                    switch (response.type) {
+                        case ResponseType.CALL_SUCCESS:
+                            pendingRequests.delete(sequence);
+                            resolve(response.data);
+                            break;
+                        case ResponseType.CALL_FAILURE:
+                            pendingRequests.delete(sequence);
+                            reject(response.data);
+                            break;
+                        default:
+                            throw new Error('[ChannelClient]: unknown response type!');
+                    }
+                },
+                dispose(error: Error) {
+                    reject(error);
+                },
+            };
+
+            this._pendingRequests.set(sequence, responseHandler);
+            this._sendRequest(request);
+        });
+    }
+
+    private _remoteSubscribe(channelName: string, method: string, args?: any): Observable<any> {
+        return new Observable((subscriber) => {
+            let sequence: number = -1;
+
+            const doSubscribe = () => {
+                sequence = ++this._lastRequestCounter;
+                const type = RequestType.SUBSCRIBE;
+                const request: IRPCRequest = { seq: sequence, type, channelName, method, args };
+
+                const responseHandler: IResponseHandler = {
+                    handle(response: IRPCResponse) {
+                        switch (response.type) {
+                            case ResponseType.SUBSCRIBE_NEXT:
+                                subscriber.next(response.data);
+                                break;
+                            case ResponseType.SUBSCRIBE_ERROR:
+                                subscriber.error(response.data);
+                                break;
+                            case ResponseType.SUBSCRIBE_COMPLETE:
+                                subscriber.complete();
+                                break;
+                            default:
+                                throw new Error('[ChannelClient]: unknown response type!');
+                        }
+                    },
+                    dispose(error: Error) {
+                        subscriber.error(error);
+                    },
+                };
+
+                this._pendingRequests.set(sequence, responseHandler);
+                this._sendRequest(request);
+            };
+
+            // Fast path: if the channel is already initialized, execute synchronously
+            // to avoid yielding to the microtask queue.
+            if (this._initialized.getValue()) {
+                doSubscribe();
+            } else {
+                this._whenReady().then(doSubscribe);
+            }
+
+            return () => {
+                if (sequence === -1) {
+                    return;
+                }
+
+                const cancelSubscriptionRequest: IRPCRequest = {
+                    type: RequestType.UNSUBSCRIBE,
+                    seq: sequence,
+                    channelName,
+                    method,
+                };
+                this._sendRequest(cancelSubscriptionRequest);
+            };
+        });
+    }
+
+    private _sendRequest(request: IRPCRequest): void {
+        this._protocol.send(request);
+    }
+
+    private _onMessage(response: IRPCResponse): void {
+        const { type: responseType, seq } = response;
+        switch (responseType) {
+            case ResponseType.INITIALIZE:
+                this._initialized.next(true);
+                break;
+            case ResponseType.CALL_SUCCESS:
+            case ResponseType.CALL_FAILURE:
+            case ResponseType.SUBSCRIBE_NEXT:
+            case ResponseType.SUBSCRIBE_COMPLETE:
+            case ResponseType.SUBSCRIBE_ERROR: {
+                const { _pendingRequests } = this;
+                _pendingRequests.get(seq)?.handle(response);
+                responseType !== ResponseType.SUBSCRIBE_NEXT && _pendingRequests.delete(seq);
+                break;
+            }
+        }
+    }
+}
+
+export class ChannelServer extends RxDisposable implements IChannelServer {
+    private _channels = new Map<string, IChannel>();
+
+    private _subscriptions = new Map<number, Subscription>();
+
+    constructor(private readonly _protocol: IMessageProtocol) {
+        super();
+
+        this._protocol.onMessage.pipe(takeUntil(this.dispose$)).subscribe((message) => this._onRequest(message));
+        this._sendInitialize();
+    }
+
+    override dispose(): void {
+        super.dispose();
+
+        for (const subscription of this._subscriptions.values()) {
+            subscription.unsubscribe();
+        }
+        this._subscriptions.clear();
+        this._channels.clear();
+    }
+
+    registerChannel(channelName: string, channel: IChannel): void {
+        this._channels.set(channelName, channel);
+    }
+
+    private _onRequest(request: IRPCRequest): void {
+        switch (request.type) {
+            case RequestType.REQUEST_INITIALIZATION:
+                this._sendInitialize();
+                break;
+            case RequestType.CALL:
+                this._onMethodCall(request);
+                break;
+            case RequestType.SUBSCRIBE:
+                this._onSubscribe(request);
+                break;
+            case RequestType.UNSUBSCRIBE:
+                this._onUnsubscribe(request);
+                break;
+            default:
+                break;
+        }
+    }
+
+    private _sendInitialize(): void {
+        this._sendResponse({ seq: -1, type: ResponseType.INITIALIZE });
+    }
+
+    private _onMethodCall(request: IRPCRequest): void {
+        const { channelName, method, args } = request;
+        const channel = this._channels.get(channelName);
+
+        let promise: Promise<any>;
+        try {
+            if (!channel) {
+                throw new Error(`[ChannelServer]: Channel ${channelName} not found!`);
+            }
+            promise = args ? channel.call(method, args) : channel.call(method);
+        } catch (err: unknown) {
+            promise = Promise.reject(err instanceof Error ? err : new Error(String(err)));
+        }
+
+        promise
+            .then((data) => {
+                this._sendResponse({ seq: request.seq, type: ResponseType.CALL_SUCCESS, data });
+            })
+            .catch((err) => {
+                if (err instanceof Error) {
+                    this._sendResponse({ seq: request.seq, type: ResponseType.CALL_FAILURE, data: err.message });
+                } else {
+                    this._sendResponse({ seq: request.seq, type: ResponseType.CALL_FAILURE, data: String(err) });
+                }
+            });
+    }
+
+    private _onSubscribe(request: IRPCRequest): void {
+        const { channelName, seq } = request;
+        const channel = this._channels.get(channelName);
+
+        try {
+            if (!channel) {
+                throw new Error(`[ChannelServer]: Channel ${channelName} not found!`);
+            }
+
+            const observable = channel.subscribe(request.method, request.args);
+            const subscription = observable.subscribe({
+                next: (data) => {
+                    this._sendResponse({ seq, type: ResponseType.SUBSCRIBE_NEXT, data });
+                },
+                error: (err) => {
+                    this._sendResponse({ seq, type: ResponseType.SUBSCRIBE_ERROR, data: err.message });
+                    this._sendResponse({ seq, type: ResponseType.SUBSCRIBE_COMPLETE });
+                },
+                complete: () => {
+                    this._sendResponse({ seq, type: ResponseType.SUBSCRIBE_COMPLETE });
+                },
+            });
+
+            this._subscriptions.set(request.seq, subscription);
+        } catch (err: unknown) {
+            if (err instanceof Error) {
+                this._sendResponse({ seq: request.seq, type: ResponseType.SUBSCRIBE_ERROR, data: err.message });
+            } else {
+                this._sendResponse({ seq: request.seq, type: ResponseType.SUBSCRIBE_ERROR, data: String(err) });
+            }
+        }
+    }
+
+    private _onUnsubscribe(request: IRPCRequest): void {
+        const subscription = this._subscriptions.get(request.seq);
+        if (subscription) {
+            subscription.unsubscribe();
+            this._subscriptions.delete(request.seq);
+        }
+    }
+
+    private _sendResponse(response: IRPCResponse): void {
+        this._protocol.send(response);
+    }
+}