Rename src to source
This commit is contained in:
@@ -0,0 +1,254 @@
|
||||
import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
|
||||
import { PrivateKey } from '@xo-cash/primitives'
|
||||
|
||||
import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js';
|
||||
|
||||
export class SSEClient<PayloadType = unknown> extends BaseSyncClient<PayloadType> {
|
||||
private readonly url: string;
|
||||
public sseSession: SSESession | undefined;
|
||||
private readonly subscriptions: Set<string> = new Set();
|
||||
|
||||
private privateKey: PrivateKey;
|
||||
private textEncoder = new TextEncoder();
|
||||
|
||||
private get subscriptionUrl() {
|
||||
return `${this.url}/data/subscribe`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new SSE client
|
||||
* @param url - The URL of the server
|
||||
* @param privateKey - The private key to use for authentication
|
||||
*/
|
||||
constructor(url: string, privateKey: PrivateKey) {
|
||||
super();
|
||||
|
||||
this.url = url;
|
||||
this.privateKey = privateKey;
|
||||
|
||||
// Whenever we get a new subscription, we need to reconnect the session with the new subscriptions in the body
|
||||
// We add a small debounce to this to prevent parallel subscriptions from causing multiple reconnects
|
||||
this.on('subscriptions-updated', () => {
|
||||
// Nothing awaits this listener, so report failures as an error event instead of an unhandled rejection
|
||||
this.reconnect().catch((error) => this.emit('error', error));
|
||||
}, 25);
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects to the server
|
||||
*/
|
||||
async connect() {
|
||||
// The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet.
|
||||
// subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session.
|
||||
if (this.subscriptions.size === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.sseSession ??= this.createSession();
|
||||
|
||||
await this.sseSession.connect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Disconnects from the server
|
||||
*/
|
||||
async disconnect() {
|
||||
await this.sseSession?.disconnect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconnects to the server
|
||||
*/
|
||||
async reconnect() {
|
||||
// Close the current session. Its subscription list is fixed in the request body, so it can't be reused
|
||||
this.sseSession?.close();
|
||||
this.sseSession = undefined;
|
||||
|
||||
// Connect a new session with the current subscriptions
|
||||
await this.connect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new SSE session
|
||||
* @returns The SSE session
|
||||
*/
|
||||
private createSession() {
|
||||
// Build the request body
|
||||
const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] });
|
||||
|
||||
// Create a new SSE session
|
||||
const session = new SSESession(this.subscriptionUrl, {
|
||||
onRequest: async (request) => {
|
||||
// Get the path from the URL
|
||||
const url = new URL(this.subscriptionUrl);
|
||||
const path = url.pathname;
|
||||
|
||||
// Sign the request body
|
||||
const authHeaders = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, request.body);
|
||||
|
||||
// Initialize the request headers if they don't exist
|
||||
request.headers ??= {};
|
||||
|
||||
// Add the authentication headers to the request
|
||||
request.headers = {
|
||||
...request.headers,
|
||||
...authHeaders,
|
||||
}
|
||||
|
||||
// Return the request with the authentication headers
|
||||
return request;
|
||||
},
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body,
|
||||
method: 'POST',
|
||||
// Reconnect when the server ends the stream (e.g. a restart or a proxy idle timeout). By default only errors are retried
|
||||
persistent: true,
|
||||
});
|
||||
|
||||
// Create a listener to re-emit the connected and disconnected events
|
||||
session.on('connected', this.handleOnConnectionUpdated.bind(this, 'connected'));
|
||||
session.on('disconnected', this.handleOnConnectionUpdated.bind(this, 'disconnected'));
|
||||
|
||||
// Create a listener to re-emit the messages
|
||||
session.on('message', (message) => {
|
||||
const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data));
|
||||
if (!parsedMessage.success) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.emit('resource-instance-updated', {
|
||||
...parsedMessage.data,
|
||||
value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType,
|
||||
});
|
||||
});
|
||||
|
||||
// Create a listener to re-emit the errors
|
||||
session.on('error', (error) => {
|
||||
this.emit('error', error);
|
||||
});
|
||||
|
||||
// Return the session
|
||||
return session;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a POST /data/write request to the server
|
||||
* @param resources - The resources to write
|
||||
* @returns The Resource Object with the decoded value
|
||||
*/
|
||||
async writeMany(resources: Record<string, PayloadType>): Promise<ResourcesRecord<PayloadType>> {
|
||||
// Sign the resource write requests
|
||||
const signedResources = Object.entries(resources).map(([resourceId, value]) => {
|
||||
const valueStr = toExtendedJson(value);
|
||||
const valueBytes = this.textEncoder.encode(valueStr);
|
||||
return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes);
|
||||
});
|
||||
|
||||
// Send the resource write requests to the server
|
||||
const res = await this.sendRequest('/data/write', { resources: signedResources });
|
||||
|
||||
// If the response is not ok, throw an error
|
||||
if (!res.ok) {
|
||||
const error = await res.json();
|
||||
throw new Error(`Failed to write resources [${Object.keys(resources).join(', ')}] (${res.statusText}): ${JSON.stringify(error)}`);
|
||||
}
|
||||
|
||||
// Decode the response and return the resources
|
||||
const decodedResources = await this.decodeResponse(res);
|
||||
|
||||
// Return the decoded resources
|
||||
return decodedResources;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a POST /data/read request to the server
|
||||
* @param resourceIds - The resource IDs to read
|
||||
* @returns The resources with the read values
|
||||
*/
|
||||
async readMany(resourceIds: string[]): Promise<ResourcesRecord<PayloadType>> {
|
||||
// Send the resource read requests to the server
|
||||
const res = await this.sendRequest('/data/get', { resourceId: resourceIds });
|
||||
|
||||
// If the response is not ok, throw an error
|
||||
if (!res.ok) {
|
||||
throw new Error(`Failed to read resources [${resourceIds.join(', ')}] (${res.statusText})`);
|
||||
}
|
||||
|
||||
// Decode the response and return the resources
|
||||
const decodedResources = await this.decodeResponse(res);
|
||||
|
||||
// Return the decoded resources
|
||||
return decodedResources;
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list
|
||||
* @param resourceId - The resource ID to add to the subscription list
|
||||
*/
|
||||
async subscribe(resourceId: string) {
|
||||
this.subscriptions.add(resourceId);
|
||||
|
||||
this.emit('subscriptions-updated', this.subscriptions);
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroy the current SSE Session, then re-create it without the resource ID in the subscription list
|
||||
* @param resourceId - The resource ID to remove from the subscription list
|
||||
*/
|
||||
async unsubscribe(resourceId: string) {
|
||||
this.subscriptions.delete(resourceId);
|
||||
|
||||
this.emit('subscriptions-updated', this.subscriptions);
|
||||
}
|
||||
|
||||
async handleOnConnectionUpdated(value: 'connected' | 'disconnected') {
|
||||
this.emit('connection-updated', value);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a request to the server
|
||||
* @param path - The path to send the request to
|
||||
* @param body - The body of the request
|
||||
* @returns The response from the server
|
||||
*/
|
||||
private async sendRequest(path: string, body: unknown): Promise<Response> {
|
||||
// Build the URL
|
||||
const url = `${this.url}${path}`;
|
||||
|
||||
// Format the body as Extended JSON
|
||||
const formattedBody = toExtendedJson(body);
|
||||
|
||||
// Send the request to the server
|
||||
const res = await fetch(url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...BaseSyncClient.signRequest(this.privateKey.toBytes(), path, body),
|
||||
},
|
||||
body: formattedBody,
|
||||
});
|
||||
|
||||
// Return the response
|
||||
return res;
|
||||
}
|
||||
|
||||
/**
|
||||
* Decodes the blobs on the response body to the type T
|
||||
* @param response - The response to decode
|
||||
* @returns The decoded resources
|
||||
*/
|
||||
private async decodeResponse(response: Response): Promise<ResourcesRecord<PayloadType>> {
|
||||
// Decode the response from Extended JSON
|
||||
const text = fromExtendedJson(await response.text())
|
||||
|
||||
// Parse the response as a record of resources
|
||||
const resources = BaseSyncClient.resourcesResponseSchema.parse(text);
|
||||
|
||||
// Decode the blobs and convert them from extended json to the type T
|
||||
const decodedResources = BaseSyncClient.decodeResourcesResponse<PayloadType>(resources);
|
||||
|
||||
return decodedResources;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
export * from './client.js';
|
||||
Reference in New Issue
Block a user