1+ import { Buffer } from 'node:buffer' ;
12import type net from 'node:net' ;
3+ import type stream from 'node:stream' ;
24
35const targetBytesWritten = Symbol ( 'targetBytesWritten' ) ;
46const targetBytesRead = Symbol ( 'targetBytesRead' ) ;
57const targets = Symbol ( 'targets' ) ;
68const calculateTargetStats = Symbol ( 'calculateTargetStats' ) ;
79
10+ // Per-target manual counters. Used as a fallback when the underlying runtime
11+ // doesn't populate `socket.bytesRead` / `socket.bytesWritten` for the kind of
12+ // socket we got (e.g. Bun 1.3 currently reports 0 on http client sockets).
13+ const manualBytesRead = Symbol ( 'manualBytesRead' ) ;
14+ const manualBytesWritten = Symbol ( 'manualBytesWritten' ) ;
15+
816type Stats = { bytesWritten : number | null , bytesRead : number | null } ;
917
1018/**
@@ -20,23 +28,81 @@ interface Extras {
2028 [ calculateTargetStats ] : ( ) => Stats ;
2129}
2230
31+ interface TargetExtras {
32+ [ manualBytesRead ] : number ;
33+ [ manualBytesWritten ] : number ;
34+ }
35+
36+ const byteLength = ( chunk : Buffer | string ) : number => {
37+ return Buffer . isBuffer ( chunk ) ? chunk . length : Buffer . byteLength ( chunk ) ;
38+ } ;
39+
40+ const liveBytesRead = ( target : SocketWithPreviousStats & TargetExtras ) : number => {
41+ const fromCounter = ( target . bytesRead || 0 ) - ( target . previousBytesRead || 0 ) ;
42+ return Math . max ( fromCounter , target [ manualBytesRead ] || 0 ) ;
43+ } ;
44+
45+ const liveBytesWritten = ( target : SocketWithPreviousStats & TargetExtras ) : number => {
46+ const fromCounter = ( target . bytesWritten || 0 ) - ( target . previousBytesWritten || 0 ) ;
47+ return Math . max ( fromCounter , target [ manualBytesWritten ] || 0 ) ;
48+ } ;
49+
2350export const countTargetBytes = (
2451 sourceSocket : net . Socket ,
25- target : SocketWithPreviousStats ,
52+ targetSocket : SocketWithPreviousStats ,
2653 registerCloseHandler ?: ( handler : ( ) => void ) => void ,
2754) : void => {
2855 const source = sourceSocket as net . Socket & Extras ;
56+ const target = targetSocket as SocketWithPreviousStats & TargetExtras ;
2957
3058 source [ targetBytesWritten ] = source [ targetBytesWritten ] || 0 ;
3159 source [ targetBytesRead ] = source [ targetBytesRead ] || 0 ;
3260 source [ targets ] = source [ targets ] || new Set ( ) ;
3361
3462 source [ targets ] . add ( target ) ;
3563
36- const closeHandler = ( ) => {
37- source [ targetBytesWritten ] += ( target . bytesWritten - ( target . previousBytesWritten || 0 ) ) ;
38- source [ targetBytesRead ] += ( target . bytesRead - ( target . previousBytesRead || 0 ) ) ;
64+ // Manual byte counters via passive listeners. Zero overhead on Node.js
65+ // (the socket-level counters always win in `liveBytesRead/Written`) and
66+ // the only source of truth on Bun 1.3 where http client/server sockets
67+ // report `bytesRead`/`bytesWritten` as 0.
68+ target [ manualBytesRead ] = target [ manualBytesRead ] || 0 ;
69+ target [ manualBytesWritten ] = target [ manualBytesWritten ] || 0 ;
70+
71+ const onRead = ( chunk : Buffer | string ) : void => {
72+ target [ manualBytesRead ] += byteLength ( chunk ) ;
73+ } ;
74+ target . on ( 'data' , onRead ) ;
75+
76+ // Bytes _written_ to target equal bytes _emitted_ by whatever is piped
77+ // into it. Track via the source readable's 'data' events when the pipe is
78+ // wired up; detach when unpiped or on close.
79+ const trackedSources = new Map < stream . Readable , ( chunk : Buffer | string ) => void > ( ) ;
80+ const onPipe = ( src : stream . Readable ) : void => {
81+ const onSrcData = ( chunk : Buffer | string ) : void => {
82+ target [ manualBytesWritten ] += byteLength ( chunk ) ;
83+ } ;
84+ trackedSources . set ( src , onSrcData ) ;
85+ src . on ( 'data' , onSrcData ) ;
86+ } ;
87+ const onUnpipe = ( src : stream . Readable ) : void => {
88+ const handler = trackedSources . get ( src ) ;
89+ if ( handler ) {
90+ src . off ( 'data' , handler ) ;
91+ trackedSources . delete ( src ) ;
92+ }
93+ } ;
94+ target . on ( 'pipe' , onPipe ) ;
95+ target . on ( 'unpipe' , onUnpipe ) ;
96+
97+ const closeHandler = ( ) : void => {
98+ source [ targetBytesWritten ] += liveBytesWritten ( target ) ;
99+ source [ targetBytesRead ] += liveBytesRead ( target ) ;
39100 source [ targets ] . delete ( target ) ;
101+ target . off ( 'data' , onRead ) ;
102+ target . off ( 'pipe' , onPipe ) ;
103+ target . off ( 'unpipe' , onUnpipe ) ;
104+ for ( const [ src , handler ] of trackedSources ) src . off ( 'data' , handler ) ;
105+ trackedSources . clear ( ) ;
40106 } ;
41107 if ( ! registerCloseHandler ) {
42108 registerCloseHandler = ( handler : ( ) => void ) => target . once ( 'close' , handler ) ;
@@ -49,8 +115,8 @@ export const countTargetBytes = (
49115 let bytesRead = source [ targetBytesRead ] ;
50116
51117 for ( const socket of source [ targets ] ) {
52- bytesWritten += ( socket . bytesWritten - ( socket . previousBytesWritten || 0 ) ) ;
53- bytesRead += ( socket . bytesRead - ( socket . previousBytesRead || 0 ) ) ;
118+ bytesWritten += liveBytesWritten ( socket as SocketWithPreviousStats & TargetExtras ) ;
119+ bytesRead += liveBytesRead ( socket as SocketWithPreviousStats & TargetExtras ) ;
54120 }
55121
56122 return {
0 commit comments