package Ġai . plat on . pulsar . crawl . fetch Ċ Ċ import Ġai . plat on . pulsar . common .* Ċ import Ġai . plat on . pulsar . common . config . Capability Types . BROWSER _ MAX _ ACTIVE _ TAB S Ċ import Ġai . plat on . pulsar . common . config . Capability Types . PRIV ACY _ CONTEXT _ NUMBER Ċ import Ġai . plat on . pulsar . common . config . Immutable Config Ċ import Ġai . plat on . pulsar . common . message . Loaded Page Formatter Ċ import Ġai . plat on . pulsar . common . metrics . App Metrics Ċ import Ġai . plat on . pulsar . common . metrics . Common Counter Ċ import Ġai . plat on . pulsar . crawl . component . Fetch Component Ċ import Ġai . plat on . pulsar . crawl . component . Parse Component Ċ import Ġai . plat on . pulsar . persist . Web Page Ċ import Ġai . plat on . pulsar . persist . gor a . generated . G Web Page Ċ import Ġkotlinx . coroutines .* Ċ import Ġorg . apache . hadoop . io . Int Writable Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġo shi . System Info Ċ import Ġjava . io . IOException Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . Array BlockingQueue Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ import Ġkotlin . math . abs Ċ Ċ /** Ċ Ġ* ĠThis Ġclass Ġp icks Ġitems Ġfrom Ġqueues Ġand Ġf etches Ġthe Ġpages Ċ Ġ*/ Ċ class ĠFetch Loop ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġfetch Monitor : ĠFetch Monitor , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġfetch Component : ĠFetch Component , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġparse Component : ĠParse Component , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġtask Scheduler : ĠTask Scheduler , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġcontext : ĠRed ucer Context < Int Writable , Ġout ĠIF etch Entry , ĠString , ĠG Web Page >, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġimmutable Config : ĠImmutable Config Ċ ): ĠAuto Closeable , ĠComparable < Fetch Loop > Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġinstance Se qu encer Ġ= ĠAtomicInteger () ĊĠĠĠĠĠĠĠ Ġval Ġpending Tasks Ġ= ĠArray BlockingQueue < Job Fetch Task >( 1 0 0 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġnum Running Tasks Ġ= ĠAtomicInteger () ĊĠĠĠĠĠĠĠ Ġval Ġillegal State Ġ= ĠAtomic Boolean () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġenum Counters Ġ= ĠApp Metrics . reg . enum Counter Registry ĊĊĠĠĠ Ġval Ġid Ġ= Ġinstance Se qu encer . increment AndGet () ĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Fetch Loop :: class . java ) ĊĠĠĠ Ġprivate Ġval Ġloop Config Ġ= Ġimmutable Config . to Volatile Config () ĊĠĠĠ Ġprivate Ġval Ġclosed Ġ= ĠAtomic Boolean ( false ) ĊĊĠĠĠ Ġprivate Ġval Ġnum Privacy Contexts Ġ= Ġimmutable Config . getInt ( PRIV ACY _ CONTEXT _ NUMBER , Ġ 2 ) ĊĠĠĠ Ġprivate Ġval Ġfetch Concurrency Ġ= Ġnum Privacy Contexts Ġ* Ġimmutable Config . getInt ( BROWSER _ MAX _ ACTIVE _ TAB S , ĠApp Context . N CPU ) ĊĊĠĠĠ Ġprivate Ġval Ġis App Active Ġget () Ġ= Ġ! fetch Monitor . is Mission Complete Ġ&& Ġ! closed . get () Ġ&& Ġ! il legal State . get () ĊĊĠĠĠ Ġprivate Ġval Ġsystem Info Ġ= ĠSystem Info () ĊĠĠĠ Ġ// ĠO SH I Ġcached Ġthe Ġvalue , Ġso Ġit 's Ġfast Ġand Ġsafe Ġto Ġbe Ġcalled Ġfrequently ĊĠĠĠ Ġprivate Ġval Ġavailable Memory Ġget () Ġ= Ġsystem Info . hardware . memory . available ĊĠĠĠ Ġprivate Ġval Ġinstance Required Memory Ġ= Ġ 5 0 0 L Ġ* Ġ 1 0 2 4 Ġ* Ġ 1 0 2 4 Ġ// Ġ 5 0 0 ĠM iB ĊĠĠĠ Ġprivate Ġval Ġmemory Remaining Ġget () Ġ= Ġavailable Memory Ġ- Ġinstance Required Memory ĊĠĠĠ Ġprivate Ġval Ġfetch Task Timeout Ġ= ĠDuration . of Minutes ( 5 ) ĊĊĠĠĠ Ġsuspend Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġloop Ġ= Ġthis ĊĊĠĠĠĠĠĠĠ Ġsupervisor Scope Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfetch Monitor . register Fetch Loop ( loop ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġj Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( is App Active ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ++ j ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġto Ġprevent ĠCPU Ġoverload ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( is App Active Ġ&& Ġnum Running Tasks . get () Ġ> Ġfetch Concurrency ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( is App Active Ġ&& Ġmemory Remaining Ġ< Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( j Ġ% Ġ 2 0 Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandle Memory Short age ( j ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Running Tasks . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( Dispatch ers . Default Ġ+ ĠCor outine Name (" w ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschedule ()?. let Ġ{ Ġfetch ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIllegal ApplicationContext StateException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠApp Context . begin Terminate () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġillegal State . set ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Illegal Ġcontext Ġstate Ġ| Ġ{}", Ġe . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠTimeout Cancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Coroutine Ġtimeout Ġcanceled Ġ| Ġ{}", Ġe . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Unexpected Ġexception ", Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Running Tasks . decrement AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġfetch Monitor . unregister Fetch Loop ( loop ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( closed . compare AndSet ( false , Ġtrue )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Fetch Ġloop Ġ# {} Ġis Ġclosed ", Ġid ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcompare To ( other : ĠFetch Loop ) Ġ= Ġid Ġ- Ġother . id ĊĊĠĠĠ Ġoverride Ġfun ĠhashCode () Ġ= Ġid ĊĊĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?) Ġ= Ġother Ġis ĠFetch Loop Ġ&& Ġid Ġ== Ġother . id ĊĊĠĠĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ"# $ id " ĊĊĠĠĠ Ġprivate Ġfun Ġschedule (): ĠJob Fetch Task ? Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġfind Ġtasks Ġfrom Ġpending Ġqueue ĊĠĠĠĠĠĠĠ Ġ// ĠTODO : Ġpending Tasks Ġis Ġnot Ġused ĊĠĠĠĠĠĠĠ Ġvar Ġfetch Task Ġ= Ġpending Tasks . poll () ĊĠĠĠĠĠĠĠ Ġif Ġ( fetch Task Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfetch Task ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfetch Task Ġ= Ġtask Scheduler . schedule () ĊĠĠĠĠĠĠĠ Ġlog . take If Ġ{ Ġit . is Trace Enabled Ġ} ?. trace (" Scheduled Ġtask Ġfrom Ġpool Ġ{} Ġ| Ġ{}", Ġfetch Task ?. pool Id , Ġfetch Task ) ĊĊĠĠĠĠĠĠĠ Ġreturn Ġfetch Task ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġfetch ( task : ĠJob Fetch Task ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! is App Active ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtask Scheduler . finish ( task . pool Id , Ġtask . itemId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpage Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfetch Component . fetch Content Deferred ( task . page ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask Scheduler . finish ( task . pool Id , Ġtask . itemId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( page . is Internal ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" Fetch Ġtask Ġis Ġan Ġinternal Ġpage Ġ| Ġ{}", Ġtask . url String ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġis Canceled Ġ= Ġpage . protocol Status . is Canceled ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! is Canceled Ġ&& Ġtask Scheduler . parse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġparse Result Ġ= Ġparse Component . parse ( page , Ġnull , Ġfalse , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( log . is Trace Enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . trace (" Parse Result : Ġ{} ĠParse Report : Ġ{}", Ġparse Result , Ġparse Component . get Trace Info ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith Context ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . take If Ġ{ Ġit . is Info Enabled Ġ} ?. info ( Loaded Page Formatter ( page ). toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! is Canceled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwrite ( page . key , Ġpage ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġenum Counters . inc ( Common Counter . r Persist ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Unexpected Ġthrowable ", Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġwrite ( key : ĠString , Ġpage : ĠWeb Page ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthe Ġpage Ġis Ġfetched Ġand Ġstatus Ġare Ġupdated , Ġwrite Ġto Ġthe Ġfile Ġsystem ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . write ( key , Ġpage . un box ()) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġwrite Ġto Ġh dfs Ġ- Ġ{}", ĠStrings . stringify Exception ( e )) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠInterruptedException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Inter rupted Ġ- Ġ{}", ĠStrings . stringify Exception ( e )) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠThread . currentThread (). interrupt () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error ( Strings . stringify Exception ( e )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Memory Short age ( j : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info ("$ j .\ t num Running : Ġ{}, Ġavailable Memory : Ġ{}, Ġrequired Memory : Ġ{}, Ġshort age : Ġ{}", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnum Running Tasks , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStrings . readable Bytes ( available Memory ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStrings . readable Bytes ( instance Required Memory ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠStrings . readable Bytes ( abs ( memory Remaining )) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ ĠSystem . gc () ĊĠĠĠ Ġ} Ċ } Ċ