/* Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 2 0 Ġ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 . manager Ċ Ċ import Ġcom . android . annotations . concurrency . Any Thread Ċ import Ġcom . android . tools . profiler . proto . Common Ċ import Ġcom . android . tools . profiler . proto . Transport . Get Event Groups Request Ċ import Ġcom . android . tools . profiler . proto . Transport . Get Event Groups Response Ċ import Ġcom . android . tools . profiler . proto . Transport Service Grpc Ċ import Ġkotlinx . coroutines . Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . flow Ċ import Ġkotlinx . coroutines . flow . flow On Ċ import Ġkotlinx . coroutines . flow . take While Ċ import Ġkotlinx . coroutines . flow . transform Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ Ċ const Ġval ĠDEL AY _ M ILL IS Ġ= Ġ 1 0 0 L Ċ Ċ /** Ċ Ġ* ĠRepresents Ġstream Ġconnection Ġactivity . Ċ Ġ*/ Ċ sealed Ġclass ĠStream Activity ( val Ġstream Channel : ĠTransport Stream Channel ) Ċ Ċ /** Ċ Ġ* ĠRepresents Ġa Ġstream Ġconnect Ġactivity . Ċ Ġ*/ Ċ class ĠStream Connected ( stream Channel : ĠTransport Stream Channel ) Ġ: ĠStream Activity ( stream Channel ) Ċ Ċ /** Ċ Ġ* ĠRepresents Ġa Ġstream Ġdisconnect Ġactivity . Ċ Ġ*/ Ċ class ĠStream Disconnected ( stream Channel : ĠTransport Stream Channel ) Ġ: ĠStream Activity ( stream Channel ) Ċ Ċ /** Ċ Ġ* ĠThis Ġsimply Ġwraps Ġ[ Common . Event ] Ġand Ġrepresents Ġan Ġevent Ġfrom Ġa Ġstream Ġchannel . Ċ Ġ*/ Ċ class ĠStream Event ( val Ġevent : ĠCommon . Event ) Ċ Ċ private Ġfun ĠTransport Service Grpc . Transport Service Blocking Stub . event Group Flow ( ĊĠ Ġpoll Period Ms : ĠLong Ġ= ĠDEL AY _ M ILL IS , ĊĠ Ġrequest Builder : Ġ() Ġ-> ĠGet Event Groups Request Ċ ): ĠFlow < Get Event Groups Response > Ġ= Ġflow Ġ{ ĊĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠ Ġemit ( getEvent Groups ( request Builder ())) ĊĠĠĠ Ġdelay ( poll Period Ms ) ĊĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠList ens Ġto Ġand Ġmanages ĠST REAM Ġevents Ġfrom Ġthe Ġtransport Ġpipeline . ĠUsers Ġcan Ġsubscribe Ġto Ġthe Ġactivity Ġflow Ġto Ġlisten Ġin Ġon Ġnew Ġstreams . Ċ Ġ* Ċ Ġ* ĠStreams Ġare Ġprovided Ġin Ġthe Ġform Ġof Ġ[ Transport Stream Channel ], Ġwhich Ġprovides Ġusers Ġthe Ġability Ġto Ġregister Ġtheir Ġown Ġlisteners Ġwhile Ċ Ġ* Ġautomatically Ġassoci ating Ġthem Ġwith Ġthe Ġstream . Ċ Ġ*/ Ċ @ Any Thread Ċ class ĠTransport Stream Manager Ġprivate Ġconstructor ( ĊĠ Ġprivate Ġval Ġclient : ĠTransport Service Grpc . Transport Service Blocking Stub , ĊĠ Ġprivate Ġval Ġdispatcher : ĠCor outine Dispatcher Ċ ) Ġ{ ĊĠ Ġfun Ġstream Activity Flow (): ĠFlow < Stream Activity > Ġ{ ĊĠĠĠ Ġval Ġstreams Ġ= Ġmutable Map Of < Long , ĠTransport Stream Channel >() ĊĠĠĠ Ġ// ĠGet Ġall Ġstreams Ġof Ġall Ġtypes . ĊĠĠĠ Ġval Ġrequest Ġ= ĠGet Event Groups Request . newBuilder () ĊĠĠĠĠĠ Ġ. set Stream Id (- 1 ) Ġ// ĠData Store Service . DATA STORE _ RESERVED _ STREAM _ ID ĊĠĠĠĠĠ Ġ. set Kind ( Common . Event . Kind . STREAM ) ĊĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġreturn Ġclient ĊĠĠĠĠĠ Ġ. event Group Flow Ġ{ Ġrequest Ġ} ĊĠĠĠĠĠ Ġ. transform Ġ{ Ġresponse Ġ-> ĊĠĠĠĠĠĠĠ Ġfor Ġ( group Ġin Ġresponse . groups List ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( group . events Count Ġ<= Ġ 0 ) Ġcontinue ĊĠĠĠĠĠĠĠĠĠ Ġval Ġstream Id Ġ= Ġgroup . groupId ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġsort Ġlist Ġby Ġtimestamp Ġin Ġdescending Ġorder ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġthe Ġlatest Ġevent Ġmay Ġsignal Ġthe Ġstream Ġis Ġalive Ġor Ġdead : ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġ 1 ) Ġalive Ġ- Ġif Ġstream Ġis Ġnew , Ġadd Ġstream Ġand Ġnotify Ġlisteners . Ġotherwise Ġdo Ġnothing ĊĠĠĠĠĠĠĠĠĠ Ġ// Ġ 2 ) Ġdead Ġ- Ġif Ġstream Ġis Ġdead Ġand Ġmanager Ġknows Ġit Ġfrom Ġbefore , Ġthen Ġremove Ġit Ġand Ġnotify Ġlisteners . ĠOtherwise Ġdo Ġnothing . ĊĠĠĠĠĠĠĠĠĠ Ġval Ġlatest Event Ġ= Ġgroup . events List . max By OrNull Ġ{ Ġit . timestamp Ġ} Ġ?: Ġcontinue ĊĠĠĠĠĠĠĠĠĠ Ġval Ġis Connected Ġ= Ġlatest Event . stream . has Stream Connected () ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( isConnected ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! streams . containsKey ( stream Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Channel Ġ= ĠTransport Stream Channel ( latest Event . stream . stream Connected . stream , Ġclient , Ġdispatcher ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( Stream Connected ( stream Channel )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstreams [ stream Id ] Ġ= Ġstream Channel ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstreams . remove ( stream Id )?. let Ġ{ Ġchannel Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . clean Up () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( Stream Disconnected ( channel )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ. flow On ( dispatcher ) ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval Ġmanagers Ġ= Ġmutable ListOf < Transport Stream Manager >() ĊĊĠĠĠ Ġfun Ġcreate Manager ( ĊĠĠĠĠĠ Ġclient : ĠTransport Service Grpc . Transport Service Blocking Stub , ĊĠĠĠĠĠ Ġdispatcher : ĠCor outine Dispatcher ĊĠĠĠ Ġ): ĠTransport Stream Manager Ġ{ ĊĠĠĠĠĠ Ġval Ġmanager Ġ= ĠTransport Stream Manager ( client , Ġdispatcher ) ĊĠĠĠĠĠ Ġmanagers . add ( manager ) ĊĠĠĠĠĠ Ġreturn Ġmanager ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġunregister Manager ( manager : ĠTransport Stream Manager ) Ġ{ ĊĠĠĠĠĠ Ġmanagers . remove ( manager ) ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠRepresents Ġthe Ġtransport Ġchannel Ġof Ġa Ġstream , Ġwhich Ġis Ġa Ġdevice Ġor Ġemulator . Ċ Ġ* Ċ Ġ* ĠUsers Ġcan Ġsubscribe Ġto Ġa Ġflow Ġof Ġevents Ġin Ġthis Ġstream Ġby Ġproviding Ġtheir Ġown Ġ[ Stream Event Query ]. Ċ Ġ*/ Ċ @ Any Thread Ċ class ĠTransport Stream Channel ( ĊĠ Ġval Ġstream : ĠCommon . Stream , ĊĠ Ġval Ġclient : ĠTransport Service Grpc . Transport Service Blocking Stub , ĊĠ Ġprivate Ġval Ġdispatcher : ĠCor outine Dispatcher Ċ ) Ġ{ ĊĊĠ Ġprivate Ġval Ġis Closed Ġ= ĠAtomic Boolean ( false ) ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠCreates Ġand Ġreturns Ġa Ġflow Ġbased Ġon Ġthe Ġfiltering Ġcriteria Ġindicated Ġby Ġ[ query ]. ĊĠĠ Ġ*/ ĊĠ Ġfun Ġevent Flow ( query : ĠStream Event Query ): ĠFlow < Stream Event > Ġ{ ĊĠĠĠ Ġvar Ġlast TimeStamp Ġ= Ġquery . startTime ?. invoke () Ġ?: ĠLong . MIN _ VALUE ĊĠĠĠ Ġreturn Ġclient ĊĠĠĠĠĠ Ġ. event Group Flow Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuilder Ġ= ĠGet Event Groups Request . newBuilder () ĊĠĠĠĠĠĠĠĠĠ Ġ. set Stream Id ( stream . stream Id ) ĊĠĠĠĠĠĠĠĠĠ Ġ. set Kind ( query . event Kind ) ĊĠĠĠĠĠĠĠĠĠ Ġ. set From Timestamp ( last TimeStamp ) ĊĠĠĠĠĠĠĠĠĠ Ġ. set To Timestamp ( query . endTime ()) ĊĠĠĠĠĠĠĠ Ġquery . process Id ?. invoke ()?. let Ġ{ Ġbuilder . pid Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġquery . groupId ?. invoke ()?. let Ġ{ Ġbuilder . groupId Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠ Ġbuilder . build () ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ. take While Ġ{ Ġ! is Closed . get () Ġ} ĊĠĠĠĠĠ Ġ. transform Ġ{ Ġresponse Ġ-> ĊĠĠĠĠĠĠĠ Ġif Ġ( response Ġ!= ĠGet Event Groups Response . getDefaultInstance ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġfiltered Ġ= Ġresponse . groups List ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġgroup Ġ-> Ġgroup . events List Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sorted With ( query . sort Order ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġevent Ġ-> Ġevent . timestamp Ġ>= Ġlast TimeStamp Ġ&& Ġquery . filter ( event ) Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġfiltered . forEach Ġ{ Ġevent Ġ-> Ġemit ( Stream Event ( event )) Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġval Ġmax Time Event Ġ= Ġfiltered . max By OrNull Ġ{ Ġit . timestamp Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġmax Time Event ?. let Ġ{ Ġlast TimeStamp Ġ= Ġkotlin . math . max ( last TimeStamp , Ġit . timestamp Ġ+ Ġ 1 ) Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ. flow On ( dispatcher ) ĊĠ Ġ} ĊĊĠ Ġinternal Ġfun Ġclean Up () Ġ{ ĊĠĠĠ Ġis Closed . set ( true ) ĊĠ Ġ} Ċ }