models: add Model and EventStream classes

src/EventStream.test.ts

@@ -0,0 +1,37 @@
+import { it, describe, expect, expectTypeOf, beforeEach, afterEach } from 'vitest';
+import { Realtime, Types } from 'ably/promises';
+import { WebSocket } from 'mock-socket';
+
+import EventStream from './EventStream';
+
+import Server from './utilities/test/mock-server.js';
+import defaultClientConfig from './utilities/test/default-client-config.js';
+
+interface EventStreamTestContext {
+  client: Types.RealtimePromise;
+  server: Server;
+}
+
+describe('EventStream', () => {
+  beforeEach<EventStreamTestContext>((context) => {
+    (Realtime as any).Platform.Config.WebSocket = WebSocket;
+    context.server = new Server('wss://realtime.ably.io/');
+    context.client = new Realtime(defaultClientConfig);
+  });
+
+  afterEach<EventStreamTestContext>((context) => {
+    context.server.stop();
+  });
+
+  it<EventStreamTestContext>('connects successfully with the Ably Client', async ({ client, server }) => {
+    server.start();
+    const connectSuccess = await client.connection.whenState('connected');
+    expect(connectSuccess.current).toBe('connected');
+  });
+
+  it<EventStreamTestContext>('expects the injected client to be of the type RealtimePromise', ({ client }) => {
+    const eventStream = new EventStream('test', client, { channel: 'foobar' });
+    expect(eventStream.name).toEqual('test');
+    expectTypeOf(eventStream.client).toMatchTypeOf<Types.RealtimePromise>();
+  });
+});

src/EventStream.ts

@@ -0,0 +1,31 @@
+import { Types } from 'ably';
+import EventStreamOptions from './options/EventStreamOptions';
+import EventEmitter from './utilities/EventEmitter';
+
+const EVENT_STREAM_OPTIONS_DEFAULTS = {};
+
+enum EventStreamState {
+  /**
+   * The event stream has been initialized but no attach has yet been attempted.
+   */
+  INITIALIZED = 'initialized',
+  /**
+   * This state is entered if the event stream encounters a failure condition that it cannot recover from.
+   */
+  FAILED = 'failed',
+}
+
+class EventStream<T> extends EventEmitter<any> {
+  private options: EventStreamOptions;
+  private connectionId?: string;
+  private state: EventStreamState = EventStreamState.INITIALIZED;
+  private data: T;
+
+  constructor(readonly name: string, readonly client: Types.RealtimePromise, options: EventStreamOptions) {
+    super();
+    this.options = { ...EVENT_STREAM_OPTIONS_DEFAULTS, ...options };
+    this.connectionId = this.client.connection.id;
+  }
+}
+
+export default EventStream;

src/Model.test.ts

@@ -0,0 +1,50 @@
+import { it, describe, expect, expectTypeOf, beforeEach, afterEach } from 'vitest';
+import { Realtime, Types } from 'ably/promises';
+import { WebSocket } from 'mock-socket';
+
+import Model from './Model';
+import EventStream from './EventStream';
+
+import Server from './utilities/test/mock-server.js';
+import defaultClientConfig from './utilities/test/default-client-config.js';
+
+interface ModelTestContext {
+  client: Types.RealtimePromise;
+  server: Server;
+}
+
+describe('Model', () => {
+  beforeEach<ModelTestContext>((context) => {
+    (Realtime as any).Platform.Config.WebSocket = WebSocket;
+    context.server = new Server('wss://realtime.ably.io/');
+    context.client = new Realtime(defaultClientConfig);
+  });
+
+  afterEach<ModelTestContext>((context) => {
+    context.server.stop();
+  });
+
+  it<ModelTestContext>('connects successfully with the Ably Client', async ({ client, server }) => {
+    server.start();
+    const connectSuccess = await client.connection.whenState('connected');
+    expect(connectSuccess.current).toBe('connected');
+  });
+
+  it<ModelTestContext>('expects the injected client to be of the type RealtimePromise', ({ client }) => {
+    const model = new Model('test', client);
+    expect(model.name).toEqual('test');
+    expectTypeOf(model.client).toMatchTypeOf<Types.RealtimePromise>();
+  });
+
+  it<ModelTestContext>('expects model to be instantiated with the provided event streams', ({ client }) => {
+    const model = new Model('test', client, {
+      streams: [new EventStream('s1', client, { channel: 's1' }), new EventStream('s2', client, { channel: 's2' })],
+    });
+    expect(model.name).toEqual('test');
+    expect(model.stream('s1')).toBeTruthy();
+    expect(model.stream('s1').name).toEqual('s1');
+    expect(model.stream('s2')).toBeTruthy();
+    expect(model.stream('s2').name).toEqual('s2');
+    expect(() => model.stream('s3')).toThrowError("stream with name 's3' not registered on model 'test'");
+  });
+});

src/Model.ts

@@ -1,18 +1,38 @@
 import { Types } from 'ably';
-
+import EventStream from './EventStream';
 import ModelOptions from './options/ModelOptions';
 import EventEmitter from './utilities/EventEmitter';
 
-const MODEL_OPTIONS_DEFAULTS = {};
+enum ModelState {
+  /**
+   * The model has been initialized but has not yet been synchronised.
+   */
+  INITIALIZED = 'initialized',
+  /**
+   * An indefinite failure condition. This state is entered if a channel error has been received from the Ably service, such as an attempt to attach without the necessary access rights.
+   */
+  FAILED = 'failed',
+}
 
-class Model extends EventEmitter<any> {
-  private options: ModelOptions;
-  private connectionId?: string;
+class Model<T> extends EventEmitter<any> {
+  private state: ModelState = ModelState.INITIALIZED;
+  private streams: Record<string, EventStream<any>> = {};
+  private data: T;
 
   constructor(readonly name: string, readonly client: Types.RealtimePromise, options?: ModelOptions) {
     super();
-    this.options = { ...MODEL_OPTIONS_DEFAULTS, ...options };
-    this.connectionId = this.client.connection.id;
+    if (options) {
+      for (let stream of options.streams) {
+        this.streams[stream.name] = stream;
+      }
+    }
+  }
+
+  stream(name: string): EventStream<any> {
+    if (!this.streams[name]) {
+      throw new Error(`stream with name '${name}' not registered on model '${this.name}'`);
+    }
+    return this.streams[name];
   }
 }
 

src/Models.test.ts

@@ -23,17 +23,17 @@ describe('Models', () => {
     context.server.stop();
   });
 
-  it<ModelsTestContext>('expects the injected client to be of the type RealtimePromise', ({ client }) => {
-    const models = new Models(client);
-    expectTypeOf(models.ably).toMatchTypeOf<Types.RealtimePromise>();
-  });
-
   it<ModelsTestContext>('connects successfully with the Ably Client', async ({ client, server }) => {
     server.start();
     const connectSuccess = await client.connection.whenState('connected');
     expect(connectSuccess.current).toBe('connected');
   });
 
+  it<ModelsTestContext>('expects the injected client to be of the type RealtimePromise', ({ client }) => {
+    const models = new Models(client);
+    expectTypeOf(models.ably).toMatchTypeOf<Types.RealtimePromise>();
+  });
+
   it<ModelsTestContext>('creates a client with default options when a key is passed in', () => {
     const models = new Models(defaultClientConfig.key);
     expect(models.ably['options'].key).toEqual(defaultClientConfig.key);
@@ -65,4 +65,44 @@ describe('Models', () => {
       'model-default-client',
     ]);
   });
+
+  it<ModelsTestContext>('creates a model that inherits the root class ably client', () => {
+    const models = new Models({ ...defaultClientConfig });
+    const model = models.Model('test');
+    expect(model.name).toEqual('test');
+    expect(model.client['options']).toContain(defaultClientConfig);
+  });
+
+  it<ModelsTestContext>('getting a model with the same name returns the same instance', () => {
+    const models = new Models({ ...defaultClientConfig });
+    const model1 = models.Model('test');
+    expect(model1.name).toEqual('test');
+    const model2 = models.Model('test');
+    expect(model2.name).toEqual('test');
+    expect(model1).toEqual(model2);
+  });
+
+  it<ModelsTestContext>('creates an event stream that inherits the root class ably client', () => {
+    const models = new Models({ ...defaultClientConfig });
+    const eventStream = models.EventStream('test', { channel: 'foobar' });
+    expect(eventStream.name).toEqual('test');
+    expect(eventStream.client['options']).toContain(defaultClientConfig);
+  });
+
+  it<ModelsTestContext>('getting an event stream without options throws', () => {
+    const models = new Models({ ...defaultClientConfig });
+    expect(() => models.EventStream('test')).toThrow('EventStream cannot be instantiated without options');
+  });
+
+  it<ModelsTestContext>('getting an event stream with the same name returns the same instance', () => {
+    const models = new Models({ ...defaultClientConfig });
+    const eventStream1 = models.EventStream('test', { channel: 'foobar' }); // first call requires options to instantiate
+    expect(eventStream1.name).toEqual('test');
+    const eventStream2 = models.EventStream('test'); // subsequent calls do not require options
+    expect(eventStream2.name).toEqual('test');
+    expect(eventStream1).toEqual(eventStream2);
+    const eventStream3 = models.EventStream('test', { channel: 'barbaz' }); // providing options to subsequent calls is allowed but ignored
+    expect(eventStream3.name).toEqual('test');
+    expect(eventStream1).toEqual(eventStream3);
+  });
 });

src/Models.ts

@@ -1,16 +1,19 @@
 import { Types, Realtime } from 'ably';
 import ModelOptions from './options/ModelOptions';
 import Model from './Model';
+import EventStreamOptions from './options/EventStreamOptions';
+import EventStream from './EventStream';
 
 class Models {
-  private models: Record<string, Model>;
-  private channel: Types.RealtimeChannelPromise;
+  private models: Record<string, Model<any>>;
+  private eventStreams: Record<string, EventStream<any>>;
   ably: Types.RealtimePromise;
 
   readonly version = '0.0.1';
 
   constructor(optionsOrAbly: Types.RealtimePromise | Types.ClientOptions | string) {
     this.models = {};
+    this.eventStreams = {};
     if (optionsOrAbly['options']) {
       this.ably = optionsOrAbly as Types.RealtimePromise;
       this.addAgent(this.ably['options'], false);
@@ -32,22 +35,35 @@ class Models {
     }
   }
 
-  async get(name: string, options?: ModelOptions): Promise<Model> {
+  Model = <T>(name: string, options?: ModelOptions) => {
     if (typeof name !== 'string' || name.length === 0) {
-      throw new Error('Models must have a non-empty name');
+      throw new Error('Model must have a non-empty name');
     }
 
     if (this.models[name]) return this.models[name];
 
-    if (this.ably.connection.state !== 'connected') {
-      await this.ably.connection.once('connected');
-    }
-
-    const model = new Model(name, this.ably, options);
+    const model = new Model<T>(name, this.ably, options);
     this.models[name] = model;
 
     return model;
-  }
+  };
+
+  EventStream = <T>(name: string, options?: EventStreamOptions) => {
+    if (typeof name !== 'string' || name.length === 0) {
+      throw new Error('EventStream must have a non-empty name');
+    }
+
+    if (this.eventStreams[name]) return this.eventStreams[name];
+
+    if (!options) {
+      throw new Error('EventStream cannot be instantiated without options');
+    }
+
+    const eventStream = new EventStream<T>(name, this.ably, options);
+    this.eventStreams[name] = eventStream;
+
+    return eventStream;
+  };
 }
 
 export default Models;

src/options/EventStreamOptions.d.ts

@@ -0,0 +1,5 @@
+type EventStreamOptions = {
+  channel: string;
+};
+
+export default EventStreamOptions;

src/options/ModelOptions.d.ts

@@ -1,3 +1,7 @@
-type ModelOptions = {};
+import EventStream from '../EventStream';
+
+type ModelOptions = {
+  streams: Array<EventStream<any>>;
+};
 
 export default ModelOptions;