/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 - 2 0 2 0 ĠJetBrains Ġs . r . o . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ package Ġkotlinx . coroutines . flow Ċ Ċ import Ġkotlinx . atomic fu .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġorg . junit .* Ċ import Ġorg . junit . Test Ċ import Ġkotlin . collections . ArrayList Ċ import Ġkotlin . test .* Ċ import Ġkotlin . time .* Ċ Ċ @ Experimental Time Ċ class ĠShared Flow St ress Test Ġ: ĠTest Base () Ġ{ ĊĠĠĠ Ġprivate Ġval Ġn Pro ducers Ġ= Ġ 5 ĊĠĠĠ Ġprivate Ġval Ġn Consum ers Ġ= Ġ 3 ĊĠĠĠ Ġprivate Ġval Ġn Seconds Ġ= Ġ 3 Ġ* Ġstress Test Multiplier ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġsf : ĠMutable Shared Flow < Long > ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġview : ĠShared Flow < Long > ĊĊĠĠĠ Ġ@ get : Rule ĊĠĠĠ Ġval Ġproducer Dispatcher Ġ= ĠExecutor Rule ( n Pro ducers ) ĊĠĠĠ Ġ@ get : Rule ĊĠĠĠ Ġval Ġconsumer Dispatcher Ġ= ĠExecutor Rule ( n Consum ers ) ĊĊĠĠĠ Ġprivate Ġval Ġtotal Pro duced Ġ= Ġatomic ( 0 L ) ĊĠĠĠ Ġprivate Ġval Ġtotal Cons umed Ġ= Ġatomic ( 0 L ) ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest St ress Replay 1 () Ġ= ĊĠĠĠĠĠĠĠ Ġtest St ress ( 1 , Ġ 0 ) ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest St ress Replay 1 Extra Buffer 1 () Ġ= ĊĠĠĠĠĠĠĠ Ġtest St ress ( 1 , Ġ 1 ) ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest St ress Replay 2 Extra Buffer 1 () Ġ= ĊĠĠĠĠĠĠĠ Ġtest St ress ( 2 , Ġ 1 ) ĊĊĠĠĠ Ġprivate Ġfun Ġtest St ress ( replay : ĠInt , Ġextra Buffer Capacity : ĠInt ) Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġsf Ġ= ĠMutable Shared Flow ( replay , Ġextra Buffer Capacity ) ĊĠĠĠĠĠĠĠ Ġview Ġ= Ġsf . as Shared Flow () ĊĠĠĠĠĠĠĠ Ġval Ġjobs Ġ= ĠArrayList < Job >() ĊĠĠĠĠĠĠĠ Ġjobs Ġ+= ĠList ( n Pro ducers ) Ġ{ Ġproducer Index Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( producer Dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġcur Ġ= Ġproducer Index . to Long () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsf . emit ( cur ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Pro duced . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcur Ġ+= Ġn Pro ducers ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġjobs Ġ+= ĠList ( n Consum ers ) Ġ{ Ġconsumer Index Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( consumer Dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġview ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. drop While Ġ{ Ġit Ġ% Ġn Consum ers Ġ!= Ġconsumer Index . to Long () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. take ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. collect Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck ( it Ġ% Ġn Consum ers Ġ== Ġconsumer Index . to Long ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtotal Cons umed . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvar Ġlast Pro duced Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġvar Ġlast Cons umed Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġfor Ġ( sec Ġin Ġ 1 .. n Seconds ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Duration . seconds ( 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproduced Ġ= Ġtotal Pro duced . value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumed Ġ= Ġtotal Cons umed . value ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("$ sec Ġsec : Ġproduced Ġ= Ġ$ pro duced ; Ġconsumed Ġ= Ġ$ consumed ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Not Equals ( last Pro duced , Ġproduced ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Not Equals ( last Cons umed , Ġconsumed ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Pro duced Ġ= Ġproduced ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Cons umed Ġ= Ġconsumed ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġjobs . forEach Ġ{ Ġit . cancel () Ġ} ĊĠĠĠĠĠĠĠ Ġjobs . forEach Ġ{ Ġit . join () Ġ} ĊĠĠĠĠĠĠĠ Ġprintln (" total : Ġproduced Ġ= Ġ${ total Pro duced . value }; Ġconsumed Ġ= Ġ${ total Cons umed . value }") ĊĠĠĠ Ġ} Ċ } Ċ