@@ -92,7 +92,7 @@ class BroadcastImpl {
9292 #options;
9393 #writer = null ;
9494 #cachedMinCursor = 0 ;
95- #minCursorDirty = false ;
95+ #cachedMinCursorConsumers = 0 ;
9696
9797 constructor ( options ) {
9898 this . #options = options ;
@@ -150,13 +150,13 @@ class BroadcastImpl {
150150 } ;
151151
152152 this . #consumers. add ( state ) ;
153- // New consumer starts at buffer start; recalculate min cursor
154- // since this consumer may now be the slowest.
155153 if ( this . #consumers. size === 1 ) {
156154 this . #cachedMinCursor = state . cursor ;
157- this . #minCursorDirty = false ;
155+ this . #cachedMinCursorConsumers = 1 ;
156+ } else if ( state . cursor === this . #cachedMinCursor) {
157+ this . #cachedMinCursorConsumers++ ;
158158 } else {
159- this . #minCursorDirty = true ;
159+ this . #recomputeMinCursor ( ) ;
160160 }
161161 const self = this ;
162162
@@ -167,9 +167,9 @@ class BroadcastImpl {
167167 state . detached = true ;
168168 state . resolve = null ;
169169 state . reject = null ;
170- self . #consumers . delete ( state ) ;
171- self . #minCursorDirty = true ;
172- self . #tryTrimBuffer ( ) ;
170+ if ( self . #deleteConsumer ( state ) ) {
171+ self . #tryTrimBuffer ( ) ;
172+ }
173173 }
174174
175175 return {
@@ -186,19 +186,19 @@ class BroadcastImpl {
186186 const bufferIndex = state . cursor - self . #bufferStart;
187187 if ( bufferIndex < self . #buffer. length ) {
188188 const chunk = self . #buffer. get ( bufferIndex ) ;
189- // If this consumer was at the min cursor, mark dirty
190- if ( state . cursor <= self . #cachedMinCursor) {
191- self . #minCursorDirty = true ;
192- }
189+ const cursor = state . cursor ;
193190 state . cursor ++ ;
194- self . #tryTrimBuffer( ) ;
191+ if ( cursor === self . #cachedMinCursor &&
192+ -- self . #cachedMinCursorConsumers === 0 ) {
193+ self . #tryTrimBuffer( ) ;
194+ }
195195 return PromiseResolve (
196196 { __proto__ : null , done : false , value : chunk } ) ;
197197 }
198198
199199 if ( self . #error) {
200200 state . detached = true ;
201- self . #consumers . delete ( state ) ;
201+ self . #deleteConsumer ( state ) ;
202202 return PromiseReject ( self . #error) ;
203203 }
204204
@@ -253,6 +253,7 @@ class BroadcastImpl {
253253 consumer . detached = true ;
254254 }
255255 this . #consumers. clear ( ) ;
256+ this . #cachedMinCursorConsumers = 0 ;
256257 }
257258
258259 [ SymbolDispose ] ( ) {
@@ -274,9 +275,11 @@ class BroadcastImpl {
274275 this . #bufferStart++ ;
275276 for ( const consumer of this . #consumers) {
276277 if ( consumer . cursor < this . #bufferStart) {
278+ this . #deleteConsumerFromMin( consumer ) ;
277279 consumer . cursor = this . #bufferStart;
278280 }
279281 }
282+ this . #recomputeMinCursor( ) ;
280283 break ;
281284 case 'drop-newest' :
282285 return true ;
@@ -297,7 +300,12 @@ class BroadcastImpl {
297300 const bufferIndex = consumer . cursor - this . #bufferStart;
298301 if ( bufferIndex < this . #buffer. length ) {
299302 const chunk = this . #buffer. get ( bufferIndex ) ;
303+ const cursor = consumer . cursor ;
300304 consumer . cursor ++ ;
305+ if ( cursor === this . #cachedMinCursor &&
306+ -- this . #cachedMinCursorConsumers === 0 ) {
307+ this . #tryTrimBuffer( ) ;
308+ }
301309 consumer . resolve ( { __proto__ : null , done : false , value : chunk } ) ;
302310 } else {
303311 consumer . resolve ( { __proto__ : null , done : true , value : undefined } ) ;
@@ -323,6 +331,7 @@ class BroadcastImpl {
323331 consumer . detached = true ;
324332 }
325333 this . #consumers. clear ( ) ;
334+ this . #cachedMinCursorConsumers = 0 ;
326335 }
327336
328337 [ kGetDesiredSize ] ( ) {
@@ -343,14 +352,14 @@ class BroadcastImpl {
343352 // Private methods
344353
345354 #recomputeMinCursor( ) {
346- const { minCursor } = getMinCursor (
355+ const { minCursor, minCursorConsumers } = getMinCursor (
347356 this . #consumers, this . #bufferStart + this . #buffer. length ) ;
348357 this . #cachedMinCursor = minCursor ;
349- this . #minCursorDirty = false ;
358+ this . #cachedMinCursorConsumers = minCursorConsumers ;
350359 }
351360
352361 #tryTrimBuffer( ) {
353- if ( this . #minCursorDirty ) {
362+ if ( this . #cachedMinCursorConsumers === 0 ) {
354363 this . #recomputeMinCursor( ) ;
355364 }
356365 const trimCount = this . #cachedMinCursor - this . #bufferStart;
@@ -377,10 +386,12 @@ class BroadcastImpl {
377386 const bufferIndex = consumer . cursor - this . #bufferStart;
378387 if ( bufferIndex < this . #buffer. length ) {
379388 const chunk = this . #buffer. get ( bufferIndex ) ;
380- if ( consumer . cursor <= this . #cachedMinCursor) {
381- this . #minCursorDirty = true ;
382- }
389+ const cursor = consumer . cursor ;
383390 consumer . cursor ++ ;
391+ if ( cursor === this . #cachedMinCursor &&
392+ -- this . #cachedMinCursorConsumers === 0 ) {
393+ this . #tryTrimBuffer( ) ;
394+ }
384395 const resolve = consumer . resolve ;
385396 consumer . resolve = null ;
386397 consumer . reject = null ;
@@ -392,6 +403,21 @@ class BroadcastImpl {
392403 }
393404 }
394405 }
406+
407+ #deleteConsumerFromMin( consumer ) {
408+ if ( consumer . cursor === this . #cachedMinCursor) {
409+ this . #cachedMinCursorConsumers-- ;
410+ return this . #cachedMinCursorConsumers === 0 ;
411+ }
412+ return false ;
413+ }
414+
415+ #deleteConsumer( consumer ) {
416+ if ( this . #consumers. delete ( consumer ) ) {
417+ return this . #deleteConsumerFromMin( consumer ) ;
418+ }
419+ return false ;
420+ }
395421}
396422
397423// =============================================================================
0 commit comments