package Ġcom . dv per my ak ov . rx . subjects Ċ Ċ import Ġcom . dv per my ak ov . rx . ob servers . Observer Ċ import Ġcom . dv per my ak ov . rx . ob servers . Observer . State Ċ Ċ class ĠAsync Subject < T > Ġ: ĠSubject < T >() Ġ{ ĊĠĠĠ Ġprivate Ġval Ġob servers : ĠMutable List < Observer < T >> Ġ= Ġmutable ListOf () ĊĠĠĠ Ġprivate Ġvar Ġlast Item : ĠT ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġstate : ĠState Ġ= ĠState . Idle ĊĊĠĠĠ Ġoverride Ġfun Ġsubscribe Actual ( observer : ĠObserver < T >) Ġ{ ĊĠĠĠĠĠĠĠ Ġsuper . subscribe Actual ( observer ) ĊĠĠĠĠĠĠĠ Ġwhen Ġ( state ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState . Idle , ĠState . Sub scribed Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġob servers . add ( observer ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠState . Error Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobserver . onError (( state Ġas ĠState . Error ). t ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState . Completed Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Item ?. let ( observer :: onNext ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobserver . onComplete () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Subscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠState . Sub scribed ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Next ( item : ĠT ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlast Item Ġ= Ġitem ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonComplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠState . Completed ĊĠĠĠĠĠĠĠ Ġob servers . forEach Ġ{ Ġobserver Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Item ?. let ( observer :: onNext ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġobserver . onComplete () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠonError ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠ Ġstate Ġ= ĠState . Error ( t ) ĊĠĠĠĠĠĠĠ Ġob servers . forEach Ġ{ Ġobserver Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġobserver . onError ( t ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }