package Ġcom . radix d lt . client . core . ledger Ċ Ċ import Ġcom . radix d lt . client . core . address . Rad ix Address Ċ import Ġcom . radix d lt . client . core . atoms . Abstract Consum able Ċ import Ġcom . radix d lt . client . core . atoms . Atom Ċ import Ġcom . radix d lt . client . core . atoms . Transaction Atom Ċ import Ġio . reactivex . Observable Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ Ċ class ĠCon sum able DataSource ( private Ġval Ġatom Store : ĠAtom Store ) Ġ: ĠParticle Store Ġ{ ĊĠĠĠ Ġprivate Ġval Ġcache Ġ= ĠConcurrent HashMap < Rad ix Address , ĠObservable < Abstract Consum able >>() ĊĊĠĠĠ Ġoverride Ġfun Ġget Consum ables ( address : ĠRad ix Address ): ĠObservable < Abstract Consum able > Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠTODO : Ġuse Ġhttps :// github . com / J ake Wh art on / Rx Replay ing Share Ġto Ġdisconnect Ġwhen Ġun subscribed ĊĠĠĠĠĠĠĠ Ġreturn Ġcache . compute If Absent Synchron ised Function ( address ) Ġ{ Ġaddr Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġatom Store . get Atoms ( address ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter ( Atom :: is Transaction Atom ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map ( Atom :: as Transaction Atom ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Iterable ( Transaction Atom :: get Abstract Consum ables ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġparticle Ġ-> Ġparticle . owners Public Keys . as Sequence (). all ( address :: own s Key ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. cache () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ. replay ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ. auto Connect () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠImplementation Ġthat Ġdoesn 't Ġblock Ġwhen Ġmap Ġalready Ċ Ġ* Ġcontains Ġthe Ġvalue Ċ Ġ*/ Ċ fun Ġ< K , ĠV > ĠConcurrent HashMap < K , ĠV >. compute If Absent Function ( key : ĠK , Ġmapping Function : Ġ( t : ĠK ) Ġ-> ĠV ): ĠV Ġ{ ĊĠĠĠ Ġreturn Ġthis [ key ] Ġ?: Ġrun Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= Ġmapping Function ( key ) ĊĠĠĠĠĠĠĠ Ġthis [ key ] Ġ= Ġvalue ĊĠĠĠĠĠĠĠ Ġreturn @ run Ġvalue ĊĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠSynchron ised Ġthat Ġdoesn 't Ġblock Ġwhen Ġmap Ġalready Ċ Ġ* Ġcontains Ġthe Ġvalue Ċ Ġ*/ Ċ fun Ġ< K , ĠV > ĠConcurrent HashMap < K , ĠV >. compute If Absent Synchron ised Function ( key : ĠK , Ġmapping Function : Ġ( t : ĠK ) Ġ-> ĠV ): ĠV Ġ{ ĊĠĠĠ Ġreturn Ġthis [ key ] Ġ?: Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġvalue Synchron ized Ġ= Ġget ( key ) ĊĠĠĠĠĠĠĠ Ġif Ġ( value Synchron ized Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvalue Synchron ized Ġ= Ġmapping Function ( key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis [ key ] Ġ= Ġvalue Synchron ized ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn @ synchronized Ġvalue Synchron ized !! ĊĠĠĠ Ġ} Ċ } Ċ