/* Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 9 ĠThe ĠAndroid ĠOpen ĠSource ĠProject Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġcom . android . tools . idea . transport . pol ler Ċ Ċ import Ġcom . android . tools . profiler . proto . Common Ċ import Ġcom . android . tools . profiler . proto . Transport Ċ import Ġcom . android . tools . profiler . proto . Transport Service Grpc Ċ import Ġcom . intellij . openapi . diagnostic . Logger Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . Copy On Write ArrayList Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . Scheduled ExecutorService Ċ import Ġjava . util . concurrent . Scheduled Future Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlin . math . max Ċ Ċ /** Ċ Ġ* ĠEn caps ulates Ġmost Ġof Ġthe Ġpolling Ġfunctionality Ġthat ĠTransport ĠPipeline Ġsubscribers Ġwould Ġneed Ġto Ġimplement Ċ Ġ* Ġto Ġlisten Ġfor Ġupdates Ġand ĠEvents Ġcoming Ġin Ġfrom Ġthe Ġpipeline Ċ Ġ*/ Ċ class ĠTransport Event Poller ( ĊĠ Ġprivate Ġval Ġtransport Client : ĠTransport Service Grpc . Transport Service Blocking Stub , ĊĠ Ġprivate Ġval Ġsort Order : ĠComparator < Common . Event > Ġ= ĠComparator . comp aring ( Common . Event :: getTimestamp ) Ċ ) Ġ{ ĊĠ Ġprivate Ġval Ġwrite Lock Ġ= ĠObject () ĊĠ Ġprivate Ġval Ġevent Listeners : ĠMutable List < Transport EventListener > Ġ= ĠCopy On Write ArrayList () Ġ// ĠUsed Ġto Ġpreserve Ġinsertion Ġorder ĊĠ Ġprivate Ġval Ġlisteners To Last Timestamp Ġ= ĠConcurrent HashMap < Transport EventListener , ĠLong >() ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠAdds Ġa Ġlistener Ġto Ġthe Ġlist Ġto Ġpoll Ġfor Ġand Ġbe Ġnotified Ġof Ġchanges . ĠList eners Ġare Ġpol led Ġin Ġinsertion Ġorder . ĊĠĠ Ġ*/ ĊĠ Ġfun Ġregister Listener ( listener : ĠTransport EventListener ) Ġ{ ĊĠĠĠ Ġsynchronized ( write Lock ) Ġ{ ĊĠĠĠĠĠ Ġevent Listeners . add ( listener ) ĊĠĠĠĠĠ Ġlisteners To Last Timestamp [ listener ] Ġ= ĠLong . MIN _ VALUE ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠRemoves Ġa Ġlistener Ġfrom Ġthe Ġlist , Ġor Ġdo Ġnothing Ġif Ġit Ġis Ġnot Ġin Ġthe Ġlist ĊĠĠ Ġ*/ ĊĠ Ġfun Ġunregister Listener ( listener : ĠTransport EventListener ) Ġ{ ĊĠĠĠ Ġsynchronized ( write Lock ) Ġ{ ĊĠĠĠĠĠ Ġevent Listeners . remove ( listener ) ĊĠĠĠĠĠ Ġlisteners To Last Timestamp . remove ( listener ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġpoll () Ġ{ ĊĠĠĠ Ġ// ĠCopy Ġthe Ġlist Ġso Ġwe Ġcan Ġremove Ġlisteners Ġwithin Ġthe Ġloop Ġin - place . ĊĠĠĠ Ġval Ġlisteners Ġ= Ġmutable ListOf < Transport EventListener >(). apply Ġ{ Ġadd All ( event Listeners ) Ġ} ĊĠĠĠ Ġ// ĠPoll Ġfor Ġeach Ġlistener ĊĠĠĠ Ġfor Ġ( event Listener Ġin Ġlisteners ) Ġ{ ĊĠĠĠĠĠ Ġ// ĠUse Ġstart / end Ġtime Ġif Ġavailable ĊĠĠĠĠĠ Ġval Ġstart Timestamp Ġ= Ġlisteners To Last Timestamp [ event Listener ] Ġ?: Ġevent Listener . startTime ?. invoke () Ġ?: ĠLong . MIN _ VALUE ĊĠĠĠĠĠ Ġval Ġend Timestamp Ġ= Ġevent Listener . endTime () ĊĊĠĠĠĠĠ Ġval Ġbuilder Ġ= ĠTransport . Get Event Groups Request . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. set Kind ( event Listener . event Kind ) ĊĠĠĠĠĠĠĠ Ġ. set From Timestamp ( start Timestamp ) ĊĠĠĠĠĠĠĠ Ġ. set To Timestamp ( end Timestamp ) ĊĠĠĠĠĠ Ġevent Listener . stream Id ?. invoke ()?. let Ġ{ Ġbuilder . stream Id Ġ= Ġit Ġ} ĊĠĠĠĠĠ Ġevent Listener . process Id ?. invoke ()?. let Ġ{ Ġbuilder . pid Ġ= Ġit Ġ} ĊĠĠĠĠĠ Ġevent Listener . groupId ?. invoke ()?. let Ġ{ Ġbuilder . groupId Ġ= Ġit Ġ} ĊĊĠĠĠĠĠ Ġval Ġrequest Ġ= Ġbuilder . build () ĊĊĠĠĠĠĠ Ġ// ĠOrder Ġby Ġtimestamp ĊĠĠĠĠĠ Ġval Ġresponse Ġ= Ġtransport Client . getEvent Groups ( request ) ĊĠĠĠĠĠ Ġif Ġ( response Ġ!= ĠTransport . Get Event Groups Response . getDefaultInstance ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfiltered Ġ= Ġresponse . groups List ĊĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġgroup Ġ-> Ġgroup . events List Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ. sorted With ( sort Order ) ĊĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġevent Ġ-> Ġevent . timestamp Ġ>= Ġstart Timestamp Ġ&& Ġevent Listener . filter ( event ) Ġ} ĊĠĠĠĠĠĠĠ Ġfiltered . forEach Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġevent Listener . executor . execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event Listener . callback ( event )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠPrevious Ġcode Ġcollected Ġthe Ġflag Ġand Ġun registered Ġonce Ġin Ġthe Ġmain Ġthread , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbut Ġthere Ġwas Ġa Ġconcurrency Ġbug Ġif Ġthe Ġmain Ġthread Ġfinishes Ġbefore Ġthe Ġlisteners . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġunregister Ġfrom Ġhere Ġinstead . ĠUn register ing Ġthe Ġsame Ġlistener Ġmultiple Ġtimes Ġis Ġharm less . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġunregister Listener ( event Listener ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmax Time Event Ġ= Ġfiltered . max By OrNull Ġ{ Ġit . timestamp Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠUpdate Ġlast Ġtimestamp Ġper Ġlistener ĊĠĠĠĠĠĠĠ Ġsynchronized ( write Lock ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ// ĠMake Ġsure Ġthe Ġlistener Ġis Ġstill Ġregistered Ġbefore Ġadding Ġa Ġnew Ġtimestamp ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( max Time Event Ġ!= Ġnull Ġ&& Ġlisteners To Last Timestamp . containsKey ( event Listener )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlisteners To Last Timestamp [ event Listener ] Ġ= Ġmax ( start Timestamp , Ġmax Time Event . timestamp Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval Ġmy ExecutorService : ĠScheduled ExecutorService Ġ= ĠExec utors . new Scheduled ThreadPool ( 1 ) ĊĠĠĠ Ġprivate Ġval Ġmy Scheduled F utures Ġ= Ġmutable Map Of < Transport Event Poller , ĠScheduled Future <* >>() ĊĊĠĠĠ Ġ@ Jvm Over loads ĊĠĠĠ Ġ@ Jvm Static ĊĠĠĠ Ġfun Ġcreate Poller ( transport Client : ĠTransport Service Grpc . Transport Service Blocking Stub , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpoll Period Ns : ĠLong , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsort Order : Ġjava . util . Comparator < Common . Event > Ġ= ĠComparator . comp aring ( Common . Event :: getTimestamp ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠexecutorService For Test : ĠScheduled ExecutorService ? Ġ= Ġnull ĊĠĠĠ Ġ): ĠTransport Event Poller Ġ{ ĊĠĠĠĠĠ Ġval Ġpol ler Ġ= ĠTransport Event Poller ( transport Client , Ġsort Order ) ĊĠĠĠĠĠ Ġval Ġscheduled Future Ġ= Ġ( executor Service For Test Ġ?: Ġmy ExecutorService ). schedule With Fixed Delay ( ĊĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpol ler . poll () ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠLogger . getInstance ( Transport Event Poller :: class . java ). warn ( t . toString ()) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠ 0 , Ġpoll Period Ns , ĠTimeUnit . NAN O SECONDS ) ĊĠĠĠĠĠ Ġmy Scheduled F utures [ pol ler ] Ġ= Ġscheduled Future ĊĠĠĠĠĠ Ġreturn Ġpol ler ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Jvm Static ĊĠĠĠ Ġfun Ġstop Poller ( pol ler : ĠTransport Event Poller ) Ġ{ ĊĠĠĠĠĠ Ġmy Scheduled F utures . remove ( pol ler )?. cancel ( false ) ĊĠĠĠ Ġ} ĊĊĠ Ġ} Ċ } Ċ