Added a trace of trigger to the subscribable dataset

packages/connected/package.json

@@ -6,7 +6,7 @@
   "scripts": {
     "build": "tsc --project tsconfig.build.json",
     "watch": "tsc --watch",
-    "test": "cross-env NODE_OPTIONS=--experimental-vm-modules jest --coverage",
+    "test": "cross-env NODE_OPTIONS=--experimental-vm-modules jest --coverage -t \"handles subscriptions if data changes\"",
     "test:watch": "jest --watch",
     "prepublishOnly": "npm run test && npm run build",
     "lint": "eslint src/** --fix --no-error-on-unmatched-pattern",

packages/connected/src/linkTraversal/ResourceLinkQuery.ts

@@ -46,8 +46,16 @@ export class ResourceLinkQuery<
     return this.fromSubject();
   }
 
-  subscribe(): Promise<string> {
+  async subscribe(): Promise<string> {
-    throw new Error("Method not implemented.");
+    await exploreLinks(
+      this.parentDataset,
+      this.shapeType,
+      this.startingResource,
+      this.startingSubject,
+      this.linkQueryInput,
+      {},
+    );
+    return "string";
   }
 
   private async fullUnsubscribe(): Promise<void> {

packages/connected/src/linkTraversal/exploreLinks.ts

@@ -4,6 +4,7 @@ import type { SubjectNode } from "@ldo/rdf-utils";
 import type { LQInput } from "../types/ILinkQuery";
 import { BasicLdSet } from "@ldo/jsonld-dataset-proxy";
 import type { IConnectedLdoDataset } from "../types/IConnectedLdoDataset";
+import { createTrackingProxyBuilder } from "../trackingProxy/createTrackingProxy";
 
 interface ExploreLinksOptions<Plugins extends ConnectedPlugin[]> {
   onResourceEncountered?: (
@@ -29,7 +30,15 @@ export async function exploreLinks<
     : await startingResource.readIfUnfetched();
   if (readResult.isError) return;
 
-  const ldObject = dataset.usingType(shapeType).fromSubject(startingSubject);
+  const trackingProxyBuilder = createTrackingProxyBuilder(
+    dataset,
+    shapeType,
+    (changes) =>
+      console.log(
+        `Got Update \nadded: ${changes.added?.toString()}\nremoved: ${changes.removed?.toString()}`,
+      ),
+  );
+  const ldObject = trackingProxyBuilder.fromSubject(startingSubject);
 
   const fetchedDuringThisExploration = new Set<string>([startingResource.uri]);
 

packages/connected/test/LinkTraversalData.ts

@@ -50,7 +50,7 @@ export const linkTraversalData: ResourceInfo = {
 
         :me a foaf:Person ;
             foaf:name "Third User" ;
-            foaf:mbox <mailto:other@example.org> ;
+            foaf:mbox <mailto:third@example.org> ;
             foaf:knows <http://localhost:3005/test-container/mainProfile.ttl#me> .
       `,
     },

packages/connected/test/LinkTraversalIntegration.test.ts

@@ -76,6 +76,8 @@ describe("Link Traversal", () => {
     expect(mainProfile.knows?.size).toBe(1);
     expect(mainProfile.knows?.toArray()[0].name).toBe("Other User");
 
+    console.log("==================");
+
     // Update to include a new document
     const cMainProfile = changeData(mainProfile, mainProfileResource);
     // eslint-disable-next-line @typescript-eslint/ban-ts-comment

packages/subscribable-dataset/src/SubscribableDataset.ts

@@ -175,7 +175,10 @@ export class SubscribableDataset<InAndOutQuad extends BaseQuad = BaseQuad>
     // A mapping of serialized QuadMatches to the changed quads
     const matchingDatasetChanges: Record<
       string,
-      DatasetChanges<InAndOutQuad>
+      {
+        changes: DatasetChanges<InAndOutQuad>;
+        triggerQuadMatch: QuadMatch;
+      }
     > = {};
 
     // Population MatchingDatasetChanges
@@ -217,13 +220,18 @@ export class SubscribableDataset<InAndOutQuad extends BaseQuad = BaseQuad>
           if (this.eventEmitter.listenerCount(eventName) > 0) {
             // Set matchingDatasetChanges to include data to emit
             if (!matchingDatasetChanges[eventName]) {
-              matchingDatasetChanges[eventName] = {};
+              matchingDatasetChanges[eventName] = {
+                triggerQuadMatch: quadMatch,
+                changes: {},
+              };
             }
-            if (!matchingDatasetChanges[eventName][changeType]) {
+            if (!matchingDatasetChanges[eventName].changes[changeType]) {
-              matchingDatasetChanges[eventName][changeType] =
+              matchingDatasetChanges[eventName].changes[changeType] =
                 this.datasetFactory.dataset();
             }
-            matchingDatasetChanges[eventName][changeType]?.add(changedQuad);
+            matchingDatasetChanges[eventName].changes[changeType]?.add(
+              changedQuad,
+            );
           }
         });
       });
@@ -233,8 +241,12 @@ export class SubscribableDataset<InAndOutQuad extends BaseQuad = BaseQuad>
 
     // Alert all listeners
     Object.entries(matchingDatasetChanges).forEach(
-      ([quadMatchString, changes]) => {
+      ([quadMatchString, info]) => {
-        this.eventEmitter.emit(quadMatchString, changes);
+        this.eventEmitter.emit(
+          quadMatchString,
+          info.changes,
+          info.triggerQuadMatch,
+        );
       },
     );
   }

packages/subscribable-dataset/src/types.ts

@@ -6,6 +6,7 @@ import type { Dataset, BaseQuad, DatasetFactory } from "@rdfjs/types";
  */
 export type nodeEventListener<InAndOutQuad extends BaseQuad = BaseQuad> = (
   changes: DatasetChanges<InAndOutQuad>,
+  triggeringQuadMatch: QuadMatch,
 ) => void;
 
 /**

packages/subscribable-dataset/test/SubscribableDataset.test.ts

@@ -62,6 +62,12 @@ describe("SubscribableDataset", () => {
     expect(callbackFunc).toBeCalledTimes(1);
     expect(callbackFunc.mock.calls[0][0].added.size).toBe(1);
     expect(callbackFunc.mock.calls[0][0].added.has(tomColorQuad)).toBe(true);
+    expect(callbackFunc.mock.calls[0][1]).toEqual([
+      namedNode("http://example.org/cartoons#Tom"),
+      null,
+      null,
+      null,
+    ]);
   });
 
   it("Alerts when a node is removed", () => {
@@ -74,6 +80,12 @@ describe("SubscribableDataset", () => {
     expect(callbackFunc).toBeCalledTimes(1);
     expect(callbackFunc.mock.calls[0][0].removed.size).toBe(1);
     expect(callbackFunc.mock.calls[0][0].removed.has(tomTypeQuad)).toBe(true);
+    expect(callbackFunc.mock.calls[0][1]).toEqual([
+      namedNode("http://example.org/cartoons#Tom"),
+      null,
+      null,
+      null,
+    ]);
   });
 
   it("Alerts when multiple quads are added", () => {
@@ -87,6 +99,12 @@ describe("SubscribableDataset", () => {
     expect(callbackFunc.mock.calls[0][0].added.size).toBe(2);
     expect(callbackFunc.mock.calls[0][0].added.has(lickyNameQuad)).toBe(true);
     expect(callbackFunc.mock.calls[0][0].added.has(lickyTypeQuad)).toBe(true);
+    expect(callbackFunc.mock.calls[0][1]).toEqual([
+      namedNode("http://example.org/cartoons#Licky"),
+      null,
+      null,
+      null,
+    ]);
   });
 
   it("Alerts when bulk updated by only adding", () => {
@@ -105,6 +123,12 @@ describe("SubscribableDataset", () => {
       ),
     ).toBe(true);
     expect(callbackFuncLicky.mock.calls[0][0].removed).toBe(undefined);
+    expect(callbackFuncLicky.mock.calls[0][1]).toEqual([
+      namedNode("http://example.org/cartoons#Licky"),
+      null,
+      null,
+      null,
+    ]);
   });
 
   it("Alerts when bulk updated by only removing", () => {
@@ -123,6 +147,12 @@ describe("SubscribableDataset", () => {
       ),
     ).toBe(true);
     expect(callbackFuncTom.mock.calls[0][0].added).toBe(undefined);
+    expect(callbackFuncTom.mock.calls[0][1]).toEqual([
+      namedNode("http://example.org/cartoons#Tom"),
+      null,
+      null,
+      null,
+    ]);
   });
 
   it("Alerts when emit is called", () => {