private Collection < String > getPages ( String startDate , String endDate , List < Dimension > dimensions , ApiDimensionFilter countryFilter , Queue < Pair < String , FilterOperator > > toProcess , int rowLimit ) { String country = GoogleWebmasterFilter . countryFilterToString ( countryFilter ) ; ConcurrentLinkedDeque < String > allPages = new ConcurrentLinkedDeque <> ( ) ; int r = 0 ; while ( r <= GET_PAGES_RETRIES ) { ++ r ; log . info ( String . format ( "Get pages at round %d with size %d." , r , toProcess . size ( ) ) ) ; ConcurrentLinkedDeque < Pair < String , FilterOperator > > nextRound = new ConcurrentLinkedDeque <> ( ) ; ExecutorService es = Executors . newFixedThreadPool ( 10 , ExecutorsUtils . newDaemonThreadFactory ( Optional . of ( log ) , Optional . of ( this . getClass ( ) . getSimpleName ( ) ) ) ) ; while ( ! toProcess . isEmpty ( ) ) { submitJob ( toProcess . poll ( ) , countryFilter , startDate , endDate , dimensions , es , allPages , nextRound , rowLimit ) ; } //wait for jobs to finish and start next round if necessary. try { es . shutdown ( ) ; boolean terminated = es . awaitTermination ( 5 , TimeUnit . MINUTES ) ; if ( ! terminated ) { es . shutdownNow ( ) ; log . warn ( "Timed out while getting all pages for country-{} at round {}. Next round now has size {}." , country , r , nextRound . size ( ) ) ; } } catch ( InterruptedException e ) { throw new RuntimeException ( e ) ; } if ( nextRound . isEmpty ( ) ) { break ; } toProcess = nextRound ; coolDown ( r , PAGES_GET_COOLDOWN_TIME ) ; } if ( r == GET_PAGES_RETRIES + 1 ) { throw new RuntimeException ( String . format ( "Getting all pages reaches the maximum number of retires %d. Date range: %s ~ %s. Country: %s." , GET_PAGES_RETRIES , startDate , endDate , country ) ) ; } return allPages ; }
Get all pages in an async mode .