package Ġcom . e ag les ak ura . ar my knife . android . reactivex Ċ Ċ import Ġandroidx . test . ext . junit . runners . Android JUnit 4 Ċ import Ġcom . e ag les ak ura . ar my knife . android . extensions . assert UIT hread Ċ import Ġcom . e ag les ak ura . ar my knife . android . junit 4 . Test Dispatch ers Ċ import Ġcom . e ag les ak ura . ar my knife . android . junit 4 . extensions . compatible Blocking Test Ċ import Ġcom . e ag les ak ura . ar my knife . runtime . extensions . use Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . Timeout Cancellation Exception Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . with Timeout Ċ import Ġkotlinx . coroutines . yield Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . junit . Assert . assertEquals Ċ import Ġorg . junit . Assert . fail Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . runner . RunWith Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ @ RunWith ( Android JUnit 4 :: class ) Ċ class ĠRx Stream Test Ġ{ ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġchannel _ background () Ġ= Ġcompatible Blocking Test ( Test Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= ĠRx Stream . create < String >() ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġstream . to Channel () ĊĠĠĠĠĠĠĠ Ġstream . next (" send 1 ") ĊĠĠĠĠĠĠĠ ĠassertThat ( channel . receive ()). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis EqualTo (" send 1 ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġchannel . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġchannel _ ui () Ġ= Ġcompatible Blocking Test ( Test Dispatch ers . Main ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= ĠRx Stream . create < String >() ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġstream . to Channel () ĊĠĠĠĠĠĠĠ Ġstream . next (" send 1 ") ĊĠĠĠĠĠĠĠ Ġyield () ĊĠĠĠĠĠĠĠ ĠassertThat ( channel . receive ()). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis EqualTo (" send 1 ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġchannel . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġchannel _ multi () Ġ= Ġcompatible Blocking Test ( Test Dispatch ers . Main ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= ĠRx Stream . create < String >() ĊĠĠĠĠĠĠĠ Ġstream . to Channel (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" send 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" send 1 ", Ġit . receive ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" send 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" send 2 ", Ġit . receive ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" send 3 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" send 3 ", Ġit . receive ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġsubscribe () Ġ= Ġcompatible Blocking Test ( Test Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= ĠRx Stream . create < String >() ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= ĠChannel < Unit >() ĊĠĠĠĠĠĠĠ Ġstream . subscribe Ġ{ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġdis posable Ġ= Ġstream . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert UIT hread () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertThat ( it ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart sWith (" send ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . endsWith (" finish ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠGlobal Scope . launch Ġ{ Ġchannel . send ( Unit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġstream . next (" send 1 ") ĊĠĠĠĠĠĠĠ Ġstream . next (" send 2 ") ĊĠĠĠĠĠĠĠ Ġstream . next (" send 3 ") ĊĠĠĠĠĠĠĠ Ġstream . next (" send _ finish ") ĊĊĠĠĠĠĠĠĠ Ġchannel . receive () Ġ// Ġawait . ĊĠĠĠĠĠĠĠ Ġdis posable . dispose () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġbuilder _ transform () Ġ= Ġcompatible Blocking Test ( Dispatch ers . Main ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Ġ= ĠRx Stream . Builder < String >(). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġobservable Transform Ġ= Ġ{ Ġorigin Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġorigin . distinct Until Changed () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. build () ĊĊĠĠĠĠĠĠĠ Ġstream . to Channel (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" value 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" value 1 ", Ġit . receive ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" value 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" value 2 ", Ġit . receive ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠDo Ġnot Ġreceive ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" value 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Timeout ( TimeUnit . SECONDS . to Millis ( 1 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . receive () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfail (" Do Ġnot Ġreceive .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( err : ĠTimeout Cancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġok , Ġdo Ġnot Ġreceive . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġreceive . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . next (" value 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" value 1 ", Ġit . receive ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }