Expose subscriberCount in WebSocket server (#13498)

Co-authored-by: Zack Radisic <zack@theradisic.com>

completions/bun.bash

@@ -82,7 +82,7 @@ _bun_completions() {
     declare -A PACKAGE_OPTIONS;
     declare -A PM_OPTIONS;
 
-    local SUBCOMMANDS="dev bun create run install add remove upgrade completions discord help init pm x test repl update link unlink build";
+    local SUBCOMMANDS="dev bun create run install add remove upgrade completions discord help init pm x test repl update outdated link unlink build";
 
     GLOBAL_OPTIONS[LONG_OPTIONS]="--use --cwd --bunfile --server-bunfile --config --disable-react-fast-refresh --disable-hmr --env-file --extension-order --jsx-factory --jsx-fragment --extension-order --jsx-factory --jsx-fragment --jsx-import-source --jsx-production --jsx-runtime --main-fields --no-summary --version --platform --public-dir --tsconfig-override --define --external --help --inject --loader --origin --port --dump-environment-variables --dump-limits --disable-bun-js";
     GLOBAL_OPTIONS[SHORT_OPTIONS]="-c -v -d -e -h -i -l -u -p";

completions/bun.fish

@@ -179,6 +179,7 @@ complete -c bun -n "__fish_use_subcommand" -a "remove" -d "Remove a dependency f
 complete -c bun -n "__fish_use_subcommand" -a "add" -d "Add a dependency to package.json" -f
 complete -c bun -n "__fish_use_subcommand" -a "init" -d "Initialize a Bun project in this directory" -f
 complete -c bun -n "__fish_use_subcommand" -a "link" -d "Register or link a local npm package" -f
-complete -c bun -n "__fish_use_subcommand" -a "link" -d "Unregister a local npm package" -f
+complete -c bun -n "__fish_use_subcommand" -a "unlink" -d "Unregister a local npm package" -f
 complete -c bun -n "__fish_use_subcommand" -a "pm" -d "Additional package management utilities" -f
 complete -c bun -n "__fish_use_subcommand" -a "x" -d "Execute a package binary, installing if needed" -f
+complete -c bun -n "__fish_use_subcommand" -a "outdated" -d "Display the latest versions of outdated dependencies" -f

completions/bun.zsh

@@ -563,6 +563,22 @@ _bun_update_completion() {
     esac
 }
 
+_bun_outdated_completion() {
+    _arguments -s -C \
+        '--cwd[Set a specific cwd]:cwd' \
+        '--verbose[Excessively verbose logging]' \
+        '--no-progress[Disable the progress bar]' \
+        '--help[Print this help menu]' &&
+        ret=0
+    
+    case $state in
+    config)
+        _bun_list_bunfig_toml
+
+        ;;
+    esac
+}
+
 _bun_test_completion() {
     _arguments -s -C \
         '1: :->cmd1' \
@@ -669,6 +685,7 @@ _bun() {
             'add\:"Add a dependency to package.json (bun a)" '
             'remove\:"Remove a dependency from package.json (bun rm)" '
             'update\:"Update outdated dependencies & save to package.json" '
+            'outdated\:"Display the latest versions of outdated dependencies" '
             'link\:"Link an npm package globally" '
             'unlink\:"Globally unlink an npm package" '
             'pm\:"More commands for managing packages" '
@@ -740,6 +757,10 @@ _bun() {
         update)
             _bun_update_completion
 
+            ;;
+        outdated)
+            _bun_outdated_completion
+
             ;;
         'test')
             _bun_test_completion
@@ -819,6 +840,10 @@ _bun() {
                 update)
                     _bun_update_completion
 
+                    ;;
+                outdated)
+                    _bun_outdated_completion
+
                     ;;
                 'test')
                     _bun_test_completion

docs/cli/outdated.md

@@ -0,0 +1,27 @@
+Use `bun outdated` to display a table of outdated dependencies with their latest versions:
+
+```sh
+$ bun outdated
+
+|--------------------------------------------------------------------|
+| Packages                               | Current | Update | Latest |
+|----------------------------------------|---------|--------|--------|
+| @types/bun (dev)                       | 1.1.6   | 1.1.7  | 1.1.7  |
+|----------------------------------------|---------|--------|--------|
+| @types/react (dev)                     | 18.3.3  | 18.3.4 | 18.3.4 |
+|----------------------------------------|---------|--------|--------|
+| @typescript-eslint/eslint-plugin (dev) | 7.16.1  | 7.18.0 | 8.2.0  |
+|----------------------------------------|---------|--------|--------|
+| @typescript-eslint/parser (dev)        | 7.16.1  | 7.18.0 | 8.2.0  |
+|----------------------------------------|---------|--------|--------|
+| esbuild (dev)                          | 0.21.5  | 0.21.5 | 0.23.1 |
+|----------------------------------------|---------|--------|--------|
+| eslint (dev)                           | 9.7.0   | 9.9.1  | 9.9.1  |
+|----------------------------------------|---------|--------|--------|
+| typescript (dev)                       | 5.5.3   | 5.5.4  | 5.5.4  |
+|--------------------------------------------------------------------|
+```
+
+The `Update` column shows the version that would be installed if you ran `bun update [package]`. This version is the latest version that satisfies the version range specified in your `package.json`.
+
+The `Latest` column shows the latest version available from the registry. `bun update --latest [package]` will update to this version.

packages/bun-types/bun.d.ts

@@ -2767,6 +2767,16 @@ declare module "bun" {
       compress?: boolean,
     ): ServerWebSocketSendStatus;
 
+    /**
+     * A count of connections subscribed to a given topic
+     *
+     * This operation will loop through each topic internally to get the count.
+     *
+     * @param topic the websocket topic to check how many subscribers are connected to
+     * @returns the number of subscribers
+     */
+    subscriberCount(topic: string): number;
+
     /**
      * Returns the client IP address and port of the given Request. If the request was closed or is a unix socket, returns null.
      *

src/bun.js/api/server.classes.ts

@@ -16,6 +16,10 @@ function generate(name) {
         fn: "doPublish",
         length: 3,
       },
+      subscriberCount: {
+        fn: "doSubscriberCount",
+        length: 1,
+      },
       reload: {
         fn: "doReload",
         length: 2,

src/bun.js/api/server.zig

@@ -4339,7 +4339,6 @@ pub const ServerWebSocket = struct {
         callframe: *JSC.CallFrame,
     ) JSValue {
         const args = callframe.arguments(4);
-
         if (args.len < 1) {
             log("publish()", .{});
             globalThis.throw("publish requires at least 1 argument", .{});
@@ -5354,6 +5353,31 @@ pub fn NewServer(comptime NamespaceType: type, comptime ssl_enabled_: bool, comp
         pub const doFetch = onFetch;
         pub const doRequestIP = JSC.wrapInstanceMethod(ThisServer, "requestIP", false);
 
+        pub fn doSubscriberCount(this: *ThisServer, globalThis: *JSC.JSGlobalObject, callframe: *JSC.CallFrame) JSC.JSValue {
+            const arguments = callframe.arguments(1);
+            if (arguments.len < 1) {
+                globalThis.throwNotEnoughArguments("subscriberCount", 1, 0);
+                return .zero;
+            }
+
+            if (arguments.ptr[0].isEmptyOrUndefinedOrNull()) {
+                globalThis.throwInvalidArguments("subscriberCount requires a topic name as a string", .{});
+                return .zero;
+            }
+
+            var topic = arguments.ptr[0].toSlice(globalThis, bun.default_allocator);
+            defer topic.deinit();
+            if (globalThis.hasException()) {
+                return .zero;
+            }
+
+            if (topic.len == 0) {
+                return JSValue.jsNumber(0);
+            }
+
+            return JSValue.jsNumber((this.app.num_subscribers(topic.slice())));
+        }
+
         pub usingnamespace NamespaceType;
         pub usingnamespace bun.New(@This());
 

src/cli.zig

@@ -1083,6 +1083,7 @@ pub const HelpCommand = struct {
         \\  <b><blue>add<r>       <d>{s:<16}<r>     Add a dependency to package.json <d>(bun a)<r>
         \\  <b><blue>remove<r>    <d>{s:<16}<r>     Remove a dependency from package.json <d>(bun rm)<r>
         \\  <b><blue>update<r>    <d>{s:<16}<r>     Update outdated dependencies
+        \\  <b><blue>outdated<r>                       Display latest versions of outdated dependencies
         \\  <b><blue>link<r>      <d>[\<package\>]<r>          Register or link a local npm package
         \\  <b><blue>unlink<r>                         Unregister a local npm package
         \\  <b><blue>patch <d>\<pkg\><r>                    Prepare a package for patching

src/cli/outdated_command.zig

@@ -111,6 +111,9 @@ fn Table(
 
 pub const OutdatedCommand = struct {
     pub fn exec(ctx: Command.Context) !void {
+        Output.prettyErrorln("<r><b>bun outdated <r><d>v" ++ Global.package_json_version_with_sha ++ "<r>", .{});
+        Output.flush();
+
         const cli = try PackageManager.CommandLineArguments.parse(ctx.allocator, .outdated);
 
         const manager = PackageManager.init(ctx, cli, .outdated) catch |err| {
@@ -183,9 +186,6 @@ pub const OutdatedCommand = struct {
         if (root_pkg_id == invalid_package_id) return;
         const root_pkg_deps = lockfile.packages.items(.dependencies)[root_pkg_id];
 
-        Output.prettyErrorln("<r><b>bun outdated <r><d>v" ++ Global.package_json_version_with_sha ++ "<r>", .{});
-        Output.flush();
-
         try updateManifestsIfNecessary(manager, log_level, root_pkg_deps);
 
         try switch (Output.enable_ansi_colors) {

src/install/semver.zig

@@ -668,7 +668,7 @@ pub const Version = extern struct {
             const diff = this.version.whichVersionIsDifferent(this.other, this.buf, this.other_buf) orelse .none;
 
             switch (diff) {
-                .major => try writer.print(Output.prettyFmt("<b><red>{d}.{d}.{d}", true), .{
+                .major => try writer.print(Output.prettyFmt("<r><b><red>{d}.{d}.{d}", true), .{
                     this.version.major, this.version.minor, this.version.patch,
                 }),
                 .minor => {
@@ -677,7 +677,7 @@ pub const Version = extern struct {
                             this.version.major, this.version.minor, this.version.patch,
                         });
                     } else {
-                        try writer.print(Output.prettyFmt("<d>{d}.<r><yellow>{d}.{d}", true), .{
+                        try writer.print(Output.prettyFmt("<d>{d}.<r><b><yellow>{d}.{d}", true), .{
                             this.version.major, this.version.minor, this.version.patch,
                         });
                     }

test/cli/install/registry/__snapshots__/bun-install-registry.test.ts.snap

@@ -46,20 +46,11 @@ what-bin@1.0.0:
 "
 `;
 
-exports[`outdated NO_COLOR works 1`] = `
-"|--------------------------------------|
-| Packages | Current | Update | Latest |
-|----------|---------|--------|--------|
-| a-dep    | 1.0.1   | 1.0.1  | 1.0.10 |
-|--------------------------------------|
-"
-`;
-
 exports[`outdated normal dep, smaller than column title 1`] = `
 "┌──────────┬─────────┬────────┬────────┐
 │ Packages │ Current │ Update │ Latest │
 ├──────────┼─────────┼────────┼────────┤
-│ no-deps  │ 1.0.0   │ 1.0.0  │ 2.0.0  │
+│ no-deps  │ 1.0.0   │ 1.0.0  │ 2.0.0  │
 └──────────┴─────────┴────────┴────────┘
 "
 `;
@@ -77,7 +68,7 @@ exports[`outdated dev dep, smaller than column title 1`] = `
 "┌───────────────┬─────────┬────────┬────────┐
 │ Packages      │ Current │ Update │ Latest │
 ├───────────────┼─────────┼────────┼────────┤
-│ no-deps (dev) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
+│ no-deps (dev) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
 └───────────────┴─────────┴────────┴────────┘
 "
 `;
@@ -95,7 +86,7 @@ exports[`outdated peer dep, smaller than column title 1`] = `
 "┌────────────────┬─────────┬────────┬────────┐
 │ Packages       │ Current │ Update │ Latest │
 ├────────────────┼─────────┼────────┼────────┤
-│ no-deps (peer) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
+│ no-deps (peer) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
 └────────────────┴─────────┴────────┴────────┘
 "
 `;
@@ -113,7 +104,7 @@ exports[`outdated optional dep, smaller than column title 1`] = `
 "┌────────────────────┬─────────┬────────┬────────┐
 │ Packages           │ Current │ Update │ Latest │
 ├────────────────────┼─────────┼────────┼────────┤
-│ no-deps (optional) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
+│ no-deps (optional) │ 1.0.0   │ 1.0.0  │ 2.0.0  │
 └────────────────────┴─────────┴────────┴────────┘
 "
 `;
@@ -126,3 +117,12 @@ exports[`outdated optional dep, larger than column title 1`] = `
 └──────────────────────────┴────────────────┴────────────────┴────────────────┘
 "
 `;
+
+exports[`outdated NO_COLOR works 1`] = `
+"|--------------------------------------|
+| Packages | Current | Update | Latest |
+|----------|---------|--------|--------|
+| a-dep    | 1.0.1   | 1.0.1  | 1.0.10 |
+|--------------------------------------|
+"
+`;

test/harness.ts

@@ -5,8 +5,6 @@ import { isAbsolute, join, dirname } from "path";
 import fs, { openSync, closeSync } from "node:fs";
 import os from "node:os";
 import { heapStats } from "bun:jsc";
-import { npm_manifest_test_helpers } from "bun:internal-for-testing";
-const { parseManifest } = npm_manifest_test_helpers;
 
 type Awaitable<T> = T | Promise<T>;
 
@@ -298,7 +296,7 @@ const binaryTypes = {
   "int8array": Int8Array,
   "int16array": Int16Array,
   "int32array": Int32Array,
-  "float16array": Float16Array,
+  "float16array": globalThis.Float16Array,
   "float32array": Float32Array,
   "float64array": Float64Array,
 } as const;
@@ -1268,6 +1266,9 @@ https://buildkite.com/docs/pipelines/security/secrets/buildkite-secrets`;
 }
 
 export function assertManifestsPopulated(absCachePath: string, registryUrl: string) {
+  const { npm_manifest_test_helpers } = require("bun:internal-for-testing");
+  const { parseManifest } = npm_manifest_test_helpers;
+
   for (const file of fs.readdirSync(absCachePath)) {
     if (!file.endsWith(".npm")) continue;
 

test/js/bun/websocket/websocket-server-fixture.js

@@ -8,6 +8,7 @@
 
 let pending = [];
 using server = Bun.serve({
+  port: 0,
   websocket: {
     open(ws) {
       globalThis.sockets ??= [];

test/js/bun/websocket/websocket-server.test.ts

@@ -492,9 +492,11 @@ describe("ServerWebSocket", () => {
           }
         }
       };
-      test(label, (done, connect) => ({
+      test(label, (done, connect, options) => ({
         async open(ws) {
+          const initial = options.server.subscriberCount(topic);
           ws.subscribe(topic);
+          expect(options.server.subscriberCount(topic)).toBe(initial + 1);
           if (ws.data.id === 0) {
             await connect();
           } else if (ws.data.id === 1) {
@@ -525,10 +527,12 @@ describe("ServerWebSocket", () => {
           }
         }
       };
-      test(label, done => ({
+      test(label, (done, _, options) => ({
         publishToSelf: true,
         async open(ws) {
+          const initial = options.server.subscriberCount(topic);
           ws.subscribe(topic);
+          expect(options.server.subscriberCount(topic)).toBe(initial + 1);
           send(ws);
         },
         drain(ws) {
@@ -690,7 +694,11 @@ describe("ServerWebSocket", () => {
 
 function test(
   label: string,
-  fn: (done: (err?: unknown) => void, connect: () => Promise<void>) => Partial<WebSocketHandler<{ id: number }>>,
+  fn: (
+    done: (err?: unknown) => void,
+    connect: () => Promise<void>,
+    options: { server: Server },
+  ) => Partial<WebSocketHandler<{ id: number }>>,
   timeout?: number,
 ) {
   it(
@@ -705,6 +713,9 @@ function test(
         }
       };
       let id = 0;
+      var options = {
+        server: undefined,
+      };
       const server: Server = serve({
         port: 0,
         fetch(request, server) {
@@ -717,9 +728,11 @@ function test(
         websocket: {
           sendPings: false,
           message() {},
-          ...fn(done, () => connect(server)),
+          ...fn(done, () => connect(server), options as any),
         },
       });
+      options.server = server;
+      expect(server.subscriberCount("empty topic")).toBe(0);
       await connect(server);
     },
     { timeout: timeout ?? 1000 },

test/js/node/fs/fs.test.ts

@@ -2376,14 +2376,13 @@ describe("fs/promises", () => {
   }, 100000);
 
   for (let withFileTypes of [false, true] as const) {
-    const warmup = 1;
     const iterCount = 200;
     const full = resolve(import.meta.dir, "../");
 
     const doIt = async () => {
-      for (let i = 0; i < warmup; i++) {
-        await promises.readdir(full, { withFileTypes });
-      }
+      await Promise.all(
+        Array.from({ length: iterCount }, () => promises.readdir(full, { withFileTypes, recursive: true })),
+      );
 
       const maxFD = getMaxFD();
       const pending = new Array(iterCount);