package Ġio . realm . internal Ċ Ċ import Ġio . realm . Callback Ċ import Ġio . realm . C ancellable Ċ import Ġio . realm . Version Id Ċ import Ġio . realm . internal . interop . Native Pointer Ċ import Ġio . realm . internal . interop . Realm Interop Ċ import Ġio . realm . internal . platform . freeze Ċ import Ġio . realm . internal . platform . run Blocking Ċ import Ġkotlinx . atomic fu . Atomic Ref Ċ import Ġkotlinx . coroutines . Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . channels . Buffer Overflow Ċ import Ġkotlinx . coroutines . channels . Channel Result Ċ import Ġkotlinx . coroutines . channels . await Close Ċ import Ġkotlinx . coroutines . ensure Active Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . Mutable Shared Flow Ċ import Ġkotlinx . coroutines . flow . as Shared Flow Ċ import Ġkotlinx . coroutines . flow . callback Flow Ċ import Ġkotlinx . coroutines . with Context Ċ Ċ /** Ċ Ġ* ĠClass Ġresponsible Ġfor Ġcontrolling Ġnotifications Ġfor Ġa ĠRealm . ĠIt Ġdoes Ġthis Ġby Ġwrapping Ġa Ġlive ĠRealm Ġon Ġwhich Ċ Ġ* Ġnotifications Ġcan Ġbe Ġregistered . ĠSince Ġall Ġobjects Ġthat Ġare Ġotherwise Ġexposed Ġto Ġusers Ġare Ġfrozen , Ġthey Ġneed Ċ Ġ* Ġto Ġbe Ġth aw ed Ġwhen Ġre aching Ġthe Ġlive ĠRealm . Ċ Ġ* Ċ Ġ* ĠFor ĠLists Ġand ĠObjects , Ġthis Ġcan Ġresult Ġin Ġthe Ġobject Ġno Ġlonger Ġexisting . ĠIn Ġthis Ġcase , ĠFlow s Ġwill Ġjust Ġcomplete . Ċ Ġ* ĠEnd Ġusers Ġcan Ġcatch Ġthis Ġcase Ġby Ġusing Ġ` flow . on Completion Ġ{ Ġ... Ġ} `. Ċ Ġ* Ċ Ġ* ĠUsers Ġare Ġonly Ġexposed Ġto Ġlive Ġobjects Ġinside Ġa Ġ[ Mutable Realm ], Ġand Ġchange Ġlisteners Ġare Ġnot Ġsupported Ċ Ġ* Ġinside Ġwrites . ĠUsers Ġcan Ġthere for Ġnot Ġregister Ġchange Ġlisteners Ġon Ġlive Ġobjects , Ġbut Ġit Ġis Ġassumed Ġthat Ġother Ċ Ġ* Ġlayers Ġcheck Ġthat Ġinvariant Ġbefore Ġmethods Ġon Ġthis Ġclass Ġare Ġcalled . Ċ Ġ*/ Ċ internal Ġclass ĠS uspend able Notifier ( ĊĠĠĠ Ġprivate Ġval Ġowner : ĠRealm Impl , ĊĠĠĠ Ġprivate Ġval Ġdispatcher : ĠCor outine Dispatcher Ċ ) Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠNO _ OP _ NOTIFICATION _ TOKEN Ġ= Ġobject Ġ: ĠC ancellable Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġcancel () Ġ{ Ġ/* ĠDo ĠNothing Ġ*/ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠFIXME ĠWork - around Ġfor Ġthe Ġglobal ĠRealm Ġchanged Ġlistener Ġnot Ġworking . ĊĠĠĠ Ġ// ĠAdding Ġextra Ġbuffer Ġcapacity Ġas Ġwe Ġare Ġotherwise Ġnever Ġable Ġto Ġemit Ġanything ĊĠĠĠ Ġ// Ġsee Ġhttps :// github . com / Kotlin / kotlin x . coroutines / blob / master / kotlin x - coroutines - core / common / src / flow / Shared Flow . kt # L 7 8 ĊĠĠĠ Ġprivate Ġval Ġ_ realm Changed Ġ= ĠMutable Shared Flow < Realm Reference >( ĊĠĠĠĠĠĠĠ Ġon Buffer Overflow Ġ= ĠBuffer Overflow . SUS PEND , ĊĠĠĠĠĠĠĠ Ġextra Buffer Capacity Ġ= Ġ 1 ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġval Ġrealm Initializer Ġ= Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdb Pointer Ġ= ĠRealm Interop . realm _ open (( owner . configuration Ġas ĠInternal Realm Configuration ). native Config , Ġdispatcher ) ĊĠĠĠĠĠĠĠ Ġobject Ġ: ĠBase Realm Impl ( owner . configuration , Ġdb Pointer ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ/* ĠReal ms Ġused Ġby Ġthe ĠNot ifier Ġis Ġjust Ġa Ġbasic ĠLive ĠRealm Ġ*/ ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġ// ĠMust Ġonly Ġbe Ġaccessed Ġfrom Ġthe Ġdispatch ers Ġthread ĊĠĠĠ Ġprivate Ġval Ġrealm : ĠBase Realm Impl Ġby Ġrealm Initializer ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFIXME ĠCurrently Ġthis Ġis Ġa Ġhack ed Ġimplementation Ġthat Ġonly Ġdoes Ġthe Ġcorrect Ġthing Ġif ĊĠĠĠĠ Ġ* Ġ Ġother ĠRealm Results Ġor ĠRealm Objects Ġare Ġbeing Ġobserved . ĠBut Ġall Ġwrites Ġshould Ġalso Ġflow ĊĠĠĠĠ Ġ* Ġ Ġfrom Ġ[ Suspend able Writer ], Ġso Ġno ĠRealm Ġupdates Ġwill Ġbe Ġlost Ġto Ġend Ġusers . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠListen Ġto Ġchanges Ġto Ġa ĠRealm . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠThis Ġflow Ġis Ġguaranteed Ġto Ġemit Ġbefore Ġany Ġother Ġstreams Ġlistening Ġto Ġindividual Ġobjects Ġor ĊĠĠĠĠ Ġ* Ġquery Ġresults . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġrealm Changed (): ĠFlow < Realm Reference > Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠWork around Ġuntil Ġproper ĠRealm ĠChanged ĠList eners Ġare Ġimplemented ĊĠĠĠĠĠĠĠ Ġ// Ġhttps :// github . com / realm / realm - core / issues / 4 6 1 3 ĊĠĠĠĠĠĠĠ Ġreturn Ġ_ realm Changed . as Shared Flow () Ċ // ĠĠĠĠĠĠĠ Ġreturn Ġcallback Flow Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtoken : ĠAtomic Ref < C ancellable > Ġ= Ġkotlinx . atomic fu . atomic ( NO _ OP _ NOTIFICATION _ TOKEN ) Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġwith Context ( dispatcher ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken . value Ġ= Ġadd Realm Changed Listener Ġ{ Ġfrozen Realm Ġ-> Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġtry Send ( frozen Realm ) Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! result . is Success ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠWhat Ġto Ġdo Ġif Ġthis Ġfails ? Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Notification Ġcould Ġnot Ġbe Ġhandled : Ġ$ result ") Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġawait Close Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken . value . cancel () Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ // ĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġ< T > Ġregister Observer ( observable : ĠObservable < T >): ĠFlow < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġcallback Flow Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtoken : ĠAtomic Ref < C ancellable > Ġ= Ġkotlinx . atomic fu . atomic ( NO _ OP _ NOTIFICATION _ TOKEN ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith Context ( dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġensure Active () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlive Ref : ĠObservable < T > Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobservable . th aw ( realm . realm Reference ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġerror (" Cannot Ġlisten Ġfor Ġchanges Ġon Ġa Ġdeleted Ġreference ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinter op Callback : Ġio . realm . internal . interop . Callback Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobject Ġ: Ġio . realm . internal . interop . Callback Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonChange ( change : ĠNative Pointer ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠHow Ġto Ġmake Ġsure Ġthe ĠRealm Ġisn 't Ġclosed Ġwhen Ġhandling Ġthis ? ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠThe ĠRealm Ġshould Ġhave Ġbeen Ġfrozen Ġin Ġ` realm Changed `, Ġbut Ġthis Ġisn 't Ġsupported Ġyet . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ ĠInstead Ġwe Ġcreate Ġthe Ġfrozen Ġversion Ġourselves Ġ( which Ġis Ġcorrect , Ġbut Ġpretty Ġin efficient ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ ĠWe Ġalso Ġsend Ġit Ġto Ġthe Ġowner ĠRealm , Ġso Ġit Ġcan Ġkeep Ġtrack Ġof Ġits Ġlifecycle ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfrozen Realm Ġ= ĠRealm Reference ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġowner , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRealm Interop . realm _ freeze ( realm . realm Reference . db Pointer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnotify Realm Changed ( frozen Realm ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNotifications Ġneed Ġto Ġbe Ġdelivered Ġwith Ġthe Ġversion Ġthey Ġwhere Ġcreated Ġon , Ġotherwise ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthe Ġfine - gr ained Ġnotification Ġdata Ġmight Ġbe Ġout Ġof Ġsync . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlive Ref . emit Frozen Update ( frozen Realm , Ġchange , Ġthis @ callback Flow ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ Ġcheck Result ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. freeze < io . realm . internal . interop . Callback >() Ġ// ĠFree ze Ġto Ġallow Ġcleaning Ġup Ġon Ġanother Ġthread ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Token Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠNotification Token < Callback < T >>( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠWhat Ġis Ġthis Ġcallback Ġfor Ġanyway ? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallback Ġ= ĠCallback Ġ{ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken Ġ= Ġlive Ref . register For Notification ( interop Callback ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken . value Ġ= Ġnew Token ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġawait Close Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtoken . value . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠListen Ġto Ġchanges Ġto Ġthe ĠRealm . ĊĠĠĠĠ Ġ* ĠThe Ġcallback Ġwill Ġhappen Ġon Ġthe Ġconfigured Ġ[ Suspend able Notifier . dispatcher ] Ġthread . ĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠFIXME ĠCall ers Ġof Ġthis Ġmethod Ġmust Ġmake Ġsure Ġit Ġis Ġcalled Ġon Ġthe Ġcorrect Ġ[ Suspend able Notifier . dispatcher ]. ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġregister Realm Changed Listener ( callback : ĠCallback < Pair < Native Pointer , ĠVersion Id >> ): ĠC ancellable Ġ{ ĊĠĠĠĠĠĠĠ ĠTODO (" Waiting Ġfor ĠRealm Interop Ġto Ġhave Ġsupport Ġfor Ġglobal ĠRealm Ġchanged ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠIs Ġit Ġsafe Ġat Ġall Ġtimes Ġto Ġclose Ġa ĠRealm ? ĠProbably Ġnot Ġduring Ġa Ġchange listener Ġcallback , Ġbut ĠMutex es ĊĠĠĠĠĠĠĠ Ġ// Ġ Ġare Ġnot Ġsupported Ġwithin Ġchange Ġlisteners Ġas Ġthey Ġare Ġnot Ġsuspend able . ĊĠĠĠĠĠĠĠ Ġrun Blocking ( dispatcher ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( realm Initializer . isInitialized ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCalling Ġclose Ġon Ġa Ġnon Ġinitialized ĠRealm Ġis Ġwast ef ul Ġsince Ġbefore Ġcalling ĠRealm Interop . close ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe ĠRealm Ġwill Ġbe Ġfirst Ġopened Ġ( Realm Interop . open ) Ġand Ġan Ġinstance Ġcreated Ġin Ġv ain . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrealm . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠVerify Ġthat Ġnotifications Ġemitted Ġto ĠStreams Ġare Ġhandled Ġin Ġan Ġuniform Ġmanner ĊĠĠĠ Ġprivate Ġfun Ġcheck Result ( result : ĠChannel Result < Unit >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( result . is Closed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġthe ĠFlow Ġwas Ġclosed , Ġwe Ġassume Ġit Ġis Ġon Ġpurpose , Ġso Ġavoid Ġraising Ġan Ġexception . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ(! result . is Success ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO ĠIs Ġthere Ġa Ġbetter Ġway Ġto Ġhandle Ġthis ? ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Notification Ġcould Ġnot Ġbe Ġsent : Ġ$ result ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġnotify Realm Changed ( frozen Realm : ĠRealm Reference ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(!_ realm Changed . try Emit ( frozen Realm )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠFIXME ĠFigure Ġout Ġwhy Ġwe Ġsometimes Ġend Ġup Ġhere ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Failed Ġto Ġsend Ġupdate Ġto ĠRealm Ġfrom Ġthe ĠNot ifier : Ġ${ owner . /** / configuration . path }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthrow ĠIllegalStateException (" Failed Ġto Ġsend Ġupdate Ġto ĠRealm Ġfrom Ġthe ĠNot ifier : Ġ${ owner . /** / configuration . path }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ