import type { Task, Subscription } from 'effection'; import { createFuture, createChannel, run, sleep } from 'effection'; import type { SubscribePayload } from 'graphql-ws'; import { createClient as createWSClient } from 'graphql-ws'; import webSocketImpl from 'isomorphic-ws'; import type { GraphQLError } from 'graphql'; export interface SimulationOptions { options?: Record; services?: Record; key?: string debug?: boolean; } export interface Client { createSimulation(simulator: string, options?: SimulationOptions): Promise; destroySimulation(simulation: Simulation): Promise; given(simulation: Simulation, scenario: string, params?: Record): Promise>; state(): AsyncIterable & AsyncIterator; dispose(): Promise; } export interface Simulation { id: string; status: 'new' | 'running' | 'failed', services: Service[]; } export interface WebSocketImpl { new(url: string): { send(data: string | ArrayBufferLike | Blob | ArrayBufferView): void; } } export interface Scenario { id: string; status: 'running' | 'failed'; data: T; } export interface Service { name: string; url: string; } interface Runnable { run(scope: Task): T; } interface Result { data: T; errors?: GraphQLError[]; } export function createClient(serverURL: string): Client { let wsurl = new URL(serverURL); wsurl.protocol = 'ws'; let url = wsurl.toString(); let ws = createWSClient({ url, webSocketImpl }); let scope = run(function*() { try { yield; } finally { let dispose = ws.dispose(); if (dispose) { yield dispose; } } }); function subscribe(payload: SubscribePayload): Runnable>> { return { run(scope: Task) { let { send, close, stream } = createChannel>(); let { future, produce } = createFuture(); scope.run(function*() { let unsubscribe = ws.subscribe>(payload, { next: send, complete: () => produce({ state: "completed", value: undefined }), error: () => null }); try { yield future; } finally { close(); unsubscribe(); } }); return stream.subscribe(scope); } }; } async function query(field: string, payload: SubscribePayload): Promise { return scope.run(function*(child) { yield sleep(10); let subscription = subscribe>(payload).run(child); let result: Result> = yield subscription.expect(); return result.data[field]; }); } return { createSimulation: async (simulator: string, options?: SimulationOptions) => { return query("createSimulation", { query: ` mutation CreateSimulation($simulator: String, $options: JSON, $debug: Boolean) { createSimulation(simulator: $simulator, options: $options, debug: $debug) { id simulators status services { name url } } }`, operationName: 'CreateSimulation', variables: { simulator, options, debug: !!options?.debug } }); }, given: (simulation: Simulation, scenario: string, params: Record = {}) => query>("given", { query: ` mutation Given($simulation: String!, $scenario: String, $params: JSON) { given(a: $scenario, simulation: $simulation, params: $params) } `, variables: { scenario, simulation: simulation.id, params } }), destroySimulation: ({ id }: Simulation) => query("destroySimulation", { query: ` mutation DestroySimulation($id: String!) { destroySimulation(id: $id) }`, variables: { id } }), state() { let child = scope.run(); let subscription = subscribe({ query: `subscription { state }` }).run(child); let iterator = { next() { return child.run(function*() { let next: IteratorResult> = yield subscription.next(); if (next.done) { return { done: true }; } else { return { done: false, value: next.value.data.state }; } }) as Promise>; }, cancel: () => child.halt(), [Symbol.asyncIterator]: () => iterator }; return iterator; }, dispose: () => scope.halt() }; }