/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 - 2 0 1 8 ĠJetBrains Ġs . r . o . ĠUse Ġof Ġthis Ġsource Ġcode Ġis Ġgoverned Ġby Ġthe ĠApache Ġ 2 . 0 Ġlicense . Ċ Ġ*/ Ċ Ċ package Ġkotlinx . coroutines . reactor Ċ Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . flow .* Ċ import Ġkotlinx . coroutines . reactive .* Ċ import Ġorg . junit .* Ċ import Ġorg . junit . Test Ċ import Ġorg . reactiv est ream s .* Ċ import Ġreactor . core . publisher .* Ċ import Ġreactor . util . context .* Ċ import Ġjava . time . Duration .* Ċ import Ġjava . util . function .* Ċ import Ġkotlin . test .* Ċ Ċ class ĠMono Test Ġ: ĠTest Base () Ġ{ ĊĠĠĠ Ġ@ Before ĊĠĠĠ Ġfun Ġsetup () Ġ{ ĊĠĠĠĠĠĠĠ Ġignore Lost Threads (" timer -", Ġ" parallel - ") ĊĠĠĠĠĠĠĠ ĠHooks . onError Dropped Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Basic Success () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" OK " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠ Ġmono . subscribe Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġvalue ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġyield () Ġ// Ġto Ġstarted Ġcoroutine ĊĠĠĠĠĠĠĠ Ġfinish ( 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Basic Failure () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRuntimeException (" OK ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠ Ġmono . subscribe ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ Ġerror Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue ( error Ġis ĠRuntimeException ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġerror . message ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġyield () Ġ// Ġto Ġstarted Ġcoroutine ĊĠĠĠĠĠĠĠ Ġfinish ( 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Basic Empty () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠ Ġmono . subscribe ({}, Ġ{ Ġthrow Ġit Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġyield () Ġ// Ġto Ġstarted Ġcoroutine ĊĠĠĠĠĠĠĠ Ġfinish ( 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Basic Un subscribe () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġyield () Ġ// Ġback Ġto Ġmain , Ġwill Ġget Ġcancelled ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠ Ġ// Ġnothing Ġis Ġcalled Ġon Ġa Ġdisposed Ġmono ĊĠĠĠĠĠĠĠ Ġval Ġsub Ġ= Ġmono . subscribe ({ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġyield () Ġ// Ġto Ġstarted Ġcoroutine ĊĠĠĠĠĠĠĠ Ġexpect ( 5 ) ĊĠĠĠĠĠĠĠ Ġsub . dispose () Ġ// Ġwill Ġcancel Ġcoroutine ĊĠĠĠĠĠĠĠ Ġyield () ĊĠĠĠĠĠĠĠ Ġfinish ( 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mono No Wait () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" OK " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mono Await () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", ĠMono . just (" O "). await Single () Ġ+ Ġ" K ") ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", ĠMono . just (" O "). await Single OrNull () Ġ+ Ġ" K ") ĊĠĠĠĠĠĠĠ Ġassert Fails With < NoSuch ElementException >{ ĠMono . empty < String >(). await Single () Ġ} ĊĠĠĠĠĠĠĠ ĠassertNull ( Mono . empty < Int >(). await Single OrNull ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠTests Ġthat Ġthe Ġversions Ġof Ġthe Ġawait Ġmethods Ġspecialized Ġfor ĠMono Ġfor Ġdeprecation Ġbehave Ġcorrectly Ġand Ġwe Ġdon 't ĊĠĠĠĠ Ġ* Ġbreak Ġany Ġcode Ġby Ġintro ducing Ġthem . Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġ@ Suppress (" DE PREC ATION ") ĊĠĠĠ Ġfun Ġtest Deprecated Await Methods () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfilled Mono Ġ= Ġmono < String > Ġ{ Ġ" OK " Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await First ()) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await FirstOrDefault (" ! ")) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await First OrNull ()) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await First OrElse Ġ{ Ġ" ELSE " Ġ}) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await Last ()) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await Single OrDefault (" ! ")) ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġfilled Mono . await Single OrElse Ġ{ Ġ" ELSE " Ġ}) ĊĠĠĠĠĠĠĠ Ġval Ġempty Mono Ġ= Ġmono < String > Ġ{ Ġnull Ġ} ĊĠĠĠĠĠĠĠ Ġassert Fails With < NoSuch ElementException > Ġ{ Ġempty Mono . await First () Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġempty Mono . await FirstOrDefault (" OK ")) ĊĠĠĠĠĠĠĠ ĠassertNull ( empty Mono . await First OrNull ()) ĊĠĠĠĠĠĠĠ ĠassertEquals (" ELSE ", Ġempty Mono . await First OrElse Ġ{ Ġ" ELSE " Ġ}) ĊĠĠĠĠĠĠĠ Ġassert Fails With < NoSuch ElementException > Ġ{ Ġempty Mono . await Last () Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġempty Mono . await Single OrDefault (" OK ")) ĊĠĠĠĠĠĠĠ ĠassertEquals (" ELSE ", Ġempty Mono . await Single OrElse Ġ{ Ġ" ELSE " Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠTests Ġthat Ġcalls Ġto Ġ[ await Single OrNull ] Ġ( and , Ġthus , Ġto Ġthe Ġrest Ġof Ġsuch Ġfunctions ) Ġthrow Ġ[ Cancellation Exception ] ĊĠĠĠĠ Ġ* Ġand Ġunsubscribe Ġfrom Ġthe Ġpublisher Ġwhen Ġtheir Ġ[ Job ] Ġis Ġcancelled . Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Await Cancellation () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ Ġdelay ( Long . MAX _ VALUE ) Ġ}. do On Subscribe Ġ{ Ġexpect ( 3 ) Ġ}. do On Cancel Ġ{ Ġexpect ( 5 ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġlaunch ( start Ġ= ĠCor outine Start . UN DISPATCH ED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmono . await Single OrNull () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 6 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠ Ġjob . cancel And Join () ĊĠĠĠĠĠĠĠ Ġfinish ( 7 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mono Emit And Await () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMono . just (" O "). await Single () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mono With Delay () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just (" O "). delay Elements ( of Millis ( 5 0 )). await Single () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Mono Exception () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just (" O ", Ġ" K "). await Single () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Err one ous ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert ( it Ġis ĠIllegalArgumentException ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Await First () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just (" O ", Ġ"# "). await First () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Await Last () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just ("# ", Ġ" O "). await Last () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Exception From Flux () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . error < String >( RuntimeException (" O ")). await First () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠRuntimeException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just ( e . message !! ). await Last () Ġ+ Ġ" K " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Mono Value ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Exception From Coroutine () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono < String > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException ( Flux . just (" O "). await Single () Ġ+ Ġ" K ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcheck Err one ous ( mono ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert ( it Ġis ĠIllegalStateException ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" OK ", Ġit . message ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Sup pressed Exception () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( start Ġ= ĠCor outine Start . ATOMIC ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠTest Exception () Ġ// Ġchild Ġcoroutine Ġfails ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠTest Exception 2 () Ġ// Ġbut Ġparent Ġthrows Ġanother Ġexception Ġwhile Ġcleaning Ġup ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmono . await Single () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠTest Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue ( e . sup pressed [ 0 ] Ġis ĠTest Exception 2 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Unhandled Exception () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠ Ġvar Ġsubscription : ĠSubscription ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġhandler Ġ= ĠBi Function < Throwable , ĠAny ?, ĠThrowable > Ġ{ Ġt , Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertTrue ( t Ġis ĠTest Exception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġt ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( current Dispatcher ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription !!. cancel () Ġ// Ġcancel Ġour Ġown Ġsubscription , Ġso Ġthat Ġdelay Ġwill Ġget Ġcancelled ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠTest Exception () Ġ// Ġwould Ġnot Ġbe Ġable Ġto Ġhandle Ġit Ġsince Ġmono Ġis Ġdisposed ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. context Write Ġ{ ĠContext . of (" reactor . on Operator Error . local ", Ġhandler ) Ġ} ĊĠĠĠĠĠĠĠ Ġmono . subscribe ( object Ġ: ĠSubscriber < Unit > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Subscribe ( s : ĠSubscription ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription Ġ= Ġs ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Next ( t : ĠUnit ?) Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonError ( t : ĠThrowable ) Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġyield () Ġ// Ġrun Ġcoroutine ĊĠĠĠĠĠĠĠ Ġfinish ( 6 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest IllegalArgumentException () Ġ{ ĊĠĠĠĠĠĠĠ Ġassert Fails With < IllegalArgumentException > Ġ{ Ġmono ( Job ()) Ġ{ Ġ} Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Exception After Cancellation () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTest Ġexception Ġis Ġnot Ġreported Ġto Ġglobal Ġhandler ĊĠĠĠĠĠĠĠ ĠFlux ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. interval ( of Millis ( 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. switch Map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmono ( coroutine Context ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtime B omb (). await Single () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Return ({ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, Ġ 4 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. block Last () ĊĠĠĠĠĠĠĠ Ġfinish ( 2 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġtime B omb () Ġ= ĠMono . delay ( of Millis ( 1 )). do On Success Ġ{ Ġthrow ĠException (" something Ġwent Ġwrong ") Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Le aked Exception () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTest Ġexception Ġis Ġnot Ġreported Ġto Ġglobal Ġhandler ĊĠĠĠĠĠĠĠ Ġval Ġflow Ġ= Ġmono < Unit > Ġ{ Ġthrow ĠTest Exception () Ġ}. to Flux (). as Flow () ĊĠĠĠĠĠĠĠ Ġrepeat ( 1 0 0 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcombine ( flow , Ġflow ) Ġ{ Ġ_, Ġ_ Ġ-> Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. catch Ġ{} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. collect Ġ{ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠTest Ġthat Ġcancell ing Ġa Ġ[ mono ] Ġdue Ġto Ġa Ġtimeout Ġdoes Ġthrow Ġan Ġexception . Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Timeout () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith Timeout ( 1 ) Ġ{ Ġdelay ( 1 0 0 ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmono . do On Subscribe Ġ{ Ġexpect ( 1 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Success Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ Ġexpect ( 2 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Cancel Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. block () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 3 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġfinish ( 4 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠTest Ġthat Ġwhen Ġthe Ġreason Ġfor Ġcancellation Ġof Ġa Ġ[ mono ] Ġis Ġthat Ġthe Ġdownstream Ġdoesn 't Ġwant Ġits Ġresults Ġanymore , ĊĠĠĠĠ Ġ* Ġthis Ġis Ġconsidered Ġnormal Ġbehavior Ġand Ġexceptions Ġare Ġnot Ġpropag ated . Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Down stream Cancellation DoesNot Throw () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġi Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġ/** ĠAttach Ġa Ġhook Ġthat Ġhandles Ġexceptions Ġfrom Ġpublish ers Ġthat Ġare Ġknown Ġto Ġbe Ġdisposed Ġof . ĠWe Ġdon 't Ġexpect Ġit ĊĠĠĠĠĠĠĠĠ Ġ* Ġto Ġbe Ġfired Ġin Ġthis Ġcase , Ġas Ġthe Ġreason Ġfor Ġthe Ġpublisher Ġin Ġthis Ġtest Ġto Ġaccept Ġan Ġexception Ġis Ġsimply ĊĠĠĠĠĠĠĠĠ Ġ* Ġcancellation Ġfrom Ġthe Ġdownstream . Ġ*/ ĊĠĠĠĠĠĠĠ ĠHooks . on Operator Error (" test Down stream Cancellation DoesNot Throw ") Ġ{ Ġt , Ġa Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect Un re ached () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġt ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ/** ĠA ĠMono Ġthat Ġdoesn 't Ġemit Ġa Ġvalue Ġand Ġinstead Ġwaits Ġin def initely . Ġ*/ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( Dispatch ers . Un conf ined ) Ġ{ Ġexpect ( 5 Ġ* Ġi Ġ+ Ġ 3 ); Ġdelay ( Long . MAX _ VALUE ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Subscribe Ġ{ Ġexpect ( 5 Ġ* Ġi Ġ+ Ġ 2 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Success Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Cancel Ġ{ Ġexpect ( 5 Ġ* Ġi Ġ+ Ġ 4 ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġ 1 0 0 0 ĊĠĠĠĠĠĠĠ Ġrepeat ( n ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġi Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 Ġ* Ġi Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmono . await Cancel And Join () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( 5 Ġ* Ġi Ġ+ Ġ 5 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġfinish ( 5 Ġ* Ġn Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠ ĠHooks . reset On Operator Error (" test Down stream Cancellation DoesNot Throw ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠTest Ġthat , Ġwhen Ġ[ Mono ] Ġis Ġcancelled Ġby Ġthe Ġdownstream Ġand Ġthrows Ġduring Ġhandling Ġthe Ġcancellation , Ġthe Ġresulting ĊĠĠĠĠ Ġ* Ġerror Ġis Ġpropag ated Ġto Ġ[ Hooks . on Operator Error ]. Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Re throw ing Down stream Cancellation () Ġ= Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġi Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġ/** ĠAttach Ġa Ġhook Ġthat Ġhandles Ġexceptions Ġfrom Ġpublish ers Ġthat Ġare Ġknown Ġto Ġbe Ġdisposed Ġof . ĠWe Ġexpect Ġit ĊĠĠĠĠĠĠĠĠ Ġ* Ġto Ġbe Ġfired Ġin Ġthis Ġcase . Ġ*/ ĊĠĠĠĠĠĠĠ ĠHooks . on Operator Error (" test Down stream Cancellation DoesNot Throw ") Ġ{ Ġt , Ġa Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 5 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġt ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ/** ĠA ĠMono Ġthat Ġdoesn 't Ġemit Ġa Ġvalue Ġand Ġinstead Ġwaits Ġin def initely , Ġand , Ġwhen Ġcancelled , Ġthrows . Ġ*/ ĊĠĠĠĠĠĠĠ Ġval Ġmono Ġ= Ġmono ( Dispatch ers . Un conf ined ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 3 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠTest Exception () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Subscribe Ġ{ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 2 ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Success Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do OnError Ġ{ Ġexpect Un re ached () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Cancel Ġ{ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 4 ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġn Ġ= Ġ 1 0 0 0 ĊĠĠĠĠĠĠĠ Ġrepeat ( n ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġi Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmono . await Cancel And Join () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpect ( i Ġ* Ġ 6 Ġ+ Ġ 6 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġfinish ( n Ġ* Ġ 6 Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠ ĠHooks . reset On Operator Error (" test Down stream Cancellation DoesNot Throw ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠRun Ġthe Ġgiven Ġ[ Mono ], Ġcancel Ġit , Ġwait Ġfor Ġthe Ġcancellation Ġhandler Ġto Ġfinish , Ġand Ġreturn Ġonly Ġthen . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠWill Ġnot Ġwork Ġin Ġthe Ġgeneral Ġcase , Ġbut Ġhere , Ġwhen Ġthe Ġpublisher Ġuses Ġ[ Dispatch ers . Un conf ined ], Ġthis Ġseems Ġto ĊĠĠĠĠ Ġ* Ġensure Ġthat Ġthe Ġcancellation Ġhandler Ġwill Ġhave Ġnow here Ġto Ġexecute Ġbut Ġser ially Ġwith Ġthe Ġcancellation . Ġ*/ ĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġ< T > ĠMono < T >. await Cancel And Join () Ġ= Ġcoroutine Scope Ġ{ ĊĠĠĠĠĠĠĠ Ġasync ( start Ġ= ĠCor outine Start . UN DISPATCH ED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġawait Single OrNull () ĊĠĠĠĠĠĠĠ Ġ}. cancel And Join () ĊĠĠĠ Ġ} Ċ } Ċ