diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java index d7a853982eb..d9d4e77d8e2 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java @@ -153,7 +153,7 @@ String getKey() { // Read requests whose direct reply and failures carry the request msgId back to the sender. // Broadcasts of the same ops stay uncorrelated. private static final Set MSG_ID_CORRELATED_READS = Collections.unmodifiableSet( - new HashSet<>(Arrays.asList(OP.GET_NOTE, OP.NOTE_REVISION))); + new HashSet<>(Arrays.asList(OP.GET_NOTE, OP.RELOAD_NOTE, OP.NOTE_REVISION))); private final ExecutorService executorService = Executors.newFixedThreadPool(10); @@ -949,11 +949,12 @@ private void reloadNote(NotebookSocket conn, ServiceContext context, Message fro return; } getNotebookService().getNote(noteId, true, context, - new WebSocketServiceCallback(conn) { + new WebSocketServiceCallback(conn, fromMessage.msgId) { @Override public void onSuccess(Note note, ServiceContext context) throws IOException { connectionManager.addNoteConnection(note.getId(), conn); - conn.send(serializeMessage(new Message(OP.NOTE).put("note", note))); + conn.send(serializeMessage( + new Message(OP.NOTE).withMsgId(fromMessage.msgId).put("note", note))); updateAngularObjectRegistry(conn, note); sendAllAngularObjects(note, context.getAutheInfo().getUser(), conn); } @@ -2288,6 +2289,7 @@ private void removeNoteAngularObject(String noteId, AngularObject angularObject, connectionManager.broadcast(noteId, new Message(OP.ANGULAR_OBJECT_REMOVE) .put("name", angularObject.getName()) + .put("interpreterGroupId", interpreterGroupId) .put("noteId", angularObject.getNoteId()) .put("paragraphId", angularObject.getParagraphId())); break; diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java index 6aae0af52a1..31880b677ce 100644 --- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java +++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java @@ -85,6 +85,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.slf4j.Logger; @@ -841,6 +843,58 @@ void unbindAngularObjectFromRemoteForParagraphs() throws Exception { } } + @Test + void interpreterRemovalPreservesGroupAndRoutesAllScopes() throws Exception { + final String noteId = "angular-removal-note"; + final String otherNoteId = "angular-removal-other-note"; + final String groupId = "angular-setting:session"; + final Notebook mockedNotebook = mock(Notebook.class, RETURNS_DEEP_STUBS); + when(mockedNotebook.getInterpreterSettingManager().getSettingIds()) + .thenReturn(Collections.singletonList("angular-setting")); + final NoteInfo noteInfo = mock(NoteInfo.class); + final NoteInfo otherNoteInfo = mock(NoteInfo.class); + when(noteInfo.getId()).thenReturn(noteId); + when(otherNoteInfo.getId()).thenReturn(otherNoteId); + when(mockedNotebook.getNotesInfo()).thenReturn(Arrays.asList(noteInfo, otherNoteInfo)); + final NotebookSocket conn = mock(NotebookSocket.class); + final NotebookSocket otherConn = mock(NotebookSocket.class); + notebookServer.getConnectionManager().noteSocketMap.put(noteId, Collections.singleton(conn)); + notebookServer.getConnectionManager().noteSocketMap.put(otherNoteId, Collections.singleton(otherConn)); + + try { + notebookServer.setNotebook(() -> mockedNotebook); + for (String paragraphId : Arrays.asList("paragraph", null)) { + AngularObject object = AngularObjectBuilder.build("value", "old", noteId, paragraphId); + + notebookServer.onRemoveAngularObject(groupId, object); + + Message expected = new Message(OP.ANGULAR_OBJECT_REMOVE) + .put("name", "value") + .put("interpreterGroupId", groupId) + .put("noteId", noteId) + .put("paragraphId", paragraphId); + verify(conn).send(notebookServer.serializeMessage(expected)); + } + verify(otherConn, never()).send(anyString()); + reset(conn); + + AngularObject globalObject = AngularObjectBuilder.build("value", "old", null, null); + notebookServer.onRemoveAngularObject(groupId, globalObject); + + Message expectedGlobal = new Message(OP.ANGULAR_OBJECT_REMOVE) + .put("name", "value") + .put("interpreterGroupId", groupId) + .put("noteId", null) + .put("paragraphId", null); + verify(conn).send(notebookServer.serializeMessage(expectedGlobal)); + verify(otherConn).send(notebookServer.serializeMessage(expectedGlobal)); + } finally { + notebookServer.setNotebook(() -> NotebookServerTest.notebook); + notebookServer.getConnectionManager().noteSocketMap.remove(noteId); + notebookServer.getConnectionManager().noteSocketMap.remove(otherNoteId); + } + } + @Test void testCreateNoteWithDefaultInterpreterId() throws IOException { // create two sockets and open it @@ -1282,14 +1336,15 @@ void setNoteRevisionIncludesNoteId() throws IOException { } } - @Test - void getNoteReplyCarriesTheRequestMsgId() throws IOException { + @ParameterizedTest + @EnumSource(value = OP.class, names = {"GET_NOTE", "RELOAD_NOTE"}) + void noteReadReplyCarriesTheRequestMsgId(OP op) throws IOException { String noteId = notebook.createNote("read-reply-msg-id", anonymous); try { NotebookSocket socket = createWebSocket(); notebookServer.onMessage(socket, - new Message(OP.GET_NOTE).withMsgId("read-1").put("id", noteId).toJson()); + new Message(op).withMsgId("read-1").put("id", noteId).toJson()); Message reply = onlySent(socket, OP.NOTE); assertEquals("read-1", reply.msgId); @@ -1298,13 +1353,14 @@ void getNoteReplyCarriesTheRequestMsgId() throws IOException { } } - @Test - void getNoteWithoutMsgIdKeepsTheReplyUncorrelated() throws IOException { + @ParameterizedTest + @EnumSource(value = OP.class, names = {"GET_NOTE", "RELOAD_NOTE"}) + void noteReadWithoutMsgIdKeepsTheReplyUncorrelated(OP op) throws IOException { String noteId = notebook.createNote("read-reply-without-msg-id", anonymous); try { NotebookSocket socket = createWebSocket(); - notebookServer.onMessage(socket, new Message(OP.GET_NOTE).put("id", noteId).toJson()); + notebookServer.onMessage(socket, new Message(op).put("id", noteId).toJson()); String reply = onlySentPayload(socket, OP.NOTE); assertNull(notebookServer.deserializeMessage(reply).msgId); @@ -1314,12 +1370,13 @@ void getNoteWithoutMsgIdKeepsTheReplyUncorrelated() throws IOException { } } - @Test - void getNoteFailuresCarryTheRequestMsgId() throws IOException { + @ParameterizedTest + @EnumSource(value = OP.class, names = {"GET_NOTE", "RELOAD_NOTE"}) + void noteReadFailuresCarryTheRequestMsgId(OP op) throws IOException { NotebookSocket socket = createWebSocket(); notebookServer.onMessage(socket, - new Message(OP.GET_NOTE).withMsgId("read-missing").put("id", "missing-note").toJson()); + new Message(op).withMsgId("read-missing").put("id", "missing-note").toJson()); assertEquals("read-missing", onlySent(socket, OP.ERROR_INFO).msgId); @@ -1329,10 +1386,10 @@ void getNoteFailuresCarryTheRequestMsgId() throws IOException { reset(socket); notebookServer.onMessage(socket, - new Message(OP.GET_NOTE).withMsgId("read-forbidden").put("id", noteId).toJson()); + new Message(op).withMsgId("read-forbidden").put("id", noteId).toJson()); assertEquals("read-forbidden", onlySent(socket, OP.AUTH_INFO).msgId); - assertTrue(sentOps(socket).stream().noneMatch(op -> op == OP.NOTE)); + assertTrue(sentOps(socket).stream().noneMatch(sentOp -> sentOp == OP.NOTE)); } finally { notebook.removeNote(noteId, anonymous); } @@ -1381,6 +1438,12 @@ void readRequestExceptionsReachTheSenderWithTheRequestMsgId() throws IOException assertEquals("read-throws", onlySent(socket, OP.ERROR_INFO).msgId); + reset(socket); + notebookServer.onMessage(socket, + new Message(OP.RELOAD_NOTE).withMsgId("reload-throws").put("id", 42).toJson()); + + assertEquals("reload-throws", onlySent(socket, OP.ERROR_INFO).msgId); + reset(socket); notebookServer.onMessage(socket, new Message(OP.NOTE_REVISION).withMsgId("revision-throws") .put("noteId", 42).put("revisionId", "some-revision").toJson()); @@ -1392,13 +1455,15 @@ void readRequestExceptionsReachTheSenderWithTheRequestMsgId() throws IOException void unrelatedFailuresAndNoteBroadcastsStayUncorrelated() throws IOException { NotebookSocket socket = createWebSocket(); notebookServer.onMessage(socket, - new Message(OP.RELOAD_NOTE).withMsgId("reload-missing").put("id", "missing-note").toJson()); + new Message(OP.PARAGRAPH_CLEAR_ALL_OUTPUT) + .withMsgId("clear-missing").put("id", "missing-note").toJson()); assertNull(onlySent(socket, OP.ERROR_INFO).msgId); reset(socket); notebookServer.onMessage(socket, - new Message(OP.RELOAD_NOTE).withMsgId("reload-throws").put("id", 42).toJson()); + new Message(OP.PARAGRAPH_CLEAR_ALL_OUTPUT) + .withMsgId("clear-throws").put("id", 42).toJson()); assertNull(onlySent(socket, OP.ERROR_INFO).msgId); diff --git a/zeppelin-web-angular/e2e/tests/notebook/paragraph/angular-object-binding.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/paragraph/angular-object-binding.spec.ts new file mode 100644 index 00000000000..5bd817fc106 --- /dev/null +++ b/zeppelin-web-angular/e2e/tests/notebook/paragraph/angular-object-binding.spec.ts @@ -0,0 +1,178 @@ +/* + * 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 { randomUUID } from 'node:crypto'; +import { expect, test } from '@playwright/test'; +import type { Interpreter } from '../../../../src/app/interfaces/interpreter'; +import { NotebookActionBarPage } from '../../../models/notebook-action-bar-page'; +import { NotebookKeyboardPage } from '../../../models/notebook-keyboard-page'; +import { + addPageAnnotationBeforeEach, + performLoginIfRequired, + PAGES, + setParagraphText, + waitForZeppelinReady +} from '../../../utils'; + +test.describe('Notebook-scoped Angular objects', () => { + addPageAnnotationBeforeEach(PAGES.WORKSPACE.NOTEBOOK_PARAGRAPH); + addPageAnnotationBeforeEach(PAGES.WORKSPACE.SHARE_RESULT); + + test('displays retained values, synchronizes both directions, and removes the binding', async ({ page }) => { + await page.goto('/#/'); + await waitForZeppelinReady(page); + await performLoginIfRequired(page); + const created = await page.request.post('/api/notebook', { + data: { + notePath: `E2E_TEST_FOLDER/AngularObject_${randomUUID()}`, + defaultInterpreterGroup: 'python', + addingEmptyParagraph: false, + paragraphs: [{ text: '%python' }, { text: '%python' }] + } + }); + expect(created.ok()).toBeTruthy(); + const noteId = (await created.json()).body as string; + + try { + const note = await page.request.get(`/api/notebook/${noteId}`); + expect(note.ok()).toBeTruthy(); + const [model, output] = (await note.json()).body.paragraphs; + const notebook = new NotebookKeyboardPage(page); + const value = page.getByLabel('Notebook model', { exact: true }); + const runPython = async (code: string, paragraphId = model.id) => { + await setParagraphText(page, noteId, paragraphId, `%python\n${code}`); + const response = await page.request.post(`/api/notebook/run/${noteId}/${paragraphId}`); + expect(response.ok()).toBeTruthy(); + const result = (await response.json()).body; + expect(result.code, JSON.stringify(result)).toBe('SUCCESS'); + }; + const template = `{{ notebookModel }} +`; + await page.goto(`/#/notebook/${noteId}`); + await expect(notebook.getParagraphByIndex(1)).toBeVisible(); + + // Explicit note ID: the Python wrapper's default bind is not notebook-scoped. + await runPython(`z.angularBind('notebookModel', 'initial', '${noteId}')`); + await expect(value).toHaveCount(0); + await runPython(`print(${JSON.stringify(`%angular\n${template}`)})`, output.id); + await expect(value).toHaveText('initial'); + + await runPython(`z.angularBind('notebookModel', 'from-server', '${noteId}')`); + await expect(value).toHaveText('from-server'); + + await setParagraphText(page, noteId, model.id, `%python\nprint('SERVER-MODEL=' + z.angular('notebookModel'))`); + await page.getByRole('button', { name: 'Update model', exact: true }).click(); + await expect(value).toHaveText('from-ui'); + // Send the read through the same WebSocket as the edit to preserve message order. + await notebook.focusParagraphHost(0); + await notebook.pressRunParagraph(); + await expect(notebook.getParagraphByIndex(0).getByText('SERVER-MODEL=from-ui', { exact: true })).toBeVisible(); + + // The Python wrapper calls a private overload; use the public Java context API. + await runPython(`z.z.angularUnbind('notebookModel')`); + await expect(value).toBeEmpty(); + } finally { + const deleted = await page.request.delete(`/api/notebook/${noteId}`); + expect(deleted.ok()).toBeTruthy(); + } + }); + + test('reconciles bindings when reloading a note after interpreter restart', async ({ page }) => { + await page.goto('/#/'); + await waitForZeppelinReady(page); + await performLoginIfRequired(page); + const settingsResponse = await page.request.get('/api/interpreter/setting'); + expect(settingsResponse.ok()).toBeTruthy(); + const settings = (await settingsResponse.json()).body as Interpreter[]; + const python = settings.find(setting => setting.name === 'python'); + if (!python) { + throw new Error('The Python interpreter setting is required'); + } + + const name = `angular_reload_${randomUUID()}`; + const settingResponse = await page.request.post('/api/interpreter/setting', { + data: { + name, + group: python.group, + properties: python.properties, + dependencies: python.dependencies, + option: { ...python.option, isExistingProcess: false } + } + }); + expect(settingResponse.ok()).toBeTruthy(); + const settingId = ((await settingResponse.json()).body as Interpreter).id; + + try { + const created = await page.request.post('/api/notebook', { + data: { + notePath: `E2E_TEST_FOLDER/AngularReload_${randomUUID()}`, + defaultInterpreterGroup: name, + addingEmptyParagraph: false, + paragraphs: [{ text: `%${name}` }, { text: `%${name}` }] + } + }); + expect(created.ok()).toBeTruthy(); + const noteId = (await created.json()).body as string; + + try { + const note = await page.request.get(`/api/notebook/${noteId}`); + expect(note.ok()).toBeTruthy(); + const [model, output] = (await note.json()).body.paragraphs; + const runPython = async (code: string, paragraphId = model.id) => { + await setParagraphText(page, noteId, paragraphId, `%${name}\n${code}`); + const response = await page.request.post(`/api/notebook/run/${noteId}/${paragraphId}`); + expect(response.ok()).toBeTruthy(); + const result = (await response.json()).body; + expect(result.code, JSON.stringify(result)).toBe('SUCCESS'); + }; + const value = page.getByLabel('Reloaded model', { exact: true }); + const template = `{{ reloadModel }} +`; + const renderAngular = `print(${JSON.stringify(`%angular\n${template}`)})`; + await page.goto(`/#/notebook/${noteId}`); + await runPython(`z.angularBind('reloadModel', 'before-restart', '${noteId}')`); + await runPython(renderAngular, output.id); + await expect(value).toHaveText('before-restart'); + // Persist the completed output without rerunning it. + await setParagraphText(page, noteId, output.id, `%${name}\n${renderAngular}`); + + const restarted = await page.request.put(`/api/interpreter/setting/restart/${settingId}`, { + data: { noteId } + }); + expect(restarted.ok()).toBeTruthy(); + await runPython(`assert z.angular('reloadModel') is None`); + await expect(value).toHaveText('before-restart'); + + await new NotebookActionBarPage(page).reloadButton.click(); + await expect(value).toHaveCount(1); + await expect(value).toBeEmpty(); + + await runPython(`z.angularBind('reloadModel', 'after-restart', '${noteId}')`); + await expect(value).toHaveText('after-restart'); + await setParagraphText(page, noteId, model.id, `%${name}\nprint('RELOADED-MODEL=' + z.angular('reloadModel'))`); + await page.getByRole('button', { name: 'Update reloaded model', exact: true }).click(); + const notebook = new NotebookKeyboardPage(page); + await notebook.focusParagraphHost(0); + await notebook.pressRunParagraph(); + await expect( + notebook.getParagraphByIndex(0).getByText('RELOADED-MODEL=from-ui', { exact: true }) + ).toBeVisible(); + } finally { + const deleted = await page.request.delete(`/api/notebook/${noteId}`); + expect(deleted.ok()).toBeTruthy(); + } + } finally { + const deleted = await page.request.delete(`/api/interpreter/setting/${settingId}`); + expect(deleted.ok()).toBeTruthy(); + } + }); +}); diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/angular-object-values.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/angular-object-values.spec.ts index 59ac8395636..7fa54c4c037 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/angular-object-values.spec.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/angular-object-values.spec.ts @@ -10,9 +10,14 @@ * limitations under the License. */ -import { afterEach, describe, expect, expectTypeOf, it, vi } from 'vitest'; +import { afterEach, assertType, describe, expect, expectTypeOf, it, vi } from 'vitest'; import { OP } from './interfaces/message-operator.interface'; +import { + AngularObjectRemove, + AngularObjectUpdate, + AngularObjectUpdated +} from './interfaces/message-paragraph.interface'; import { Message } from './message'; afterEach(() => vi.restoreAllMocks()); @@ -22,6 +27,45 @@ it('accepts arbitrary values through the public Angular object helpers', () => { expectTypeOf[2]>().toEqualTypeOf(); }); +it('accepts notebook and global wire scopes without requiring Gson-omitted fields', () => { + const updates: AngularObjectUpdate[] = [ + { + noteId: 'route-note', + interpreterGroupId: 'group', + angularObject: { name: 'value', object: false, noteId: 'route-note' } + }, + { + noteId: 'route-note', + paragraphId: null, + interpreterGroupId: 'group', + angularObject: { name: 'value', object: false, noteId: null, paragraphId: null } + }, + { + noteId: 'route-note', + interpreterGroupId: 'group', + angularObject: { name: 'value', object: false } + } + ]; + const removals: AngularObjectRemove[] = [ + { noteId: 'route-note', name: 'value', interpreterGroupId: 'group' }, + { noteId: 'route-note', paragraphId: null, name: 'value', interpreterGroupId: 'group' }, + { name: 'value', interpreterGroupId: 'group' }, + { noteId: null, name: 'value', interpreterGroupId: 'group' }, + { ...updates[0], angularObject: updates[0].angularObject }, + { ...updates[1], angularObject: updates[1].angularObject }, + { ...updates[2], angularObject: updates[2].angularObject } + ]; + const outgoing: AngularObjectUpdated[] = [ + { noteId: 'route-note', name: 'value', value: false, interpreterGroupId: 'group' }, + { noteId: 'route-note', paragraphId: null, name: 'value', value: false, interpreterGroupId: 'group' } + ]; + + assertType(updates); + assertType(removals); + assertType(outgoing); + expectTypeOf[1]>().toEqualTypeOf(); +}); + describe('Angular object send values', () => { it.each([false, 42, null, ['a', false], { count: 0, enabled: false }])( 'preserves %j in update and bind payloads', diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts index 88a4c06f095..2b98bfc6910 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts @@ -234,21 +234,21 @@ export interface MoveParagraph { export interface AngularObjectUpdated { noteId: string; - paragraphId: string; + paragraphId?: string | null; name: string; value: unknown; interpreterGroupId: string; } interface AngularObjectRemoveContext { - noteId: string; - paragraphId: string; + noteId?: string | null; + paragraphId?: string | null; angularObject?: { name: string; object: unknown; // Omitted for global (noteId) and note (paragraphId) scoped objects. - noteId?: string; - paragraphId?: string; + noteId?: string | null; + paragraphId?: string | null; }; interpreterGroupId?: string; } @@ -258,13 +258,13 @@ export type AngularObjectRemove = AngularObjectRemoveContext & export interface AngularObjectUpdate { noteId: string; - paragraphId: string; + paragraphId?: string | null; interpreterGroupId: string; angularObject: { name: string; object: unknown; - noteId: string; - paragraphId: string; + noteId?: string | null; + paragraphId?: string | null; }; } diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/notebook-wire-fields.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/notebook-wire-fields.spec.ts index f35ec59b52e..55d7217ac10 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/notebook-wire-fields.spec.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/notebook-wire-fields.spec.ts @@ -19,15 +19,6 @@ it('separates received wire fields from backward-compatible import input', () => expectTypeOf().toHaveProperty('progress').toEqualTypeOf(); expectTypeOf>().toHaveProperty('version').toEqualTypeOf(); expectTypeOf().toHaveProperty('completionKey').toEqualTypeOf(); - expectTypeOf().toHaveProperty('angularObject').toEqualTypeOf< - | { - name: string; - object: unknown; - noteId?: string; - paragraphId?: string; - } - | undefined - >(); expectTypeOf().toHaveProperty('interpreterGroupId').toEqualTypeOf(); const legacyImportParagraph = { text: '%md legacy import', diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 79e27e3e9a6..c6b3532480b 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -317,7 +317,7 @@ export class Message { angularObjectUpdate( noteId: string, - paragraphId: string, + paragraphId: string | null | undefined, name: string, value: unknown, interpreterGroupId: string diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/angular-context-manager.ts b/zeppelin-web-angular/src/app/core/paragraph-base/angular-context-manager.ts index 7fc1afd9cff..7122fd08d3c 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/angular-context-manager.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/angular-context-manager.ts @@ -12,18 +12,32 @@ import { Observable } from 'rxjs'; +export interface AngularObjectBinding { + noteId: string | null; + paragraphId: string | null; + interpreterGroupId?: string; +} + export interface AngularContext { + noteId: string; paragraphId: string; key: string; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - value: any; + value: unknown; emit: boolean; set: boolean; + binding?: AngularObjectBinding; } export interface AngularContextManager { - setContextValue(key: string, value: unknown, paragraphId: string, emit?: boolean): void; - unsetContextValue(key: string, paragraphId: string, emit?: boolean): unknown; + registerParagraphBindings(noteId: string, paragraphId: string): void; + setContextValue( + key: string, + value: unknown, + paragraphId: string, + emit?: boolean, + binding?: AngularObjectBinding + ): void; + unsetContextValue(key: string, paragraphId: string, emit?: boolean, binding?: AngularObjectBinding): void; contextChanged(): Observable; runParagraphAction(): Observable; } diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts index ca3ec2c9223..201085ae0a6 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts @@ -11,7 +11,7 @@ */ import { ChangeDetectorRef } from '@angular/core'; -import { AngularObjectRemove, DatasetType, Message, OP, ParagraphItem, ParagraphState } from '@zeppelin/sdk'; +import { DatasetType, Message, OP, ParagraphItem, ParagraphState } from '@zeppelin/sdk'; import { EMPTY } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; @@ -22,7 +22,6 @@ class TestParagraph extends ParagraphBase { changeColWidth = vi.fn(); updateParagraphResult = vi.fn(); protected currentNoteId = 'note'; - unsetContextValue = vi.fn(); hydrate(snapshot: ParagraphItem) { this.setParagraphSnapshot(snapshot); @@ -32,7 +31,7 @@ class TestParagraph extends ParagraphBase { super( { receive: () => EMPTY, receiveEnvelope: () => EMPTY } as unknown as Message, { isParagraphRunning: item => item.status === 'RUNNING', isEntireNoteRunning: () => false }, - { unsetContextValue: (...args) => this.unsetContextValue(...args) } as AngularContextManager, + { registerParagraphBindings: vi.fn() } as unknown as AngularContextManager, { markForCheck: vi.fn() } as unknown as ChangeDetectorRef ); this.paragraph = paragraph; @@ -74,25 +73,6 @@ const beginOutput = (component: TestParagraph) => const appendOutput = (component: TestParagraph) => component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 0, data: 'second\n' }); -describe('ParagraphBase Angular object removal', () => { - const payloads: AngularObjectRemove[] = [ - { noteId: 'note', paragraphId: 'A', name: 'bound' }, - { noteId: 'note', paragraphId: 'A', angularObject: { name: 'bound', object: 42 }, interpreterGroupId: 'group' } - ]; - - it.each(payloads)('removes the binding for %j', data => { - const component = new TestParagraph(paragraph('A')); - component.angularObjectRemove(data); - expect(component.unsetContextValue).toHaveBeenCalledWith('bound', 'A', false); - }); - - it('does not remove a binding belonging to another paragraph', () => { - const component = new TestParagraph(paragraph('A')); - component.angularObjectRemove({ noteId: 'note', paragraphId: 'other', name: 'bound' }); - expect(component.unsetContextValue).not.toHaveBeenCalled(); - }); -}); - describe('ParagraphBase streaming state isolation', () => { it.each(['append', 'update'])('ignores %s from another note with the same paragraph ID', kind => { const component = new TestParagraph(paragraph('A')); @@ -307,6 +287,7 @@ const createSaveTestParagraph = (text: string, dirtyText?: string): SaveTestPara { receive: () => EMPTY, receiveEnvelope: () => EMPTY } as unknown as Message, { isParagraphRunning: () => false, isEntireNoteRunning: () => false }, { + registerParagraphBindings: vi.fn(), setContextValue: vi.fn(), unsetContextValue: vi.fn(), contextChanged: () => EMPTY, diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts index 75a2a81dba3..aabdd3ecb0d 100644 --- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts +++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts @@ -14,6 +14,7 @@ import { ChangeDetectorRef } from '@angular/core'; import { AngularObjectRemove, AngularObjectUpdate, + getAngularObjectRemovalName, GraphConfig, Message, MessageReceiveDataTypeMap, @@ -24,7 +25,6 @@ import { ParagraphEditorSetting, ParagraphSettings, ParagraphStates, - getAngularObjectRemovalName, isTerminalParagraphState, ParagraphItem, ParagraphIResultsMsgItem, @@ -217,10 +217,22 @@ export abstract class ParagraphBase extends MessageListenersManager { if (!this.paragraph) { throw new Error('paragraph is not defined'); } - if (data.paragraphId === this.paragraph.id) { - const { name, object } = data.angularObject; - this.angularContextManager.setContextValue(name, object, data.paragraphId, false); + const noteId = this.currentNoteId; + const object = data.angularObject; + const paragraphId = object.paragraphId ?? data.paragraphId ?? null; + if ( + !noteId || + data.noteId !== noteId || + (object.noteId != null && object.noteId !== noteId) || + (paragraphId !== null && paragraphId !== this.paragraph.id) + ) { + return; } + this.angularContextManager.setContextValue(object.name, object.object, this.paragraph.id, false, { + noteId: object.noteId ?? null, + paragraphId, + interpreterGroupId: data.interpreterGroupId + }); } @MessageListener(OP.ANGULAR_OBJECT_REMOVE) @@ -228,11 +240,24 @@ export abstract class ParagraphBase extends MessageListenersManager { if (!this.paragraph) { throw new Error('paragraph is not defined'); } - if (data.paragraphId === this.paragraph.id) { - const name = getAngularObjectRemovalName(data); - if (name !== undefined) { - this.angularContextManager.unsetContextValue(name, data.paragraphId, false); - } + const noteId = this.currentNoteId; + const objectNoteId = data.angularObject ? (data.angularObject.noteId ?? null) : (data.noteId ?? null); + const paragraphId = data.angularObject?.paragraphId ?? data.paragraphId ?? null; + if ( + !noteId || + (data.noteId != null && data.noteId !== noteId) || + (objectNoteId !== null && objectNoteId !== noteId) || + (paragraphId !== null && paragraphId !== this.paragraph.id) + ) { + return; + } + const name = getAngularObjectRemovalName(data); + if (name !== undefined) { + this.angularContextManager.unsetContextValue(name, this.paragraph.id, false, { + noteId: objectNoteId, + paragraphId, + interpreterGroupId: data.interpreterGroupId + }); } } @@ -460,6 +485,10 @@ export abstract class ParagraphBase extends MessageListenersManager { } protected setParagraphSnapshot(paragraph: ParagraphItem | undefined): void { this.paragraph = paragraph; + const noteId = this.currentNoteId; + if (paragraph && noteId) { + this.angularContextManager.registerParagraphBindings(noteId, paragraph.id); + } this.results = []; this.configs = {}; if (paragraph) { diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts index 62186de41f4..1d8f07926f6 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.spec.ts @@ -20,7 +20,6 @@ import { OP, type MessageReceiveDataTypeMap, type WebSocketMessage } from '@zepp import type { MessageService, - NgZService, NoteStatusService, NoteVarShareService, ReactFeatureService, @@ -28,6 +27,7 @@ import type { ThemeService, TicketService } from '@zeppelin/services'; +import { NgZService } from '../../../services/ng-z.service'; vi.mock('./paragraph/paragraph.component', () => ({ NotebookParagraphComponent: class NotebookParagraphComponent {} @@ -39,21 +39,22 @@ const createComponent = () => { const messageService = { receive: vi.fn(() => NEVER), receiveEnvelope: vi.fn(() => NEVER), - consumeLocalAddFocusMsgId: vi.fn(() => false) + consumeLocalAddFocusMsgId: vi.fn(() => false), + getInterpreterBindings: vi.fn() }; const router = { navigate: vi.fn(() => Promise.resolve(true)) } as unknown as Router; + const ngZService = new NgZService(); + const routeParams: Record = { noteId: 'note-b' }; const component = new NotebookComponent( messageService as unknown as MessageService, - {} as NgZService, + ngZService, { snapshot: { - params: { - noteId: 'note-b' - } + params: routeParams } } as unknown as ActivatedRoute, { @@ -62,17 +63,19 @@ const createComponent = () => { {} as NoteStatusService, {} as NoteVarShareService, {} as TicketService, - {} as SecurityService, + { getPermissions: () => NEVER } as unknown as SecurityService, router, - {} as Title, - {} as ThemeService, + { setTitle: vi.fn() } as unknown as Title, + { updateMonacoTheme: vi.fn() } as unknown as ThemeService, {} as ReactFeatureService ); return { component, router, - messageService + messageService, + ngZService, + routeParams }; }; @@ -194,4 +197,111 @@ describe('NotebookComponent', () => { expect(messageService.consumeLocalAddFocusMsgId).toHaveBeenCalledWith('local-add-msg'); }); + + it('clears models missing from a reconnected NOTE snapshot without disconnecting existing outputs', () => { + const { component, ngZService } = createComponent(); + const note = { + id: 'note-b', + name: 'Notebook', + config: {}, + paragraphs: [{ id: 'A', config: {} }] + } as NonNullable; + component.note = note; + ngZService.registerParagraphBindings(note.id, 'A'); + const context: Record = {}; + ngZService.bindParagraphContext('A', context); + const binding = { noteId: note.id, paragraphId: null, interpreterGroupId: 'spark' }; + ngZService.setContextValue('controlValue', 'deleted-while-disconnected', 'A', false, binding); + + component.getNote({ op: OP.NOTE, data: { note } }); + expect(context.controlValue).toBe('deleted-while-disconnected'); + + component.getNote({ op: OP.NOTE, msgId: 'reload-read', data: { note } }); + + expect(context.controlValue).toBeUndefined(); + ngZService.setContextValue('currentValue', 'replayed', 'A', false, binding); + expect(context.currentValue).toBe('replayed'); + expect(context.controlValue).toBeUndefined(); + }); + + it('cleans up removed paragraph state when a revision replaces the notebook snapshot', () => { + const { component, ngZService } = createComponent(); + component.note = { + id: 'note-b', + config: {}, + paragraphs: [{ id: 'removed', config: {} }] + } as NonNullable; + ngZService.registerParagraphBindings('note-b', 'removed'); + ngZService.setContextValue('localValue', 'old-paragraph', 'removed'); + + component.getNoteRevision({ + noteId: 'note-b', + revisionId: 'r1', + note: { ...component.note, paragraphs: [] } + }); + + ngZService.registerParagraphBindings('note-b', 'removed'); + const context: Record = {}; + ngZService.bindParagraphContext('removed', context); + expect(context.localValue).toBeUndefined(); + }); + + it('releases bindings outside the visible snapshot when switching a single-paragraph route', () => { + const { component, ngZService, routeParams } = createComponent(); + const note = { + id: 'note-b', + name: 'Notebook', + config: {}, + paragraphs: [ + { id: 'A', config: {} }, + { id: 'B', config: {} } + ] + } as NonNullable; + component.note = { ...note, paragraphs: [note.paragraphs[0]] }; + ngZService.registerParagraphBindings(note.id, 'A'); + ngZService.setContextValue('privateValue', 'previous-output', 'A'); + routeParams.paragraphId = 'B'; + + component.getNote({ op: OP.NOTE, msgId: 'single-paragraph-read', data: { note } }); + + expect(component.note?.paragraphs.map(paragraph => paragraph.id)).toEqual(['B']); + const staleOutput: Record = {}; + ngZService.bindParagraphContext('A', staleOutput); + ngZService.setContextValue('privateValue', 'late-old-output', 'A'); + expect(staleOutput.privateValue).toBeUndefined(); + }); + + it('preserves bindings while viewing a revision and reconciles them only on a live read', () => { + const { component, ngZService } = createComponent(); + const note = { + id: 'note-b', + name: 'Notebook', + config: {}, + paragraphs: [{ id: 'A', config: {} }] + } as NonNullable; + component.note = note; + ngZService.registerParagraphBindings(note.id, 'A'); + const context: Record = {}; + const release = ngZService.bindParagraphContext('A', context); + const binding = { noteId: note.id, paragraphId: null, interpreterGroupId: 'spark' }; + ngZService.setContextValue('controlValue', 'before-revision', 'A', false, binding); + ngZService.setContextValue('localValue', 'client-local', 'A'); + + component.getNoteRevision({ noteId: note.id, revisionId: 'r1', note }); + + expect(context.controlValue).toBe('before-revision'); + expect(context.localValue).toBe('client-local'); + release(); + const recreatedOutput: Record = {}; + ngZService.bindParagraphContext('A', recreatedOutput); + expect(recreatedOutput.controlValue).toBe('before-revision'); + expect(recreatedOutput.localValue).toBe('client-local'); + + component.getNote({ op: OP.NOTE, msgId: 'return-to-live', data: { note } }); + + expect(recreatedOutput.controlValue).toBeUndefined(); + expect(recreatedOutput.localValue).toBeUndefined(); + ngZService.setContextValue('controlValue', 'live-model', 'A', false, binding); + expect(recreatedOutput.controlValue).toBe('live-model'); + }); }); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index 21d761b99cf..0e3cce4244a 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -89,17 +89,21 @@ export class NotebookComponent extends MessageListenersManager implements OnInit useReactFooter = false; useReactAssistant = false; - @MessageListener(OP.NOTE) - getNote(data: MessageReceiveDataTypeMap[OP.NOTE]) { - const note = data.note; + @MessageEnvelopeListener(OP.NOTE) + getNote(message: WebSocketMessage) { + const note = message.data?.note; if (isNil(note)) { this.router.navigate(['/']).then(); } else { - this.removeParagraphFromNgZ(); - this.note = note; const { paragraphId } = this.activatedRoute.snapshot.params; + const nextNote = paragraphId ? this.cleanParagraphExcept(note, paragraphId) : note; + this.removeObsoleteParagraphBindings(nextNote); + // Read replies replay the registry; ordinary NOTE broadcasts do not. + if (message.msgId) { + this.ngZService.clearNoteContextValues(note.id); + } + this.note = nextNote; if (paragraphId) { - this.note = this.cleanParagraphExcept(this.note, paragraphId); this.initializeLookAndFeel(this.note); } else { this.initializeLookAndFeel(this.note); @@ -146,6 +150,7 @@ export class NotebookComponent extends MessageListenersManager implements OnInit return; } const definedNote = this.note; + this.ngZService.removeParagraphBindings(data.id); const paragraphIndex = definedNote.paragraphs.findIndex(p => p.id === data.id); definedNote.paragraphs = definedNote.paragraphs.filter((p, index) => index !== paragraphIndex); const adjustedCursorIndex = @@ -208,6 +213,7 @@ export class NotebookComponent extends MessageListenersManager implements OnInit if (isNil(note)) { this.router.navigate(['/']).then(); } else { + this.removeObsoleteParagraphBindings(note); this.note = note; this.initializeLookAndFeel(this.note); this.cdr.markForCheck(); @@ -525,11 +531,15 @@ export class NotebookComponent extends MessageListenersManager implements OnInit }); } - removeParagraphFromNgZ(): void { - if (this.note && Array.isArray(this.note.paragraphs)) { - this.note.paragraphs.forEach(p => { - this.ngZService.removeParagraph(p.id); - }); + removeObsoleteParagraphBindings(nextNote?: NonNullable): void { + if (this.note) { + const retainedIds = + nextNote?.id === this.note.id ? new Set(nextNote.paragraphs.map(p => p.id)) : new Set(); + for (const paragraph of this.note.paragraphs) { + if (!retainedIds.has(paragraph.id)) { + this.ngZService.removeParagraphBindings(paragraph.id); + } + } } } @@ -537,6 +547,7 @@ export class NotebookComponent extends MessageListenersManager implements OnInit super.ngOnDestroy(); this.killSaveTimer(); this.saveNote(); + this.removeObsoleteParagraphBindings(); this.destroy$.next(); this.destroy$.complete(); this.titleService.setTitle('Zeppelin'); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts index 72c293e9602..bd788f81890 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts @@ -640,9 +640,19 @@ export class NotebookParagraphComponent .contextChanged() .pipe(takeUntil(this.destroy$)) .subscribe(change => { - if (change.paragraphId === this.paragraph.id && change.emit) { + if (change.noteId === this.note.id && change.paragraphId === this.paragraph.id && change.emit) { if (change.set) { - this.messageService.angularObjectClientBind(this.note.id, change.key, change.value, change.paragraphId); + if (change.binding?.interpreterGroupId !== undefined) { + this.messageService.angularObjectUpdate( + this.note.id, + change.binding.paragraphId, + change.key, + change.value, + change.binding.interpreterGroupId + ); + } else { + this.messageService.angularObjectClientBind(this.note.id, change.key, change.value, change.paragraphId); + } } else { this.messageService.angularObjectClientUnbind(this.note.id, change.key, change.paragraphId); } diff --git a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts index 461c88955e0..f969fdd438b 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts @@ -12,9 +12,10 @@ import { QueryList } from '@angular/core'; import { convertToParamMap } from '@angular/router'; -import { DatasetType, Note, ParagraphItem, ParagraphState } from '@zeppelin/sdk'; +import { DatasetType, Note, OP, ParagraphItem, ParagraphState } from '@zeppelin/sdk'; import { BehaviorSubject, EMPTY, of } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { NgZService } from '../../../../services/ng-z.service'; vi.mock('@zeppelin/services', () => ({ MessageService: class {}, @@ -51,7 +52,7 @@ const snapshot = (noteId: string, status: ParagraphState, text: string): Note => const components: PublishedParagraphComponent[] = []; afterEach(() => components.splice(0).forEach(component => component.ngOnDestroy())); -const setup = (useReact = false) => { +const setup = (useReact = false, ngZService = new NgZService()) => { const params = new BehaviorSubject({ noteId: 'note-A', paragraphId: 'same-paragraph' }); const component = new PublishedParagraphComponent( { receive: () => EMPTY, receiveEnvelope: () => EMPTY, getNote: vi.fn() } as never, @@ -60,12 +61,12 @@ const setup = (useReact = false) => { {} as never, { isEnabled: () => useReact } as never, { isParagraphRunning: (p: ParagraphItem) => p.status === 'RUNNING' } as never, - {} as never, + ngZService, { markForCheck: vi.fn() } as never ); component.notebookParagraphResultComponents = new QueryList(); components.push(component); - return { component, params }; + return { component, params, ngZService }; }; const append = (component: PublishedParagraphComponent, noteId: string, data: string) => @@ -74,43 +75,110 @@ const append = (component: PublishedParagraphComponent, noteId: string, data: st describe('Published paragraph authoritative snapshots', () => { it('resumes from a same-note refreshed snapshot instead of cached output', () => { const { component } = setup(); - component.getNote(snapshot('note-A', 'RUNNING', 'old')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'old') }); append(component, 'note-A', ' tail'); - component.getNote(snapshot('note-A', 'RUNNING', 'fresh')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'fresh') }); append(component, 'note-A', ' next'); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'fresh next' }]); }); it('unfreezes terminal output when the reused route loads a running paragraph', () => { const { component, params } = setup(); - component.getNote(snapshot('note-A', 'FINISHED', 'A final')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'FINISHED', 'A final') }); append(component, 'note-A', ' ignored'); params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' }); - component.getNote(snapshot('note-B', 'RUNNING', 'B first')); + component.getNote({ op: OP.NOTE, data: snapshot('note-B', 'RUNNING', 'B first') }); append(component, 'note-B', ' second'); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B first second' }]); }); it('rejects stale NOTE and stream events while waiting for the new route snapshot', () => { const { component, params } = setup(); - component.getNote(snapshot('note-A', 'RUNNING', 'A first')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'A first') }); append(component, 'note-A', ' tail'); params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' }); - component.getNote(snapshot('note-A', 'RUNNING', 'stale A')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'stale A') }); append(component, 'note-A', ' stale tail'); append(component, 'note-B', ' before snapshot'); expect(component.results).toEqual([]); - component.getNote(snapshot('note-B', 'RUNNING', 'B first')); + component.getNote({ op: OP.NOTE, data: snapshot('note-B', 'RUNNING', 'B first') }); append(component, 'note-A', ' foreign'); append(component, 'note-B', ' second'); expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B first second' }]); }); + + it('does not hydrate a removed server model when a published view is reopened in the same application', () => { + const service = new NgZService(); + const first = setup(false, service); + first.component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'FINISHED', 'old output') }); + first.component.angularObjectUpdate({ + noteId: 'note-A', + paragraphId: 'same-paragraph', + interpreterGroupId: 'spark', + angularObject: { + name: 'controlValue', + object: 'deleted-while-away', + noteId: 'note-A', + paragraphId: 'same-paragraph' + } + }); + components.pop()!.ngOnDestroy(); + + const second = setup(false, service); + second.component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'FINISHED', 'saved output') }); + const context: Record = {}; + service.bindParagraphContext('same-paragraph', context); + + expect(context.controlValue).toBeUndefined(); + }); + + it('releases the previous published paragraph when the route changes within a notebook', () => { + const { component, params, ngZService } = setup(); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'FINISHED', 'first output') }); + ngZService.setContextValue('privateValue', 'previous-paragraph', 'same-paragraph'); + params.next({ noteId: 'note-A', paragraphId: 'next-paragraph' }); + const next = snapshot('note-A', 'FINISHED', 'next output'); + next.note!.paragraphs[0].id = 'next-paragraph'; + component.getNote({ op: OP.NOTE, data: next }); + const staleOutput: Record = {}; + ngZService.bindParagraphContext('same-paragraph', staleOutput); + ngZService.setContextValue('privateValue', 'late-old-output', 'same-paragraph'); + expect(staleOutput.privateValue).toBeUndefined(); + const currentOutput: Record = {}; + ngZService.bindParagraphContext('next-paragraph', currentOutput); + ngZService.setContextValue('privateValue', 'current-paragraph', 'next-paragraph'); + expect(currentOutput.privateValue).toBe('current-paragraph'); + }); + + it('preserves values on published NOTE broadcasts but clears missing models on an authoritative read', () => { + const { component, ngZService } = setup(); + const data = snapshot('note-A', 'FINISHED', 'output'); + component.getNote({ op: OP.NOTE, data }); + component.angularObjectUpdate({ + noteId: 'note-A', + interpreterGroupId: 'spark', + angularObject: { name: 'controlValue', object: 'old-model', noteId: 'note-A' } + }); + const context: Record = {}; + ngZService.bindParagraphContext('same-paragraph', context); + + component.getNote({ op: OP.NOTE, data }); + expect(context.controlValue).toBe('old-model'); + component.getNote({ op: OP.NOTE, msgId: 'published-read', data }); + expect(context.controlValue).toBeUndefined(); + component.angularObjectUpdate({ + noteId: 'note-A', + interpreterGroupId: 'spark', + angularObject: { name: 'controlValue', object: 'current-model', noteId: 'note-A' } + }); + expect(context.controlValue).toBe('current-model'); + }); }); describe('Published React streaming output', () => { it.each(['append', 'update'])('refreshes React props after a streaming %s', operation => { const { component } = setup(true); - component.getNote(snapshot('note-A', 'RUNNING', 'initial')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'initial') }); const initialProps = component.reactProps; if (operation === 'append') { append(component, 'note-A', ' tail'); @@ -134,7 +202,7 @@ describe('Published React streaming output', () => { component.reactFailed = true; const updateResult = vi.fn(); component.notebookParagraphResultComponents.reset([{ updateResult } as never]); - component.getNote(snapshot('note-A', 'RUNNING', 'initial')); + component.getNote({ op: OP.NOTE, data: snapshot('note-A', 'RUNNING', 'initial') }); append(component, 'note-A', ' tail'); expect(updateResult).toHaveBeenCalledWith(expect.anything(), { type: DatasetType.TEXT, data: 'initial tail' }); }); diff --git a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts index bf925cb5eca..d1558329bf9 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts @@ -19,13 +19,14 @@ import { ViewChildren } from '@angular/core'; import { ActivatedRoute, Router } from '@angular/router'; -import { publishedSymbol, MessageListener, ParagraphBase, Published } from '@zeppelin/core'; +import { publishedSymbol, MessageEnvelopeListener, MessageListener, ParagraphBase, Published } from '@zeppelin/core'; import { MessageReceiveDataTypeMap, OP, ParagraphConfigResult, ParagraphItem, - ParagraphIResultsMsgItem + ParagraphIResultsMsgItem, + WebSocketMessage } from '@zeppelin/sdk'; import { MessageService, NgZService, NoteStatusService, ReactFeatureService } from '@zeppelin/services'; import { isNil } from 'lodash'; @@ -57,6 +58,8 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis @ViewChildren(NotebookParagraphResultComponent) notebookParagraphResultComponents!: QueryList; + private readonly ngZService: NgZService; + protected get currentNoteId(): string | null { return this.noteId; } @@ -72,6 +75,7 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis cdr: ChangeDetectorRef ) { super(messageService, noteStatusService, ngZService, cdr); + this.ngZService = ngZService; this.activatedRoute.queryParamMap.subscribe(params => { this.useReact = this.reactFeature.isEnabled('publishedParagraph', params); }); @@ -80,6 +84,9 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis if (typeof params.noteId !== 'string') { throw new Error(`noteId path parameter should be string, but got ${typeof params.noteId} instead.`); } + if (this.paragraph) { + this.ngZService.removeParagraphBindings(this.paragraph.id); + } this.noteId = params.noteId; this.paragraphId = params.paragraphId!; this.setParagraphSnapshot(undefined); @@ -87,11 +94,18 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis }); } - @MessageListener(OP.NOTE) - getNote(data: MessageReceiveDataTypeMap[OP.NOTE]) { - const note = data.note; + @MessageEnvelopeListener(OP.NOTE) + getNote(message: WebSocketMessage) { + const note = message.data?.note; if (!isNil(note) && note.id === this.noteId) { - this.setParagraphSnapshot(note.paragraphs.find(p => p.id === this.paragraphId)); + const paragraph = note.paragraphs.find(p => p.id === this.paragraphId); + if (this.paragraph && this.paragraph.id !== paragraph?.id) { + this.ngZService.removeParagraphBindings(this.paragraph.id); + } + if (message.msgId) { + this.ngZService.clearNoteContextValues(note.id); + } + this.setParagraphSnapshot(paragraph); if (this.paragraph) { if (!this.paragraph.results) { this.showRunConfirmationModal(); @@ -113,6 +127,13 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis this.cdr.markForCheck(); } + override ngOnDestroy(): void { + super.ngOnDestroy(); + if (this.paragraph) { + this.ngZService.removeParagraphBindings(this.paragraph.id); + } + } + @MessageListener(OP.ERROR_INFO) handleError(data: MessageReceiveDataTypeMap[OP.ERROR_INFO]) { if (data.info && data.info.includes('404')) { diff --git a/zeppelin-web-angular/src/app/services/message.service.ts b/zeppelin-web-angular/src/app/services/message.service.ts index 7f714585105..5559cfbebe3 100644 --- a/zeppelin-web-angular/src/app/services/message.service.ts +++ b/zeppelin-web-angular/src/app/services/message.service.ts @@ -195,7 +195,7 @@ export class MessageService extends Message implements OnDestroy { angularObjectUpdate( noteId: string, - paragraphId: string, + paragraphId: string | null | undefined, name: string, value: unknown, interpreterGroupId: string diff --git a/zeppelin-web-angular/src/app/services/ng-z.service.spec.ts b/zeppelin-web-angular/src/app/services/ng-z.service.spec.ts new file mode 100644 index 00000000000..3745293ee0b --- /dev/null +++ b/zeppelin-web-angular/src/app/services/ng-z.service.spec.ts @@ -0,0 +1,453 @@ +/* + * 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 { ElementRef, QueryList } from '@angular/core'; +import { AngularObjectRemove, AngularObjectUpdate, Note, OP, ParagraphItem } from '@zeppelin/sdk'; +import { EMPTY, Subject } from 'rxjs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('@zeppelin/services', () => ({ + MessageService: class {}, + NgZService: class {}, + NoteStatusService: class {}, + NoteVarShareService: class {}, + ShortcutService: class {} +})); +// The hoisted module mock must load the real listener owner before unrelated view modules. +vi.mock('@zeppelin/core', async () => ({ + ...(await import('../core/paragraph-base/paragraph-base')), + ...(await import('../core/message-listener/message-listener')) +})); +vi.mock('../pages/workspace/share/result/result.component', () => ({ + NotebookParagraphResultComponent: class {} +})); +vi.mock('../pages/workspace/notebook/paragraph/code-editor/code-editor.component', () => ({ + NotebookParagraphCodeEditorComponent: class {} +})); + +import { NotebookParagraphComponent } from '../pages/workspace/notebook/paragraph/paragraph.component'; +import { NgZService } from './ng-z.service'; + +const disposables: Array<() => void> = []; +afterEach(() => disposables.splice(0).forEach(dispose => dispose())); + +const setup = () => { + const service = new NgZService(); + const updates = new Subject(); + const removals = new Subject(); + const message = { + receive: (op: OP) => + op === OP.ANGULAR_OBJECT_UPDATE ? updates : op === OP.ANGULAR_OBJECT_REMOVE ? removals : EMPTY, + receiveEnvelope: () => EMPTY, + angularObjectUpdate: vi.fn(), + angularObjectClientBind: vi.fn(), + angularObjectClientUnbind: vi.fn() + }; + const createParagraph = (id: string, noteId = 'note', revisionView = false) => { + const component = new NotebookParagraphComponent( + message as never, + new ElementRef(document.createElement('div')), + { set: vi.fn() } as never, + {} as never, + { forkByElement: () => ({ bindShortcut: () => EMPTY }) } as never, + { isParagraphRunning: () => false, isEntireNoteRunning: () => false } as never, + { markForCheck: vi.fn() } as never, + service + ); + component.paragraph = { + id, + text: '', + status: 'FINISHED', + config: {}, + settings: { params: {}, forms: {} } + } as ParagraphItem; + component.note = { id: noteId, paragraphs: [component.paragraph], config: {} } as NonNullable; + component.revisionView = revisionView; + component.notebookParagraphResultComponents = new QueryList(); + component.ngOnInit(); + disposables.push(() => component.ngOnDestroy()); + return component; + }; + const a = createParagraph('A'); + const b = createParagraph('B'); + disposables.push(() => service.ngOnDestroy()); + const update = (value: unknown, paragraphId?: string, group = 'spark') => + updates.next({ + noteId: 'note', + paragraphId, + interpreterGroupId: group, + angularObject: { name: 'controlValue', object: value, noteId: 'note', paragraphId } + }); + const bind = (paragraphId: string) => { + const context: Record = {}; + const release = service.bindParagraphContext(paragraphId, context); + return { context, release }; + }; + return { service, updates, removals, message, createParagraph, a, b, update, bind }; +}; + +describe('Paragraph Angular object binding', () => { + it.each(['notebook', 'global'] as const)('notifies each paragraph once per %s server update and removal', scope => { + const { service, updates, removals, createParagraph, bind, message } = setup(); + const paragraphIds = ['A', 'B', ...Array.from({ length: 18 }, (_, index) => `extra-${index}`)]; + paragraphIds.slice(2).forEach(id => createParagraph(id)); + const contexts = paragraphIds.map(id => bind(id).context); + const changes = vi.fn(); + const subscription = service.contextChanged().subscribe(changes); + disposables.push(() => subscription.unsubscribe()); + const angularObject = { + name: 'controlValue', + object: 'server', + noteId: scope === 'notebook' ? 'note' : null, + paragraphId: null + }; + updates.next({ noteId: 'note', interpreterGroupId: 'spark', angularObject }); + expect(contexts.map(context => context.controlValue)).toEqual(paragraphIds.map(() => 'server')); + expect(changes.mock.calls.map(([change]) => change.paragraphId)).toEqual(paragraphIds); + + changes.mockClear(); + contexts.forEach(context => (context.controlValue = 'template-local')); + updates.next({ noteId: 'note', interpreterGroupId: 'spark', angularObject }); + expect(contexts.map(context => context.controlValue)).toEqual(paragraphIds.map(() => 'server')); + expect(changes.mock.calls.map(([change]) => change.paragraphId)).toEqual(paragraphIds); + + changes.mockClear(); + removals.next({ noteId: 'note', interpreterGroupId: 'spark', angularObject }); + expect(contexts.map(context => context.controlValue)).toEqual(paragraphIds.map(() => undefined)); + expect(changes.mock.calls.map(([change]) => change.paragraphId)).toEqual(paragraphIds); + expect(message.angularObjectUpdate).not.toHaveBeenCalled(); + expect(message.angularObjectClientBind).not.toHaveBeenCalled(); + expect(message.angularObjectClientUnbind).not.toHaveBeenCalled(); + }); + + it('retains initial values received by paragraph listeners until outputs are created', () => { + const { update, bind } = setup(); + update('initial'); + update('latest'); + expect(bind('A').context.controlValue).toBe('latest'); + expect(bind('B').context.controlValue).toBe('latest'); + }); + + it('updates multiple outputs and hydrates replacements without retaining destroyed contexts', () => { + const { update, bind } = setup(); + const first = bind('A'); + const second = bind('A'); + update('initial'); + first.release(); + update('next'); + expect(first.context.controlValue).toBe('initial'); + expect(second.context.controlValue).toBe('next'); + expect(bind('A').context.controlValue).toBe('next'); + }); + + it('keeps paragraph overrides isolated from notebook updates and restores fallback after removal', () => { + const { update, bind, removals } = setup(); + update('private', 'A'); + update('shared'); + const a = bind('A').context; + const b = bind('B').context; + update('next-shared'); + expect(a.controlValue).toBe('private'); + expect(b.controlValue).toBe('next-shared'); + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBe('next-shared'); + }); + + it.each(['top-level', 'nested'] as const)('handles %s notebook removals without resurrecting values', shape => { + const { update, removals, bind } = setup(); + update('initial'); + const a = bind('A'); + const b = bind('B'); + removals.next( + shape === 'top-level' + ? { noteId: 'note', name: 'controlValue', interpreterGroupId: 'spark' } + : { + noteId: 'note', + interpreterGroupId: 'spark', + angularObject: { name: 'controlValue', object: 'initial', noteId: 'note' } + } + ); + expect(a.context.controlValue).toBeUndefined(); + expect(b.context.controlValue).toBeUndefined(); + a.release(); + expect(bind('A').context.controlValue).toBeUndefined(); + }); + + it('retains deletion received before output creation', () => { + const { update, removals, bind } = setup(); + update('initial'); + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(bind('A').context.controlValue).toBeUndefined(); + }); + + it('rejects foreign-note and different-paragraph messages in the paragraph listener', () => { + const { updates, removals, update, bind } = setup(); + update('private', 'A'); + const a = bind('A').context; + updates.next({ + noteId: 'other', + paragraphId: 'A', + interpreterGroupId: 'spark', + angularObject: { name: 'controlValue', object: 'foreign', noteId: 'other', paragraphId: 'A' } + }); + removals.next({ noteId: 'other', paragraphId: 'A', name: 'controlValue' }); + update('other-paragraph', 'B'); + removals.next({ noteId: 'note', paragraphId: 'B', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBe('private'); + }); + + it('uses the existing paragraph outbound subscriber to update the original notebook model only once', () => { + const { update, bind, service, message } = setup(); + update('initial', undefined, 'original-group'); + const a = bind('A').context; + const b = bind('B').context; + service.setContextValue('controlValue', false, 'A'); + expect(a.controlValue).toBe(false); + expect(b.controlValue).toBe(false); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'note', + null, + 'controlValue', + false, + 'original-group' + ); + expect(message.angularObjectClientBind).not.toHaveBeenCalled(); + update('remote', undefined, 'original-group'); + expect(a.controlValue).toBe('remote'); + expect(message.angularObjectUpdate).toHaveBeenCalledTimes(1); + }); + + it('preserves paragraph scope and interpreter group for edits without updating the notebook fallback', () => { + const { update, bind, service, message } = setup(); + update('shared', undefined, 'shared-group'); + update('private', 'A', 'private-group'); + const a = bind('A').context; + const b = bind('B').context; + service.setContextValue('controlValue', 0, 'A'); + expect(a.controlValue).toBe(0); + expect(b.controlValue).toBe('shared'); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'note', + 'A', + 'controlValue', + 0, + 'private-group' + ); + }); + + it('keeps a client-created paragraph value ahead of later shared updates and edits its original scope', () => { + const { update, bind, service, message } = setup(); + service.setContextValue('controlValue', 'client-private', 'A'); + update('shared'); + const a = bind('A').context; + expect(a.controlValue).toBe('client-private'); + expect(bind('B').context.controlValue).toBe('shared'); + service.setContextValue('controlValue', 'edited-private', 'A'); + expect(a.controlValue).toBe('edited-private'); + expect(message.angularObjectClientBind).toHaveBeenLastCalledWith('note', 'controlValue', 'edited-private', 'A'); + expect(message.angularObjectUpdate).not.toHaveBeenCalled(); + }); + + it('removes a client-created value even when another scope shares its name', () => { + const { service, update, bind, removals } = setup(); + service.setContextValue('controlValue', 'client-private', 'A'); + update('shared'); + const a = bind('A').context; + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBe('shared'); + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBeUndefined(); + expect(bind('A').context.controlValue).toBeUndefined(); + }); + + it('reconciles a server snapshot without losing live output registrations or retaining removed values', () => { + const { service, update, bind } = setup(); + update('old-server-value'); + const a = bind('A').context; + service.clearNoteContextValues('note'); + expect(a.controlValue).toBeUndefined(); + expect(bind('B').context.controlValue).toBeUndefined(); + update('new-server-value'); + expect(a.controlValue).toBe('new-server-value'); + }); + + it('binds existing shared values into a newly registered paragraph without a new server event', () => { + const { update, createParagraph, bind, service } = setup(); + update('shared'); + createParagraph('C'); + expect(bind('C').context.controlValue).toBe('shared'); + service.removeParagraphBindings('C'); + expect(bind('A').context.controlValue).toBe('shared'); + }); + + it.each([undefined, null])('treats a top-level noteId=%j removal as global, not notebook scoped', noteId => { + const { updates, removals, update, bind } = setup(); + updates.next({ + noteId: 'note', + interpreterGroupId: 'spark', + angularObject: { name: 'controlValue', object: 'global' } + }); + update('notebook'); + const a = bind('A').context; + removals.next({ noteId, name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBe('notebook'); + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(a.controlValue).toBeUndefined(); + }); + + it('preserves the original identity even when the initial bound value is undefined', () => { + const { service, update, bind, message } = setup(); + update(undefined); + expect(bind('A').context.controlValue).toBeUndefined(); + service.setContextValue('controlValue', 'edited', 'A'); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'note', + null, + 'controlValue', + 'edited', + 'spark' + ); + expect(message.angularObjectClientBind).not.toHaveBeenCalled(); + }); + + it('supports local z.set bindings in revision paragraphs without depending on NOTE messages', () => { + const { createParagraph, service, bind, message } = setup(); + createParagraph('revision', 'revision-note', true); + service.setContextValue('localValue', 'revision-input', 'revision'); + expect(bind('revision').context.localValue).toBe('revision-input'); + expect(message.angularObjectClientBind).toHaveBeenCalledExactlyOnceWith( + 'revision-note', + 'localValue', + 'revision-input', + 'revision' + ); + }); + + it('removes only the named interpreter-group binding and keeps the surviving model editable', () => { + const { update, removals, bind, service, message } = setup(); + update('first-model', 'A', 'group-one'); + update('second-model', 'A', 'group-two'); + const context = bind('A').context; + + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue', interpreterGroupId: 'group-one' }); + expect(context.controlValue).toBe('second-model'); + service.setContextValue('controlValue', 'edited-survivor', 'A'); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'note', + 'A', + 'controlValue', + 'edited-survivor', + 'group-two' + ); + + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue' }); + expect(context.controlValue).toBeUndefined(); + expect(bind('A').context.controlValue).toBeUndefined(); + }); + + it('does not guess the interpreter group when a removal matches multiple model identities', () => { + const { update, removals, bind } = setup(); + update('first-model', 'A', 'group-one'); + update('second-model', 'A', 'group-two'); + const context = bind('A').context; + const originalValue = context.controlValue; + + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue' }); + expect(context.controlValue).toBe(originalValue); + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue', interpreterGroupId: 'group-one' }); + expect(context.controlValue).toBe('second-model'); + removals.next({ noteId: 'note', paragraphId: 'A', name: 'controlValue', interpreterGroupId: 'group-two' }); + expect(context.controlValue).toBeUndefined(); + }); + + it('isolates retained values, clearing, and outbound edits between notebooks', () => { + const { createParagraph, update, updates, bind, service, message } = setup(); + createParagraph('other-paragraph', 'other-note'); + update('first-notebook'); + updates.next({ + noteId: 'other-note', + interpreterGroupId: 'other-group', + angularObject: { name: 'controlValue', object: 'second-notebook', noteId: 'other-note' } + }); + const first = bind('A').context; + const second = bind('other-paragraph').context; + expect(first.controlValue).toBe('first-notebook'); + expect(second.controlValue).toBe('second-notebook'); + + service.clearNoteContextValues('note'); + service.setContextValue('controlValue', 'edited-second', 'other-paragraph'); + expect(first.controlValue).toBeUndefined(); + expect(second.controlValue).toBe('edited-second'); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'other-note', + null, + 'controlValue', + 'edited-second', + 'other-group' + ); + }); + it('preserves a private draft when other scopes or interpreter groups change', () => { + const { update, removals, bind } = setup(); + update('private', 'B', 'selected-group'); + const context = bind('B').context; + context.controlValue = 'private-draft'; + + update('other-paragraph', 'A'); + update('notebook'); + update('hidden-private', 'B', 'other-group'); + expect(context.controlValue).toBe('private-draft'); + removals.next({ noteId: 'note', paragraphId: 'B', name: 'controlValue', interpreterGroupId: 'other-group' }); + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'spark' }); + expect(context.controlValue).toBe('private-draft'); + + update('new-private', 'B', 'selected-group'); + expect(context.controlValue).toBe('new-private'); + }); + + it('refreshes every consumer when removal selects a surviving shared interpreter-group binding', () => { + const { update, removals, bind } = setup(); + update('first', undefined, 'first-group'); + update('second', undefined, 'second-group'); + const a = bind('A').context; + const b = bind('B').context; + expect(a.controlValue).toBe('first'); + expect(b.controlValue).toBe('first'); + + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'first-group' }); + expect(a.controlValue).toBe('second'); + expect(b.controlValue).toBe('second'); + removals.next({ noteId: 'note', name: 'controlValue', interpreterGroupId: 'second-group' }); + expect(a.controlValue).toBeUndefined(); + expect(b.controlValue).toBeUndefined(); + }); + + it('reconciles a same-value client update for shared consumers and sends an explicit no-change unbind', () => { + const { update, bind, service, message } = setup(); + update('shared'); + const a = bind('A').context; + const b = bind('B').context; + b.controlValue = 'draft'; + service.setContextValue('controlValue', 'shared', 'A'); + expect(a.controlValue).toBe('shared'); + expect(b.controlValue).toBe('shared'); + expect(message.angularObjectUpdate).toHaveBeenCalledExactlyOnceWith( + 'note', + null, + 'controlValue', + 'shared', + 'spark' + ); + + service.unsetContextValue('not-cached', 'A'); + expect(message.angularObjectClientUnbind).toHaveBeenCalledExactlyOnceWith('note', 'not-cached', 'A'); + }); +}); diff --git a/zeppelin-web-angular/src/app/services/ng-z.service.ts b/zeppelin-web-angular/src/app/services/ng-z.service.ts index f7979c9a09c..1e3e0cd0d63 100644 --- a/zeppelin-web-angular/src/app/services/ng-z.service.ts +++ b/zeppelin-web-angular/src/app/services/ng-z.service.ts @@ -11,20 +11,34 @@ */ import { Injectable, OnDestroy } from '@angular/core'; - -import { AngularContext } from '@zeppelin/core'; -import { isRecord } from '@zeppelin/utility'; import { Subject } from 'rxjs'; +import { + AngularContext, + AngularContextManager, + AngularObjectBinding +} from '../core/paragraph-base/angular-context-manager'; + +interface ContextValue { + value: unknown; + binding?: AngularObjectBinding; +} + +interface ParagraphBindings { + noteId: string; + outputContexts: Set>; + localValues: Map; + resolvedValues: Map; +} + @Injectable({ providedIn: 'root' }) -export class NgZService implements OnDestroy { - private paragraphMap: Map = new Map>(); - private contextChange$ = new Subject(); - private runParagraph$ = new Subject(); - - constructor() {} +export class NgZService implements AngularContextManager, OnDestroy { + private readonly paragraphBindingsById = new Map(); + private readonly boundValuesByNote = new Map>>(); + private readonly contextChange$ = new Subject(); + private readonly runParagraph$ = new Subject(); contextChanged() { return this.contextChange$.asObservable(); @@ -34,48 +48,241 @@ export class NgZService implements OnDestroy { return this.runParagraph$.asObservable(); } - removeParagraph(paragraphId: string) { - this.paragraphMap.delete(paragraphId); + registerParagraphBindings(noteId: string, paragraphId: string): void { + const existing = this.paragraphBindingsById.get(paragraphId); + if (existing?.noteId === noteId) { + return; + } + this.removeParagraphBindings(paragraphId); + this.paragraphBindingsById.set(paragraphId, { + noteId, + outputContexts: new Set(), + localValues: new Map(), + resolvedValues: new Map() + }); + } + + removeParagraphBindings(paragraphId: string): void { + const paragraph = this.paragraphBindingsById.get(paragraphId); + if (!paragraph) { + return; + } + this.paragraphBindingsById.delete(paragraphId); + if (![...this.paragraphBindingsById.values()].some(target => target.noteId === paragraph.noteId)) { + this.boundValuesByNote.delete(paragraph.noteId); + return; + } + const noteBindings = this.boundValuesByNote.get(paragraph.noteId); + for (const [key, values] of noteBindings ?? []) { + for (const [identity, entry] of values) { + if (entry.binding?.paragraphId === paragraphId) { + values.delete(identity); + } + } + if (values.size === 0) { + noteBindings!.delete(key); + } + } + } + + clearNoteContextValues(noteId: string): void { + const keys = new Set(this.boundValuesByNote.get(noteId)?.keys()); + this.boundValuesByNote.delete(noteId); + for (const [paragraphId, paragraph] of this.paragraphBindingsById) { + if (paragraph.noteId !== noteId) { + continue; + } + const paragraphKeys = new Set([...keys, ...paragraph.localValues.keys()]); + paragraph.localValues.clear(); + for (const key of paragraphKeys) { + this.updateParagraphContextValue(paragraphId, paragraph, key, false, false); + } + } } runParagraph(paragraphId: string) { this.runParagraph$.next(paragraphId); } - bindParagraph(paragraphId: string, context: Record) { - this.paragraphMap.set(paragraphId, context); + bindParagraphContext(paragraphId: string, context: Record): () => void { + const paragraph = this.paragraphBindingsById.get(paragraphId); + if (!paragraph) { + return () => {}; + } + paragraph.outputContexts.add(context); + for (const key of new Set([ + ...paragraph.localValues.keys(), + ...(this.boundValuesByNote.get(paragraph.noteId)?.keys() ?? []) + ])) { + const current = this.resolveContextValue(paragraphId, paragraph, key); + if (current) { + paragraph.resolvedValues.set(key, current); + } + context[key] = current?.value; + } + return () => paragraph.outputContexts.delete(context); } - setContextValue(key: string, value: unknown, paragraphId: string, emit = true) { - const context = this.paragraphMap.get(paragraphId); - if (isRecord(context)) { - context[key] = value; + setContextValue(key: string, value: unknown, paragraphId: string, emit = true, binding?: AngularObjectBinding): void { + const paragraph = this.paragraphBindingsById.get(paragraphId); + if (!paragraph) { + return; + } + let updatedValue: ContextValue; + if (binding) { + let noteBindings = this.boundValuesByNote.get(paragraph.noteId); + if (!noteBindings) { + noteBindings = new Map(); + this.boundValuesByNote.set(paragraph.noteId, noteBindings); + } + let values = noteBindings.get(key); + if (!values) { + values = new Map(); + noteBindings.set(key, values); + } + const identity = JSON.stringify([binding.interpreterGroupId, binding.noteId, binding.paragraphId]); + const previous = values.get(identity); + updatedValue = previous && previous.value === value ? previous : { value, binding }; + if (updatedValue !== previous) { + values.set(identity, updatedValue); + } + // Equal cached values must still reconcile fields changed directly by a template. + if (binding.paragraphId !== null) { + this.paragraphBindingsById.get(binding.paragraphId)?.localValues.delete(key); + } + } else { + const current = this.resolveContextValue(paragraphId, paragraph, key); + if (current?.binding) { + updatedValue = current; + current.value = value; + } else { + updatedValue = { value }; + paragraph.localValues.set(key, updatedValue); + } + } + this.refreshNoteContextValue(paragraph.noteId, key, paragraphId, emit, true, updatedValue); + } + + unsetContextValue(key: string, paragraphId: string, emit = true, binding?: AngularObjectBinding): void { + const paragraph = this.paragraphBindingsById.get(paragraphId); + if (!paragraph) { + return; + } + const scope: AngularObjectBinding = binding ?? { noteId: paragraph.noteId, paragraphId }; + if (scope.noteId === paragraph.noteId && scope.paragraphId === paragraphId) { + // Client bind excludes this connection from its echo; a local value may have no cached server entry yet. + paragraph.localValues.delete(key); + } + const values = this.boundValuesByNote.get(paragraph.noteId)?.get(key); + const matches = [...(values ?? [])].filter( + ([, entry]) => + entry.binding?.noteId === scope.noteId && + entry.binding.paragraphId === scope.paragraphId && + (scope.interpreterGroupId === undefined || entry.binding.interpreterGroupId === scope.interpreterGroupId) + ); + if (scope.interpreterGroupId !== undefined || matches.length <= 1) { + for (const [identity] of matches) { + values!.delete(identity); + } + } + if (values?.size === 0) { + this.boundValuesByNote.get(paragraph.noteId)!.delete(key); + } + this.refreshNoteContextValue(paragraph.noteId, key, paragraphId, emit, false); + } + + ngOnDestroy(): void { + this.paragraphBindingsById.clear(); + this.boundValuesByNote.clear(); + this.contextChange$.complete(); + this.runParagraph$.complete(); + } + + private resolveContextValue( + paragraphId: string, + paragraph: ParagraphBindings, + key: string + ): ContextValue | undefined { + let resolved = paragraph.localValues.get(key); + let priority = resolved ? 2 : -1; + for (const entry of this.boundValuesByNote.get(paragraph.noteId)?.get(key)?.values() ?? []) { + const scope = entry.binding!; + if (scope.paragraphId !== null && scope.paragraphId !== paragraphId) { + continue; + } + // Match the server's lookup order, including client-created paragraph values. + const candidatePriority = scope.paragraphId !== null ? 2 : scope.noteId !== null ? 1 : 0; + if (candidatePriority > priority) { + resolved = entry; + priority = candidatePriority; + } + } + return resolved; + } + + private refreshNoteContextValue( + noteId: string, + key: string, + sourceParagraphId: string, + emit: boolean, + set: boolean, + updatedValue?: ContextValue + ): void { + // Server messages already reach every matching paragraph listener. + if (!emit) { + const paragraph = this.paragraphBindingsById.get(sourceParagraphId); + if (paragraph) { + this.updateParagraphContextValue(sourceParagraphId, paragraph, key, false, set, updatedValue); + } + return; + } + for (const [paragraphId, paragraph] of this.paragraphBindingsById) { + if (paragraph.noteId === noteId) { + this.updateParagraphContextValue( + paragraphId, + paragraph, + key, + paragraphId === sourceParagraphId, + set, + updatedValue + ); + } } - this.contextChange$.next({ - paragraphId, - key, - value, - emit, - set: true - }); } - unsetContextValue(key: string, paragraphId: string, emit = true) { - const context = this.paragraphMap.get(paragraphId); - if (isRecord(context)) { - context[key] = undefined; + private updateParagraphContextValue( + paragraphId: string, + paragraph: ParagraphBindings, + key: string, + emit: boolean, + set: boolean, + updatedValue?: ContextValue + ): void { + const current = this.resolveContextValue(paragraphId, paragraph, key); + // Other scopes and interpreter groups must not overwrite this output's local edits. + if ( + !emit && + paragraph.resolvedValues.get(key) === current && + (updatedValue === undefined || current !== updatedValue) + ) { + return; + } + if (current) { + paragraph.resolvedValues.set(key, current); + } else { + paragraph.resolvedValues.delete(key); + } + for (const context of paragraph.outputContexts) { + context[key] = current?.value; } this.contextChange$.next({ + noteId: paragraph.noteId, paragraphId, key, + value: current?.value, + binding: current?.binding, emit, - value: undefined, - set: false + set }); } - - ngOnDestroy(): void { - this.paragraphMap.clear(); - this.contextChange$.complete(); - } } diff --git a/zeppelin-web-angular/src/app/services/runtime-compiler.service.spec.ts b/zeppelin-web-angular/src/app/services/runtime-compiler.service.spec.ts new file mode 100644 index 00000000000..79bfedb357a --- /dev/null +++ b/zeppelin-web-angular/src/app/services/runtime-compiler.service.spec.ts @@ -0,0 +1,178 @@ +/* + * 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 { NgComponentOutlet } from '@angular/common'; +import { Component, provideZoneChangeDetection } from '@angular/core'; +import { TestBed } from '@angular/core/testing'; +import { expect, it } from 'vitest'; +import { NgZService } from './ng-z.service'; +import { DynamicTemplate, RuntimeCompilerService } from './runtime-compiler.service'; + +@Component({ + imports: [NgComponentOutlet], + template: + '' +}) +class OutputHost { + dynamic: DynamicTemplate | null = null; +} + +it('renders retained values, supports native Angular input, and does not resurrect removals on output recreation', async () => { + TestBed.configureTestingModule({ imports: [OutputHost], providers: [provideZoneChangeDetection()] }); + const service = TestBed.inject(NgZService); + service.registerParagraphBindings('note', 'A'); + const binding = { noteId: 'note', paragraphId: null, interpreterGroupId: 'spark-group' }; + service.setContextValue('controlValue', 'hello', 'A', false, binding); + const compiler = TestBed.inject(RuntimeCompilerService); + const dynamic = await compiler.createAndCompileTemplate( + 'A', + "{{controlValue}}" + ); + const fixture = TestBed.createComponent(OutputHost); + fixture.componentInstance.dynamic = dynamic; + fixture.detectChanges(); + const element = fixture.nativeElement as HTMLElement; + expect(element.querySelector('output')?.textContent).toBe('hello'); + element.querySelector('button')!.click(); + fixture.detectChanges(); + expect(element.querySelector('output')?.textContent).toBe('edited'); + fixture.componentInstance.dynamic = null; + fixture.detectChanges(); + service.unsetContextValue('controlValue', 'A', false, binding); + fixture.componentInstance.dynamic = dynamic; + fixture.detectChanges(); + expect(element.querySelector('output')?.textContent).toBe(''); +}); + +it('uses current binding state when a model is removed while template compilation is pending', async () => { + TestBed.configureTestingModule({ imports: [OutputHost], providers: [provideZoneChangeDetection()] }); + const service = TestBed.inject(NgZService); + service.registerParagraphBindings('note', 'A'); + const binding = { noteId: 'note', paragraphId: null, interpreterGroupId: 'spark' }; + service.setContextValue('controlValue', 'removed-during-compilation', 'A', false, binding); + const pending = TestBed.inject(RuntimeCompilerService).createAndCompileTemplate( + 'A', + '{{controlValue}}' + ); + service.unsetContextValue('controlValue', 'A', false, binding); + + const fixture = TestBed.createComponent(OutputHost); + fixture.componentInstance.dynamic = await pending; + fixture.detectChanges(); + const element = fixture.nativeElement as HTMLElement; + expect(element.querySelector('output')?.textContent).toBe(''); + service.setContextValue('controlValue', 'rebound', 'A', false, binding); + fixture.detectChanges(); + expect(element.querySelector('output')?.textContent).toBe('rebound'); +}); + +it('reapplies repeated server values after a native template-local edit without changing sibling outputs', async () => { + TestBed.configureTestingModule({ imports: [OutputHost], providers: [provideZoneChangeDetection()] }); + const service = TestBed.inject(NgZService); + service.registerParagraphBindings('note', 'A'); + const binding = { noteId: 'note', paragraphId: null, interpreterGroupId: 'spark' }; + service.setContextValue('controlValue', 'server', 'A', false, binding); + const dynamic = await TestBed.inject(RuntimeCompilerService).createAndCompileTemplate( + 'A', + '{{controlValue}}' + ); + const first = TestBed.createComponent(OutputHost); + const second = TestBed.createComponent(OutputHost); + first.componentInstance.dynamic = dynamic; + second.componentInstance.dynamic = dynamic; + first.detectChanges(); + second.detectChanges(); + const firstElement = first.nativeElement as HTMLElement; + const secondElement = second.nativeElement as HTMLElement; + expect(firstElement.querySelector('output')?.textContent).toBe('server'); + expect(secondElement.querySelector('output')?.textContent).toBe('server'); + + firstElement.querySelector('button')!.click(); + first.detectChanges(); + second.detectChanges(); + expect(firstElement.querySelector('output')?.textContent).toBe('local'); + expect(secondElement.querySelector('output')?.textContent).toBe('server'); + + service.setContextValue('controlValue', 'server', 'A', false, binding); + first.detectChanges(); + second.detectChanges(); + expect(firstElement.querySelector('output')?.textContent).toBe('server'); + expect(secondElement.querySelector('output')?.textContent).toBe('server'); +}); + +it('keeps a releaseContext model separate from output cleanup', async () => { + TestBed.configureTestingModule({ imports: [OutputHost], providers: [provideZoneChangeDetection()] }); + const service = TestBed.inject(NgZService); + service.registerParagraphBindings('note', 'A'); + const binding = { noteId: 'note', paragraphId: null, interpreterGroupId: 'spark' }; + service.setContextValue('releaseContext', 'initial', 'A', false, binding); + const dynamic = await TestBed.inject(RuntimeCompilerService).createAndCompileTemplate( + 'A', + '{{releaseContext}}' + ); + const fixture = TestBed.createComponent(OutputHost); + fixture.componentInstance.dynamic = dynamic; + fixture.detectChanges(); + const element = fixture.nativeElement as HTMLElement; + expect(element.querySelector('output')?.textContent).toBe('initial'); + + service.setContextValue('releaseContext', 'updated', 'A', false, binding); + fixture.detectChanges(); + expect(element.querySelector('output')?.textContent).toBe('updated'); + fixture.componentInstance.dynamic = null; + fixture.detectChanges(); + service.setContextValue('releaseContext', 'after-destroy', 'A', false, binding); + fixture.componentInstance.dynamic = dynamic; + fixture.detectChanges(); + expect(element.querySelector('output')?.textContent).toBe('after-destroy'); +}); + +it('propagates a native shared edit without resetting a different paragraph binding draft', async () => { + TestBed.configureTestingModule({ imports: [OutputHost], providers: [provideZoneChangeDetection()] }); + const service = TestBed.inject(NgZService); + service.registerParagraphBindings('note', 'A'); + service.registerParagraphBindings('note', 'B'); + service.registerParagraphBindings('note', 'C'); + const shared = { noteId: 'note', paragraphId: null, interpreterGroupId: 'spark' }; + service.setContextValue('controlValue', 'shared', 'A', false, shared); + service.setContextValue('controlValue', 'private', 'B', false, { ...shared, paragraphId: 'B' }); + const compiler = TestBed.inject(RuntimeCompilerService); + const fixtures = []; + for (const id of ['A', 'B', 'C']) { + const dynamic = await compiler.createAndCompileTemplate( + id, + id === 'A' + ? "{{controlValue}}" + : '{{controlValue}}' + ); + const fixture = TestBed.createComponent(OutputHost); + fixture.componentInstance.dynamic = dynamic; + fixture.detectChanges(); + fixtures.push(fixture); + } + const [a, b, c] = fixtures; + const aElement = a.nativeElement as HTMLElement; + const bElement = b.nativeElement as HTMLElement; + const cElement = c.nativeElement as HTMLElement; + expect(cElement.querySelector('output')?.textContent).toBe('shared'); + bElement.querySelector('button')!.click(); + b.detectChanges(); + expect(bElement.querySelector('output')?.textContent).toBe('draft'); + + aElement.querySelector('button')!.click(); + a.detectChanges(); + b.detectChanges(); + c.detectChanges(); + expect(aElement.querySelector('output')?.textContent).toBe('edited'); + expect(bElement.querySelector('output')?.textContent).toBe('draft'); + expect(cElement.querySelector('output')?.textContent).toBe('edited'); +}); diff --git a/zeppelin-web-angular/src/app/services/runtime-compiler.service.ts b/zeppelin-web-angular/src/app/services/runtime-compiler.service.ts index 83e30aeda92..1ee76aa7e9b 100644 --- a/zeppelin-web-angular/src/app/services/runtime-compiler.service.ts +++ b/zeppelin-web-angular/src/app/services/runtime-compiler.service.ts @@ -33,9 +33,14 @@ export class DynamicTemplateError { providedIn: 'root' }) export class RuntimeCompilerService { + constructor( + private compiler: Compiler, + private ngZService: NgZService + ) {} + public async createAndCompileTemplate(paragraphId: string, template: string): Promise { const ngZService = this.ngZService; - const dynamicComponent = Component({ template, selector: `dynamic-${paragraphId}` })( + const dynamicComponent = Component({ template, selector: `dynamic-${paragraphId}`, standalone: false })( class DynamicTemplateComponent { z = { set: (key: string, value: unknown, id: string) => ngZService.setContextValue(key, value, id), @@ -43,10 +48,16 @@ export class RuntimeCompilerService { run: (id: string) => ngZService.runParagraph(id) }; + readonly #releaseContext: () => void; + constructor() { - ngZService.bindParagraph(paragraphId, this as Record); + this.#releaseContext = ngZService.bindParagraphContext(paragraphId, this as Record); Object.freeze(this.z); } + + ngOnDestroy() { + this.#releaseContext(); + } } ); const dynamicModule = NgModule({ @@ -63,9 +74,4 @@ export class RuntimeCompilerService { throw new DynamicTemplateError(`${e}`); } } - - constructor( - private compiler: Compiler, - private ngZService: NgZService - ) {} }