EventsSubscribe

src/event-log/event-emitter-stream.ts

@@ -6,17 +6,13 @@ import { EventEmitter } from 'events';
 
 const EVENTS_LISTENER_CHANNEL = 'events';
 
-type EventStreamEmitterConfig = {
-  emitter?: EventEmitter;
-};
-
 export class EventEmitterStream implements EventStream {
   private eventEmitter: EventEmitter;
   private isOpen: boolean = false;
 
-  constructor(config?: EventStreamEmitterConfig) {
+  constructor() {
     // we capture the rejections and currently just log the errors that are produced
-    this.eventEmitter = config?.emitter || new EventEmitter({ captureRejections: true });
+    this.eventEmitter = new EventEmitter({ captureRejections: true });
     this.eventEmitter.on('error', this.eventError);
   }
 

src/interfaces/records-delete.ts

@@ -70,8 +70,8 @@ export class RecordsDelete extends AbstractMessage<RecordsDeleteMessage> {
   }
 
   /**
-    * Indexed properties needed for MessageStore indexing.
-    */
+   * Indexed properties needed for MessageStore indexing.
+   */
   public constructIndexes(
     initialWrite: RecordsWriteMessage,
   ): KeyValues {

src/types/events-types.ts

@@ -3,6 +3,7 @@ import type { AuthorizationModel, GenericMessage, GenericMessageReply, MessageSu
 import type { DwnInterfaceName, DwnMethodName } from '../enums/dwn-interface-method.js';
 import type { PaginationCursor, RangeCriterion, RangeFilter } from './query-types.js';
 
+// filters used when filtering for any type of Message across interfaces
 export type EventsMessageFilter = {
   interface?: string;
   method?: string;
@@ -23,6 +24,8 @@ export type EventsRecordsFilter = {
   dateCreated?: RangeCriterion;
 };
 
+// a union type of the different types of filters a user can use when issuing an EventsQuery or EventsSubscribe
+//TODO: simplify the EventsFilters to only the necessary in order to reduce complexity https://github.com/TBD54566975/dwn-sdk-js/issues/663
 export type EventsFilter = EventsMessageFilter | EventsRecordsFilter | ProtocolsQueryFilter;
 
 export type EventsGetDescriptor = {

tests/event-log/event-emitter-stream.spec.ts

@@ -1,5 +1,4 @@
-import { EventEmitter } from 'events';
-
+import type { EventEmitter } from 'events';
 import type { MessageStore } from '../../src/index.js';
 
 import { EventEmitterStream } from '../../src/event-log/event-emitter-stream.js';
@@ -37,8 +36,8 @@ describe('EventEmitterStream', () => {
   });
 
   it('should remove listeners when `close` method is used', async () => {
-    const emitter = new EventEmitter();
-    eventStream = new EventEmitterStream({ emitter });
+    eventStream = new EventEmitterStream();
+    const emitter = (eventStream as any).eventEmitter as EventEmitter;
 
     // count the `events` listeners, which represents all listeners
     expect(emitter.listenerCount('events')).to.equal(0);
@@ -53,8 +52,8 @@ describe('EventEmitterStream', () => {
 
   it('logs message when the emitter experiences an error', async () => {
     const eventErrorSpy = sinon.spy(EventEmitterStream.prototype as any, 'eventError');
-    const emitter = new EventEmitter({ captureRejections: true });
-    eventStream = new EventEmitterStream({ emitter });
+    eventStream = new EventEmitterStream();
+    const emitter = (eventStream as any).eventEmitter as EventEmitter;
     emitter.emit('error', new Error('random error'));
     expect(eventErrorSpy.callCount).to.equal(1);
   });

tests/event-log/event-stream.spec.ts

@@ -23,12 +23,20 @@ describe('EventStream', () => {
   });
 
   it('emits all messages to each subscriptions', async () => {
-    const messageCids: string[] = [];
-    const handler = async (_tenant: string, message: GenericMessage, _indexes: KeyValues): Promise<void> => {
+    const messageCids1: string[] = [];
+    const handler1 = async (_tenant: string, message: GenericMessage, _indexes: KeyValues): Promise<void> => {
       const messageCid = await Message.getCid(message);
-      messageCids.push(messageCid);
+      messageCids1.push(messageCid);
     };
-    const subcription = await eventStream.subscribe('sub-1', handler);
+
+    const messageCids2: string[] = [];
+    const handler2 = async (_tenant: string, message: GenericMessage, _indexes: KeyValues): Promise<void> => {
+      const messageCid = await Message.getCid(message);
+      messageCids2.push(messageCid);
+    };
+
+    const subscription1 = await eventStream.subscribe('sub-1', handler1);
+    const subscription2 = await eventStream.subscribe('sub-2', handler2);
 
     const message1 = await TestDataGenerator.generateRecordsWrite({});
     const message1Cid = await Message.getCid(message1.message);
@@ -40,11 +48,13 @@ describe('EventStream', () => {
     const message3Cid = await Message.getCid(message3.message);
     eventStream.emit('did:alice', message3.message, {});
 
-    await subcription.close();
+    await subscription1.close();
+    await subscription2.close();
 
     await Time.minimalSleep();
 
-    expect(messageCids).to.have.members([ message1Cid, message2Cid, message3Cid ]);
+    expect(messageCids1).to.have.members([ message1Cid, message2Cid, message3Cid ]);
+    expect(messageCids2).to.have.members([ message1Cid, message2Cid, message3Cid ]);
   });
 
   it('does not emit messages if subscription is closed', async () => {

tests/interfaces/events-subscribe.spec.ts

@@ -1,3 +1,4 @@
+import { authorizeOwner } from '../../src/core/auth.js';
 import { EventsSubscribe } from '../../src/interfaces/events-subscribe.js';
 import { DidKeyResolver, DwnInterfaceName, DwnMethodName, Jws, Time } from '../../src/index.js';
 
@@ -8,15 +9,19 @@ describe('EventsSubscribe', () => {
     it('should be able to create and authorize EventsSubscribe', async () => {
       const alice = await DidKeyResolver.generate();
       const timestamp = Time.getCurrentTimestamp();
-      const { message } = await EventsSubscribe.create({
+      const eventsSubscribe = await EventsSubscribe.create({
         signer           : Jws.createSigner(alice),
         messageTimestamp : timestamp,
       });
 
+      const message = eventsSubscribe.message;
       expect(message.descriptor.interface).to.eql(DwnInterfaceName.Events);
       expect(message.descriptor.method).to.eql(DwnMethodName.Subscribe);
       expect(message.authorization).to.exist;
       expect(message.descriptor.messageTimestamp).to.equal(timestamp);
+
+      // EventsSubscribe authorizes against owner
+      await authorizeOwner(alice.did, eventsSubscribe);
     });
   });
 });

tests/scenarios/subscriptions.spec.ts

@@ -266,11 +266,11 @@ export function testSubscriptionScenarios(): void {
         expect(write2Proto2Response.status.code).equals(202);
 
         // proto1 messages from handler do not change.
-        expect(proto1Messages.length).to.equal(1, 'proto1 after subscription.off()');
+        expect(proto1Messages.length).to.equal(1, 'proto1 after subscription.close()');
         expect(proto1Messages).to.include(await Message.getCid(write1proto1.message));
 
         //proto2 messages from handler have the new message.
-        expect(proto2Messages.length).to.equal(2, 'proto2 after subscription.off()');
+        expect(proto2Messages.length).to.equal(2, 'proto2 after subscription.close()');
         expect(proto2Messages).to.have.members([await Message.getCid(write1proto2.message), await Message.getCid(write2proto2.message)]);
       });
 

tests/test-event-stream.ts

@@ -3,24 +3,24 @@ import type { EventStream } from '../src/index.js';
 import { EventEmitterStream } from '../src/index.js';
 
 /**
- * Class that manages store implementations for testing.
+ * Class that manages the EventStream implementation for testing.
  * This is intended to be extended as the single point of configuration
- * that allows different store implementations to be swapped in
- * to test compatibility with default/built-in store implementations.
+ * that allows different EventStream implementations to be swapped in
+ * to test compatibility with default/built-in implementation.
  */
 export class TestEventStream {
   private static eventStream?: EventStream;
 
   /**
-   * Overrides test stores with given implementation.
+   * Overrides the event stream with a given implementation.
    * If not given, default implementation will be used.
    */
   public static override(overrides?: { eventStream?: EventStream }): void {
     TestEventStream.eventStream = overrides?.eventStream;
   }
 
   /**
-   * Initializes and return the stores used for running the test suite.
+   * Initializes and returns the event stream used for running the test suite.
    */
   public static get(): EventStream {
     TestEventStream.eventStream ??= new EventEmitterStream();