package Ġde . br ing me ister . spring . aws . k inesis Ċ Ċ import Ġcom . amazonaws . services . k inesis . Amazon K inesis Ċ import Ġcom . amazonaws . services . k inesis . model . Describe Stream Result Ċ import Ġcom . amazonaws . services . k inesis . model . Limit Exceeded Exception Ċ import Ġcom . amazonaws . services . k inesis . model . Resource NotFoundException Ċ import Ġcom . amazonaws . services . k inesis . model . Stream Description Ċ import Ġcom . n ha ar man . mockito _ kotlin . do Throw Ċ import Ġcom . n ha ar man . mockito _ kotlin . mock Ċ import Ġcom . n ha ar man . mockito _ kotlin . times Ċ import Ġcom . n ha ar man . mockito _ kotlin . verify Ċ import Ġcom . n ha ar man . mockito _ kotlin . wh enever Ċ import Ġorg . junit . Test Ċ import Ġorg . mockito . Mockito . do Return Ċ import Ġjava . lang . IllegalStateException Ċ import Ġjava . util . concurrent . Cyc lic Barrier Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ class ĠStream Initializer Test Ġ{ ĊĊĠĠĠ Ġprivate Ġvar Ġk inesis : ĠAmazon K inesis Ġ= Ġmock Ġ{ Ġ} ĊĠĠĠ Ġprivate Ġvar Ġsettings : ĠAws K inesis Settings Ġ= Ġmock Ġ{ Ġ} ĊĠĠĠ Ġprivate Ġvar Ġstream Initializer : ĠStream Initializer Ġ= ĠStream Initializer ( k inesis , Ġsettings ) ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġdo Ġnothing Ġif Ġstream Ġalready Ġexists `() Ġ{ ĊĠĠĠĠĠĠĠ Ġdo Return ( a Description Of An Active Stream ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( k inesis ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. describe Stream (" MY _ STREAM ") ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġcreate Ġmissing Ġstream `() Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġdo Return ( TimeUnit . SECONDS . to Millis ( 3 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. creation Timeout In Milli Seconds ĊĊĠĠĠĠĠĠĠ Ġwhenever ( k inesis . describe Stream (" MY _ STREAM ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do Throw ( Resource NotFoundException (" Stream Ġnot Ġfound ! ")) Ġ// Ġnot Ġfound Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( a Description Of A Stream In Creation ()) Ġ// Ġin Ġcreation ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( a Description Of An Active Stream ()) Ġ// Ġfinally Ġactive ĊĊĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĊĠĠĠĠĠĠĠ Ġverify ( k inesis ). create Stream (" MY _ STREAM ", Ġ 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ( expected Ġ= ĠIllegalStateException :: class ) ĊĠĠĠ Ġfun Ġ` should Ġwait Ġfor Ġstream Ġin Ġcreation Ġand Ġtimeout `() Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġdo Return ( 1 L ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. creation Timeout In Milli Seconds ĊĊĠĠĠĠĠĠĠ Ġdo Return ( a Description Of A Stream In Creation ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( k inesis ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. describe Stream (" MY _ STREAM ") ĊĊĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġcreate Ġmissing Ġstream Ġwithout Ġrace Ġcondition `() Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġdo Return ( TimeUnit . SECONDS . to Millis ( 3 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. creation Timeout In Milli Seconds ĊĊĠĠĠĠĠĠĠ Ġ// Ġthe Ġbarrier Ġwill Ġtrap Ġour Ġtwo Ġthreads Ġuntil Ġboth Ġhave Ġreached ĊĠĠĠĠĠĠĠ Ġval Ġbarrier Ġ= ĠC yc lic Barrier ( 2 ) ĊĊĠĠĠĠĠĠĠ Ġwhenever ( k inesis . describe Stream (" MY _ STREAM ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbarrier . await () Ġ// Ġtrap Ġthe Ġcaller Ġthread ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠResource NotFoundException (" Stream Ġnot Ġfound !") Ġ// Ġnot Ġfound Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbarrier . await () Ġ// Ġtrap Ġthe Ġcaller Ġthread ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠResource NotFoundException (" Stream Ġnot Ġfound !") Ġ// Ġnot Ġfound Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( a Description Of An Active Stream ()) Ġ// Ġfinally Ġactive ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( a Description Of An Active Stream ()) Ġ// Ġfinally Ġactive ĊĊĠĠĠĠĠĠĠ ĠThread Ġ{ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") Ġ}. start () ĊĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĊĠĠĠĠĠĠĠ Ġverify ( k inesis ). create Stream (" MY _ STREAM ", Ġ 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ( expected Ġ= ĠLimit Exceeded Exception :: class ) ĊĠĠĠ Ġfun Ġ` should Ġpass Ġexception Ġto Ġcaller Ġwhen Ġcreate Ġstream Ġfails `() Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġdo Return ( TimeUnit . SECONDS . to Millis ( 3 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. creation Timeout In Milli Seconds ĊĊĠĠĠĠĠĠĠ Ġdo Throw ( Resource NotFoundException (" Stream Ġnot Ġfound ! ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( k inesis ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. describe Stream (" MY _ STREAM ") ĊĊĠĠĠĠĠĠĠ Ġdo Throw ( Limit Exceeded Exception (" Limit Ġreached ! ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( k inesis ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Stream (" MY _ STREAM ", Ġ 1 ) ĊĊĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ( timeout Ġ= Ġ 6 0 L ) ĊĠĠĠ Ġfun Ġ` should Ġnot Ġdead lock Ġwhen Ġexception Ġis Ġthrown Ġduring Ġstream Ġcreation `() Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdo Return ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Streams ĊĊĠĠĠĠĠĠĠ Ġdo Return ( TimeUnit . SECONDS . to Millis ( 3 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. wh enever ( settings ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. creation Timeout In Milli Seconds ĊĊĠĠĠĠĠĠĠ Ġwhenever ( k inesis . describe Stream (" MY _ STREAM ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then Throw ( Resource NotFoundException (" Stream Ġnot Ġfound ! ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then Throw ( Resource NotFoundException (" Stream Ġnot Ġfound ! ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( a Description Of An Active Stream ()) ĊĊĠĠĠĠĠĠĠ Ġwhenever ( k inesis . create Stream (" MY _ STREAM ", Ġ 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then Throw ( Limit Exceeded Exception (" Limit Ġreached ! ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( null ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠLimit Exceeded Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġexpected ; Ġif Ġthis Ġcall Ġdead locks Ġour Ġtest Ġwill Ġtimeout Ġand Ġfail ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream Initializer . create Stream If Missing (" MY _ STREAM ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġmake Ġsure Ġour Ġsubsequent Ġcall Ġalso Ġtries Ġto Ġcreate Ġthe Ġstream , ĊĠĠĠĠĠĠĠ Ġ// Ġwhich Ġfailed Ġpreviously ĊĠĠĠĠĠĠĠ Ġverify ( k inesis , Ġtimes ( 2 )). create Stream (" MY _ STREAM ", Ġ 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġa Description Of An Active Stream (): ĠDescribe Stream Result Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠDescribe Stream Result () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Stream Description ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStream Description () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Stream Status (" ACTIVE ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġa Description Of A Stream In Creation (): ĠDescribe Stream Result Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠDescribe Stream Result () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Stream Description ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStream Description () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Stream Status (" CRE ATING ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} Ċ }