package Ġai . plat on . pulsar . common . collect Ċ Ċ import Ġai . plat on . pulsar . common . Priority 1 3 Ċ import Ġai . plat on . pulsar . common . collect . collector . Abstract Priority Data Collector Ċ import Ġai . plat on . pulsar . common . config . Volatile Config Ċ import Ġai . plat on . pulsar . common . metrics . App Metrics Ċ import Ġai . plat on . pulsar . common . options . Load Options Ċ import Ġai . plat on . pulsar . common . urls . Hyper link Ċ import Ġai . plat on . pulsar . common . urls . Hyper links Ċ import Ġai . plat on . pulsar . common . urls . Url Aware Ċ import Ġcom . co da h ale . metrics . Gauge Ċ import Ġcom . google . common . collect . Iter ators Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . nio . file . Path Ċ import Ġjava . time . Instant Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ Ċ open Ġclass ĠLocal File Hyper link Collector ( ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġpath Ġof Ġthe Ġfile Ġsource ĊĠĠĠĠĠĠĠĠ Ġ* Ġ*/ ĊĠĠĠĠĠĠĠ Ġval Ġpath : ĠPath , ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠThe Ġpriority ĊĠĠĠĠĠĠĠĠ Ġ* Ġ*/ ĊĠĠĠĠĠĠĠ Ġpriority : ĠInt Ġ= ĠPriority 1 3 . NORMAL . value , Ċ ): ĠAbstract Priority Data Collector < Url Aware >( priority ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Local File Hyper link Collector :: class . java ) ĊĠĠĠ Ġprivate Ġval Ġurl Loader Ġ= ĠLocal File Url Loader ( path ) ĊĠĠĠ Ġprivate Ġval Ġis Loaded Ġ= ĠAtomic Boolean () ĊĠĠĠ Ġprivate Ġval Ġcache : ĠMutable List < Hyper link > Ġ= ĠCollections . synchronized List ( LinkedList ()) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġcache Ġcapacity , Ġwe Ġassume Ġthat Ġall Ġitems Ġin Ġthe Ġfile Ġare Ġloaded Ġinto Ġthe Ġcache ĊĠĠĠĠ Ġ* Ġ*/ ĊĠĠĠ Ġoverride Ġval Ġcapacity : ĠInt Ġ= Ġ 1 _ 0 0 0 _ 0 0 0 ĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġcollector Ġname ĊĠĠĠĠ Ġ* Ġ*/ ĊĠĠĠ Ġoverride Ġvar Ġname : ĠString Ġ= Ġpath . fileName . toString () ĊĊĠĠĠ Ġval ĠfileName Ġ= Ġpath . fileName . toString () ĊĊĠĠĠ Ġvar Ġload Args : ĠString ? Ġ= Ġnull ĊĊĠĠĠ Ġval Ġhyper links : ĠList < Hyper link > Ġget () Ġ= Ġensure Loaded (). cache ĊĊĠĠĠ Ġoverride Ġval Ġsize : ĠInt Ġget () Ġ= Ġhyper links . size ĊĊĠĠĠ Ġconstructor ( path : ĠPath , Ġpriority : ĠPriority 1 3 ): Ġthis ( path , Ġpriority . value ) ĊĊĠĠĠ Ġoverride Ġfun Ġhas More () Ġ= Ġhyper links . isNotEmpty () ĊĊĠĠĠ Ġoverride Ġfun Ġcollect To ( sink : ĠMutable List < Url Aware >): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġbefore Collect () ĊĊĠĠĠĠĠĠĠ Ġval Ġcount Ġ= Ġcache . remove First OrNull ()?. take If Ġ{ Ġsink . add ( it ) Ġ} ?. let Ġ{ Ġ 1 Ġ} Ġ?: Ġ 0 ĊĊĠĠĠĠĠĠĠ Ġreturn Ġafter Collect ( count ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Synchron ized ĊĠĠĠ Ġoverride Ġfun Ġdump (): ĠList < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġhyper links . map Ġ{ Ġit . toString () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġensure Loaded (): ĠLocal File Hyper link Collector Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( is Loaded . compare AndSet ( false , Ġtrue )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġremaining Capacity Ġ= Ġcapacity Ġ- Ġcache . size ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġgroup Ġ= ĠUrl Topic ("", Ġ 0 , Ġpriority , Ġcapacity ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġurl Loader . load To Now ( cache , Ġremaining Capacity , Ġgroup ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġargs Ġ= ĠLoad Options . merge ( it . args , Ġload Args , ĠVol atile Config . UN SAFE ). toString () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHyper links . to Hyper link ( it ). also Ġ{ Ġit . args Ġ= Ġargs Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġif Ġ( load Args Ġ!= Ġnull ) Ġ" Ġ| Ġ$ load Args Ġ" Ġelse Ġ"" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Loaded Ġtotal Ġ{} Ġurls Ġfrom Ġfile Ġ| Ġ$ msg {}", Ġcache . size , Ġpath ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġthis ĊĠĠĠ Ġ} Ċ } Ċ Ċ open Ġclass ĠCircular Local File Hyper link Collector ( ĊĠĠĠĠĠĠĠ Ġpath : ĠPath , ĊĠĠĠĠĠĠĠ Ġpriority : ĠPriority 1 3 Ġ= ĠPriority 1 3 . NORMAL Ċ ): ĠLocal File Hyper link Collector ( path , Ġpriority . value ) Ġ{ ĊĊĠĠĠ Ġoverride Ġvar Ġname : ĠString Ġ= Ġ" Circular LF HC " ĊĊĠĠĠ Ġprotected Ġval Ġiterator Ġ= ĠIter ators . cycle ( hyper links ) ĊĊĠĠĠ Ġoverride Ġfun Ġcollect To ( sink : ĠMutable List < Url Aware >): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġbefore Collect () ĊĊĠĠĠĠĠĠĠ Ġvar Ġcount Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġif Ġ( hasMore () Ġ&& Ġiterator . hasNext ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcount Ġ+= Ġcollect To ( iterator . next (), Ġsink ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġafter Collect ( count ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ open Ġclass ĠPeriod ical Local File Hyper link Collector ( ĊĠĠĠĠĠĠĠ Ġpath : ĠPath , ĊĠĠĠĠĠĠĠ Ġval Ġoptions : ĠLoad Options , ĊĠĠĠĠĠĠĠ Ġpriority : ĠPriority 1 3 Ġ= ĠPriority 1 3 . NORMAL , Ċ ): ĠCircular Local File Hyper link Collector ( path , Ġpriority ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Period ical Local File Hyper link Collector :: class . java ) ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdata Ġclass ĠCounter s ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġcollect s : ĠInt Ġ= Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġcollected : ĠInt Ġ= Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġround : ĠInt Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġglobal Counters Ġ= ĠCounter s () ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġg au ges Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" collect s " Ġto ĠG auge Ġ{ Ġglobal Counters . collect s Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" collected " Ġto ĠG auge Ġ{ Ġglobal Counters . collected Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" round " Ġto ĠG auge Ġ{ Ġglobal Counters . round Ġ} ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠApp Metrics . reg . register All ( this , Ġg au ges ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġvar Ġname : ĠString Ġ= Ġ" Period ical LF HC " ĊĊĠĠĠ Ġprivate Ġval Ġposition Ġ= ĠAtomicInteger () ĊĠĠĠ Ġval Ġuuid Ġ= ĠUUID . randomUUID () ĊĠĠĠ Ġval Ġcounters Ġ= ĠCounter s () ĊĊĠĠĠ Ġvar Ġbatch Size Ġ= Ġ 1 0 ĊĊĠĠĠ Ġval Ġstart Times Ġ= Ġmutable Map Of < Int , ĠInstant >() ĊĠĠĠ Ġval Ġfinish Times Ġ= Ġmutable Map Of < Int , ĠInstant >() ĊĊĠĠĠ Ġval Ġround Ġget () Ġ= Ġcounters . round ĊĠĠĠ Ġvar Ġround Collected Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġprivate Ġset ĊĠĠĠ Ġval ĠstartTime Ġget () Ġ= Ġstart Times [ counters . round ] ?: ĠInstant . EPOCH ĊĠĠĠ Ġval Ġfinish Time Ġget () Ġ= Ġfinish Times [ counters . round ] ?: ĠInstant . EPOCH ĊĠĠĠ Ġval Ġexpires Ġget () Ġ= Ġoptions . expires ĊĠĠĠ Ġval Ġis Started Ġget () Ġ= Ġcounters . round Ġ> Ġ 0 ĊĠĠĠ Ġval Ġis Expired Ġget () Ġ= Ġis Finished Ġ&& Ġ( startTime Ġ+ Ġexpires Ġ< ĠInstant . now ()) ĊĠĠĠ Ġval Ġis Finished Ġget () Ġ= Ġposition . get () Ġ>= Ġhyper links . size ĊĊĠĠĠ Ġoverride Ġfun Ġhas More () Ġ= Ġ(! is Finished Ġ|| Ġis Expired ) Ġ&& Ġiterator . hasNext () ĊĊĠĠĠ Ġoverride Ġfun Ġcollect To ( sink : ĠMutable List < Url Aware >): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġbefore Collect () ĊĠĠĠĠĠĠĠ Ġ++ counters . collect s ĊĊĠĠĠĠĠĠĠ Ġreset If Ne cessary () ĊĊĠĠĠĠĠĠĠ Ġvar Ġi Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġcount Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġwhile Ġ( i ++ Ġ< Ġbatch Size Ġ&& Ġhas More () Ġ&& Ġiterator . hasNext ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcount Ġ+= Ġcollect To ( iterator . next (), Ġsink ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ++ counters . collected ĊĠĠĠĠĠĠĠĠĠĠĠ Ġposition . increment AndGet () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( is Finished ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfinish Times [ round ] Ġ= ĠInstant . now () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġround Collected Ġ+= Ġcount ĊĊĠĠĠĠĠĠĠ Ġreturn Ġafter Collect ( count ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġ"$ name Ġ- Ġround : Ġ$ round Ġcollected : Ġ${ counters . collected } Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" startTime : Ġ$ startTime Ġexpires : Ġ$ expires Ġpriority : Ġ$ priority Ġ| Ġ${ super . toString () }" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġreset If Ne cessary () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( is Expired ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġposition . set ( 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( position . get () Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ++ counters . round ĊĠĠĠĠĠĠĠĠĠĠĠ Ġround Collected Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstart Times [ round ] Ġ= ĠInstant . now () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Round Ġ{} Ġfetching Ġ{} Ġhyper links Ġin Ġlocal Ġfile Ġ| Ġ{} Ġ{} Ġ| Ġ{}", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġround , Ġhyper links . size , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Times [ round ], Ġexpires , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpath ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ