wip: expanding event handling

- Removed `EventBus` handling adding node information to the nodeGraph.

[ci skip]

Author: tegefaulkes

package-lock.json

@@ -74,7 +74,7 @@
     "@matrixai/timer": "^1.1.1",
     "@matrixai/workers": "^1.3.7",
     "@matrixai/quic": "^0.0.21",
-    "@matrixai/events": "^1.0.0",
+    "@matrixai/events": "^2.0.2",
     "@peculiar/asn1-pkcs8": "^2.3.0",
     "@peculiar/asn1-schema": "^2.3.0",
     "@peculiar/asn1-x509": "^2.3.0",

package.json

@@ -97,10 +97,8 @@ class PolykeyAgent {
    */
   public static readonly eventSymbols = {
     [CertManager.name]: Symbol(CertManager.name),
-    [QUICServer.name]: Symbol(QUICServer.name),
   } as {
     readonly CertManager: unique symbol;
-    readonly QUICServer: unique symbol;
   };
 
   /**
@@ -714,26 +712,6 @@ class PolykeyAgent {
           this.logger.info(`${KeyRing.name} change propagated`);
         },
       );
-      this.events.on(
-        PolykeyAgent.eventSymbols.QUICServer,
-        async (data: ConnectionData) => {
-          if (this.keyRing.getNodeId().equals(data.remoteNodeId)) return;
-          const address = networkUtils.buildAddress(
-            data.remoteHost,
-            data.remotePort,
-          );
-          const nodeIdEncoded = nodesUtils.encodeNodeId(data.remoteNodeId);
-          this.logger.info(
-            `Connection adding ${nodeIdEncoded}:${address} to ${NodeGraph.name}`,
-          );
-          // Reverse connection was established and authenticated,
-          //  add it to the node graph
-          await this.nodeManager.setNode(data.remoteNodeId, {
-            host: data.remoteHost,
-            port: data.remotePort,
-          });
-        },
-      );
       const _networkConfig = {
         ...config.defaults.networkConfig,
         ...utils.filterEmptyObject(networkConfig),

src/PolykeyAgent.ts

@@ -4,19 +4,15 @@ import type { NodeId } from './types';
 import type { Host, Hostname, Port, TLSConfig } from '../network/types';
 import type { Certificate, CertificatePEM } from '../keys/types';
 import type { ClientManifest } from '../rpc/types';
-import type {
-  QUICSocket,
-  ClientCrypto,
-  QUICConnection,
-  events as quicEvents,
-} from '@matrixai/quic';
+import type { QUICSocket, ClientCrypto, QUICConnection } from '@matrixai/quic';
 import type { ContextTimedInput } from '@matrixai/contexts/dist/types';
 import type { X509Certificate } from '@peculiar/x509';
 import Logger from '@matrixai/logger';
 import { CreateDestroy } from '@matrixai/async-init/dist/CreateDestroy';
 import { timedCancellable, context } from '@matrixai/contexts/dist/decorators';
 import { Evented } from '@matrixai/events';
-import { QUICClient } from '@matrixai/quic';
+import { QUICClient, events as quicEvents } from '@matrixai/quic';
+import { QUICClientEvent } from '@matrixai/quic/dist/events';
 import * as nodesErrors from './errors';
 import * as nodesEvents from './events';
 import RPCClient from '../rpc/RPCClient';
@@ -31,6 +27,7 @@ import { never } from '../utils';
  */
 // eslint-disable-next-line @typescript-eslint/no-unused-vars -- False positive for M
 interface NodeConnection<M extends ClientManifest> extends CreateDestroy {}
+// eslint-disable-next-line @typescript-eslint/no-unused-vars -- False positive for M
 interface NodeConnection<M extends ClientManifest> extends Evented {}
 @CreateDestroy()
 @Evented()
@@ -196,33 +193,55 @@ class NodeConnection<M extends ClientManifest> {
         }:${quicConnection.remotePort}]`,
       ),
     });
-    // Setting up stream handling
-    const handleConnectionStream = (
-      streamEvent: quicEvents.QUICConnectionStreamEvent,
-    ) => {
-      const stream = streamEvent.detail;
-      nodeConnection.dispatchEvent(
-        new nodesEvents.EventNodeStream({ detail: stream }),
-      );
-    };
-    // TODO: All quicConnection event handling should be done under a unified handler.
+    // TODO: remove this later based on testing
     quicConnection.removeEventListener('connectionStream', throwFunction);
-    quicConnection.addEventListener('connectionStream', handleConnectionStream);
+    // QUICClient events are...
+    //   - QUICClientEvent,
+    //   - QUICClientDestroyEvent,
+    //   - QUICClientErrorEvent,
+    // QUICConnection events are...
+    //   - QUICConnectionEvent,
+    //   - QUICConnectionStreamEvent,
+    //   - QUICConnectionStopEvent,
+    //   - QUICConnectionErrorEvent,
+    // QUICStream events are...
+    //   - QUICStreamEvent,
+    //   - QUICStreamDestroyEvent,
+    const handleNodeConnectionEvents = (e) => {
+      if (e instanceof quicEvents.QUICClientDestroyEvent) {
+        // Trigger the nodeConnection destroying
+        void nodeConnection.destroy({ force: false });
+      } else if (e instanceof quicEvents.QUICConnectionStreamEvent) {
+        const quicStream = e.detail;
+        // Setting up stream handling
+        quicStream.addEventListener(
+          quicEvents.QUICStreamDestroyEvent.name,
+          (e) => {
+            nodeConnection.dispatchEvent(e.clone());
+          },
+          { once: true },
+        );
+        nodeConnection.dispatchEvent(
+          new nodesEvents.EventNodeStream({ detail: quicStream }),
+        );
+      }
+      nodeConnection.dispatchEvent(e.clone());
+    };
+    // Setting up QUICConnection events
+    quicConnection.addEventListener(handleNodeConnectionEvents);
     quicConnection.addEventListener(
-      'connectionStop',
+      quicEvents.QUICConnectionStopEvent.name,
       () => {
-        quicConnection.removeEventListener(
-          'connectionStream',
-          handleConnectionStream,
-        );
+        quicConnection.removeEventListener(handleNodeConnectionEvents);
       },
       { once: true },
     );
+    // Setting up QUICClient events
+    quicClient.addEventListener(handleNodeConnectionEvents);
     quicClient.addEventListener(
-      'clientDestroy',
+      quicEvents.QUICClientDestroyEvent.name,
       () => {
-        // Trigger the nodeConnection destroying
-        void nodeConnection.destroy({ force: false });
+        quicClient.removeEventListener(handleNodeConnectionEvents);
       },
       { once: true },
     );
@@ -269,31 +288,41 @@ class NodeConnection<M extends ClientManifest> {
       rpcClient,
       logger,
     });
-    // Setting up stream handling
-    // TODO: All quicConnection event handling should be done under a unified handler.
-    const handleConnectionStream = (
-      streamEvent: quicEvents.QUICConnectionStreamEvent,
-    ) => {
-      nodeConnection.dispatchEvent(
-        new nodesEvents.EventNodeStream({ detail: streamEvent.detail }),
-      );
+    // QUICClient events are...
+    //   - QUICClientEvent,
+    //   - QUICClientDestroyEvent,
+    //   - QUICClientErrorEvent,
+    // QUICConnection events are...
+    //   - QUICConnectionEvent,
+    //   - QUICConnectionStreamEvent,
+    //   - QUICConnectionStopEvent,
+    //   - QUICConnectionErrorEvent,
+    const handleNodeConnectionEvents = (e) => {
+      if (e instanceof quicEvents.QUICClientDestroyEvent) {
+        // Trigger the nodeConnection destroying
+        void nodeConnection.destroy({ force: false });
+      } else if (e instanceof quicEvents.QUICConnectionStreamEvent) {
+        const quicStream = e.detail;
+        // Setting up stream handling
+        quicStream.addEventListener(
+          quicEvents.QUICStreamDestroyEvent.name,
+          (e) => {
+            nodeConnection.dispatchEvent(e.clone());
+          },
+          { once: true },
+        );
+        nodeConnection.dispatchEvent(
+          new nodesEvents.EventNodeStream({ detail: quicStream }),
+        );
+      }
+      nodeConnection.dispatchEvent(e.clone());
     };
-    quicConnection.addEventListener('connectionStream', handleConnectionStream);
+    // Setting up QUICConnection events
+    quicConnection.addEventListener(handleNodeConnectionEvents);
     quicConnection.addEventListener(
-      'connectionStop',
+      quicEvents.QUICConnectionStopEvent.name,
       () => {
-        quicConnection.removeEventListener(
-          'connectionStream',
-          handleConnectionStream,
-        );
-      },
-      { once: true },
-    );
-    quicConnection.addEventListener(
-      'connectionStop',
-      async () => {
-        // Trigger the nodeConnection destroying
-        await nodeConnection.destroy({ force: false });
+        quicConnection.removeEventListener(handleNodeConnectionEvents);
       },
       { once: true },
     );