chore: initial commit

.gitignore

@@ -0,0 +1,17 @@
+# Build
+build
+
+# Node & Dependencies
+node_modules
+coverage
+package-lock.json
+pnpm-lock.yaml
+
+# Build tools specific
+npm-debug.log
+yarn-error.log
+
+# Editors specific
+.fleet
+.idea
+.vscode

LICENSE.md

@@ -0,0 +1,9 @@
+# The MIT License
+
+Copyright 2024 Romain Lanz, contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

bin/test.ts

@@ -0,0 +1,10 @@
+import { configure, processCLIArgs, run } from '@japa/runner'
+import { assert } from '@japa/assert'
+
+processCLIArgs(process.argv.splice(2))
+configure({
+  files: ['tests/**/*.spec.ts'],
+  plugins: [assert()],
+})
+
+void run()

package.json

@@ -0,0 +1,97 @@
+{
+  "name": "@boringnode/transmit",
+  "description": "A framework agnostic Server-Sent-Event library",
+  "version": "0.1.0",
+  "engines": {
+    "node": ">=20.11.1"
+  },
+  "main": "build/index.js",
+  "type": "module",
+  "files": [
+    "build"
+  ],
+  "exports": {
+    ".": "./build/index.js",
+    "./transports": "./build/transports.js",
+    "./types": "./build/src/types/main.js"
+  },
+  "scripts": {
+    "build": "pnpm compile",
+    "clean": "del-cli build",
+    "compile": "pnpm clean && tsc",
+    "format": "prettier --write .",
+    "lint": "eslint . --ext=.ts",
+    "precompile": "pnpm lint",
+    "prepublishOnly": "pnpm build",
+    "pretest": "pnpm lint",
+    "quick:test": "node --enable-source-maps --loader=ts-node/esm bin/test.ts",
+    "release": "release-it",
+    "test": "c8 pnpm quick:test",
+    "typecheck": "tsc --noEmit",
+    "version": "npm run build"
+  },
+  "dependencies": {
+    "@boringnode/bus": "^0.6.0",
+    "@poppinss/utils": "^6.7.3",
+    "emittery": "^1.0.3",
+    "matchit": "^1.1.0"
+  },
+  "devDependencies": {
+    "@adonisjs/eslint-config": "^1.3.0",
+    "@adonisjs/prettier-config": "^1.3.0",
+    "@adonisjs/tsconfig": "^1.3.0",
+    "@japa/assert": "^3.0.0",
+    "@japa/runner": "^3.1.4",
+    "@swc/core": "^1.6.13",
+    "@types/node": "^20.14.10",
+    "c8": "^10.1.2",
+    "del-cli": "^5.1.0",
+    "eslint": "^8.57.0",
+    "prettier": "^3.3.2",
+    "release-it": "^17.5.0",
+    "ts-node": "^10.9.2",
+    "tsup": "^8.1.0",
+    "typescript": "^5.5.3"
+  },
+  "author": "Romain Lanz <[email protected]>",
+  "license": "MIT",
+  "keywords": [
+    "sse",
+    "server-sent-event",
+    "realtime",
+    "real-time"
+  ],
+  "eslintConfig": {
+    "extends": "@adonisjs/eslint-config/package"
+  },
+  "prettier": "@adonisjs/prettier-config",
+  "publishConfig": {
+    "access": "public",
+    "tag": "latest"
+  },
+  "release-it": {
+    "git": {
+      "commitMessage": "chore(release): ${version}",
+      "tagAnnotation": "v${version}",
+      "tagName": "v${version}"
+    },
+    "github": {
+      "release": true,
+      "releaseName": "v${version}",
+      "web": true
+    }
+  },
+  "c8": {
+    "reporter": [
+      "text",
+      "html"
+    ],
+    "exclude": [
+      "tests/**"
+    ]
+  },
+  "volta": {
+    "node": "20.11.1"
+  },
+  "packageManager": "[email protected]+sha512.140036830124618d624a2187b50d04289d5a087f326c9edfc0ccd733d76c4f52c3a313d4fc148794a2a9d81553016004e6742e8cf850670268a7387fc220c903"
+}

src/storage.ts

@@ -0,0 +1,135 @@
+/*
+ * @boringnode/transmit
+ *
+ * @license MIT
+ * @copyright Boring Node
+ */
+
+import matchit from 'matchit'
+import { Stream } from './stream.js'
+import type { Route } from 'matchit'
+
+export class Storage {
+  /**
+   * Channels subscribed to a given stream
+   */
+  #subscriptions = new Map<Stream, Set<string>>()
+
+  /**
+   * Channels subscribed to a given Stream UID
+   */
+  #channelsByUid = new Map<string, Set<string>>()
+
+  /**
+   * Secured channels definition
+   */
+  #securedChannelsDefinition: Route[] = []
+
+  /**
+   * Secure a channel
+   */
+  secure(channel: string) {
+    const encodedDefinition = matchit.parse(channel)
+
+    this.#securedChannelsDefinition.push(encodedDefinition)
+  }
+
+  /**
+   * Check if a channel is secured and return the matched channel
+   */
+  getSecuredChannelDefinition(channel: string) {
+    const matchedChannel = matchit.match(channel, this.#securedChannelsDefinition)
+
+    if (matchedChannel.length > 0) {
+      const params = matchit.exec(channel, matchedChannel)
+      return { params, channel: matchedChannel[0].old }
+    }
+  }
+
+  /**
+   * Get the number of secured channels
+   */
+  getSecuredChannelCount() {
+    return this.#securedChannelsDefinition.length
+  }
+
+  /**
+   * Get the number of streams
+   */
+  getStreamCount() {
+    return this.#subscriptions.size
+  }
+
+  /**
+   * Add a stream to the storage
+   */
+  add(stream: Stream) {
+    const channels = new Set<string>()
+
+    this.#subscriptions.set(stream, channels)
+    this.#channelsByUid.set(stream.getUid(), channels)
+  }
+
+  /**
+   * Remove a stream from the storage
+   */
+  remove(stream: Stream) {
+    this.#subscriptions.delete(stream)
+    this.#channelsByUid.delete(stream.getUid())
+  }
+
+  /**
+   * Add a channel to a stream
+   */
+  subscribe(uid: string, channel: string) {
+    const channels = this.#channelsByUid.get(uid)
+
+    if (!channels) return false
+
+    channels.add(channel)
+
+    return true
+  }
+
+  /**
+   * Remove a channel from a stream
+   */
+  unsubscribe(uid: string, channel: string) {
+    const channels = this.#channelsByUid.get(uid)
+
+    if (!channels) return false
+
+    channels.delete(channel)
+
+    return true
+  }
+
+  /**
+   * Find all subscribers to a channel
+   */
+  findByChannel(channel: string) {
+    const subscribers = new Set<Stream>()
+
+    for (const [stream, streamChannels] of this.#subscriptions) {
+      if (streamChannels.has(channel)) {
+        subscribers.add(stream)
+      }
+    }
+
+    return subscribers
+  }
+
+  /**
+   * Get channels for a given client
+   */
+  getChannelByClient(uid: string) {
+    return this.#channelsByUid.get(uid)
+  }
+
+  /**
+   * Get all subscribers
+   */
+  getAllSubscribers() {
+    return this.#subscriptions
+  }
+}

src/stream.ts

@@ -0,0 +1,90 @@
+/*
+ * @boringnode/transmit
+ *
+ * @license MIT
+ * @copyright Boring Node
+ */
+
+import { Transform } from 'node:stream'
+import { dataToString } from './utils.js'
+import type { IncomingMessage, OutgoingHttpHeaders } from 'node:http'
+import type { Broadcastable } from './types/main.js'
+
+interface Message {
+  data: Broadcastable
+}
+
+interface WriteHeaders {
+  writeHead?(statusCode: number, headers?: OutgoingHttpHeaders): WriteHeaders
+  flushHeaders?(): void
+}
+
+export type HeaderStream = NodeJS.WritableStream & WriteHeaders
+
+export class Stream extends Transform {
+  readonly #uid: string
+
+  constructor(uid: string, request?: IncomingMessage) {
+    super({ objectMode: true })
+
+    this.#uid = uid
+
+    if (request?.socket) {
+      request.socket.setKeepAlive(true)
+      request.socket.setNoDelay(true)
+      request.socket.setTimeout(0)
+    }
+  }
+
+  getUid() {
+    return this.#uid
+  }
+
+  pipe<T extends HeaderStream>(
+    destination: T,
+    options?: { end?: boolean },
+    forwardHeaders?: Record<string, any>
+  ): T {
+    if (destination.writeHead) {
+      // @see https://github.com/dunglas/mercure/blob/9e080c8dc9a141d4294412d14efdecfb15bf7f43/subscribe.go#L219
+      destination.writeHead(200, {
+        ...forwardHeaders,
+        'Cache-Control': 'private, no-cache, no-store, must-revalidate, max-age=0, no-transform',
+        'Connection': 'keep-alive',
+        'Content-Type': 'text/event-stream',
+        'Expire': '0',
+        'Pragma': 'no-cache',
+        // @see https://www.nginx.com/resources/wiki/start/topics/examples/x-accel/#x-accel-buffering
+        'X-Accel-Buffering': 'no',
+      })
+
+      destination.flushHeaders?.()
+    }
+
+    // Some clients (Safari) don't trigger onopen until the first frame is received.
+    destination.write(':ok\n\n')
+    return super.pipe(destination, options)
+  }
+
+  _transform(
+    message: Message,
+    _encoding: string,
+    callback: (error?: Error | null, data?: any) => void
+  ) {
+    if (message.data) {
+      this.push(dataToString(message.data))
+    }
+
+    this.push('\n')
+
+    callback()
+  }
+
+  writeMessage(
+    message: Message,
+    encoding?: BufferEncoding,
+    cb?: (error: Error | null | undefined) => void
+  ): boolean {
+    return this.write(message, encoding, cb)
+  }
+}