action-cable-source.ts

import { Source, SourceSettings, buildTransform } from '@orbit/data';
import {
  createConsumer,
  Cable,
  ChannelNameWithParams
} from '@rails/actioncable';
import { JSONAPISerializer, ResourceOperationsDocument } from '@orbit/jsonapi';

export interface ActionCableSourceSettings extends SourceSettings {
  cable?: Cable;
}

export class ActionCableSource extends Source {
  consumer: Cable;
  serializer: JSONAPISerializer;

  constructor(settings: ActionCableSourceSettings) {
    super(settings);

    this.serializer = new JSONAPISerializer({
      schema: this.schema
    });
    this.consumer = settings.cable as Cable;
  }

  async deactivate() {
    await super.deactivate();
    this.consumer.disconnect();
  }

  subscribe(query: Query) {
    const channel = { channel: 'QueryChannel', expressions: query.expressions, options: query.options };
    return this.consumer.subscriptions.create(channel, {
      received: async (data: ResourceOperationsDocument) => {
        const operations = this.serializer.deserializeOperationsDocument(data);
        const transform = buildTransform(operations);
        await this.transformed([transform]);
      }
    });
  }
}