From 8ec7ce33638b23bf02bf40ff5efb2809ccefb52f Mon Sep 17 00:00:00 2001 From: Alex Osborne Date: Fri, 23 Jul 2021 17:38:56 +0900 Subject: [PATCH] ExtractorChrome: replay the recorded CrawlURI response to the browser By intercepting the browser's request and fulfilling it using the response previously recorded by FetchHTTP we avoid sending duplicate requests for the CrawlURI to the web server. A size limit (maxReplayLength) is applied as a safety measure since the browser's Fetch.fulfillRequest API requires us to load the entire response body into memory. Note: This only applies to the main CrawlURI. The browser can still make duplicate requests when loading sub-resources. Solving this for sub-resources will require implementing the ability to read back previously written WARC records. --- .../modules/extractor/ExtractorChrome.java | 60 ++++++++++++-- .../org/archive/net/chrome/ChromeRequest.java | 9 +++ .../org/archive/net/chrome/ChromeWindow.java | 33 +++++--- .../net/chrome/InterceptedRequest.java | 78 +++++++++++++++++++ .../extractor/ExtractorChromeTest.java | 25 +++++- 5 files changed, 185 insertions(+), 20 deletions(-) create mode 100644 contrib/src/main/java/org/archive/net/chrome/InterceptedRequest.java diff --git a/contrib/src/main/java/org/archive/modules/extractor/ExtractorChrome.java b/contrib/src/main/java/org/archive/modules/extractor/ExtractorChrome.java index 8bc74da5..35676992 100644 --- a/contrib/src/main/java/org/archive/modules/extractor/ExtractorChrome.java +++ b/contrib/src/main/java/org/archive/modules/extractor/ExtractorChrome.java @@ -19,14 +19,12 @@ package org.archive.modules.extractor; +import org.apache.commons.io.IOUtils; import org.archive.crawler.event.CrawlURIDispositionEvent; import org.archive.crawler.framework.CrawlController; import org.archive.crawler.framework.Frontier; import org.archive.modules.CrawlURI; -import org.archive.net.chrome.ChromeClient; -import org.archive.net.chrome.ChromeProcess; -import org.archive.net.chrome.ChromeRequest; -import org.archive.net.chrome.ChromeWindow; +import org.archive.net.chrome.*; import org.archive.spring.KeyedProperties; import org.archive.util.Recorder; import org.json.JSONArray; @@ -37,6 +35,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.SequenceInputStream; import java.util.Arrays; +import java.util.Map; import java.util.concurrent.ExecutionException; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -46,11 +45,11 @@ import java.util.regex.Pattern; import static java.nio.charset.StandardCharsets.US_ASCII; import static java.util.Collections.enumeration; -import static java.util.logging.Level.INFO; -import static java.util.logging.Level.WARNING; +import static java.util.logging.Level.*; import static java.util.regex.Pattern.CASE_INSENSITIVE; import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.FAILED; import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.SUCCEEDED; +import static org.archive.modules.CoreAttributeConstants.A_HTTP_RESPONSE_HEADERS; import static org.archive.modules.CrawlURI.FetchType.*; /** @@ -115,6 +114,12 @@ public class ExtractorChrome extends ContentExtractor { */ private boolean captureRequests = true; + /** + * The maximum size response body that can be replayed to the browser. Setting this to -1 will cause all requests by + * the browser to be made against the live web. + */ + private int maxReplayLength = 100 * 1024 * 1024; + private Semaphore openWindowsSemaphore = null; private ChromeProcess process = null; private ChromeClient client = null; @@ -149,6 +154,8 @@ public class ExtractorChrome extends ContentExtractor { private void visit(CrawlURI curi) throws InterruptedException { try (ChromeWindow window = client.createWindow(windowWidth, windowHeight)) { + window.interceptRequests(request -> handleInterceptedRequest(curi, request)); + if (captureRequests) { window.captureRequests(request -> handleCapturedRequest(curi, request)); } @@ -170,7 +177,48 @@ public class ExtractorChrome extends ContentExtractor { } } + private void handleInterceptedRequest(CrawlURI curi, InterceptedRequest interceptedRequest) { + ChromeRequest request = interceptedRequest.getRequest(); + if (request.getMethod().equals("GET") && request.getUrl().equals(curi.getURI())) { + replayResponseToBrowser(curi, interceptedRequest); + } else { + interceptedRequest.continueNormally(); + } + } + + @SuppressWarnings("unchecked") + private void replayResponseToBrowser(CrawlURI curi, InterceptedRequest interceptedRequest) { + // There seems to be no easy way to stream the body to the browser so we slurp it into + // memory with a size limit. The one way I can see to achieve streaming is to have Heritrix + // serve the request over its HTTP server and pass a Heritrix URL to Fetch.fulfillRequest + // instead of the body directly. We might need to do that if memory pressure becomes a + // problem but for now just keep it simple. + + long bodyLength = curi.getRecorder().getResponseContentLength(); + if (bodyLength > maxReplayLength) { + logger.log(FINE, "Page body too large to replay: {0}", curi.getURI()); + interceptedRequest.continueNormally(); + return; + } + + byte[] body = new byte[(int)bodyLength]; + try (InputStream stream = curi.getRecorder().getContentReplayInputStream()) { + IOUtils.readFully(stream, body); + } catch (IOException e) { + logger.log(WARNING, "Error reading back page body: " + curi.getURI(), e); + interceptedRequest.continueNormally(); + return; + } + + Map headers = (Map) curi.getData().get(A_HTTP_RESPONSE_HEADERS); + interceptedRequest.fulfill(curi.getFetchStatus(), headers.entrySet(), body); + } + private void handleCapturedRequest(CrawlURI via, ChromeRequest request) { + if (request.isResponseFulfilledByInterception()) { + return; + } + Recorder recorder = new Recorder(controller.getScratchDir().getFile(), controller.getRecorderOutBufferBytes(), controller.getRecorderInBufferBytes()); diff --git a/contrib/src/main/java/org/archive/net/chrome/ChromeRequest.java b/contrib/src/main/java/org/archive/net/chrome/ChromeRequest.java index 68ce072b..17530e9c 100644 --- a/contrib/src/main/java/org/archive/net/chrome/ChromeRequest.java +++ b/contrib/src/main/java/org/archive/net/chrome/ChromeRequest.java @@ -33,6 +33,7 @@ public class ChromeRequest { private JSONObject rawResponseHeaders; private String responseHeadersText; private final long beginTime = System.currentTimeMillis(); + private boolean responseFulfilledByInterception; public ChromeRequest(ChromeWindow window, String id) { this.window = window; @@ -151,4 +152,12 @@ public class ChromeRequest { void setRequestJson(JSONObject requestJson) { this.requestJson = requestJson; } + + void setResponseFulfilledByInterception(boolean responseFulfilledByInterception) { + this.responseFulfilledByInterception = responseFulfilledByInterception; + } + + public boolean isResponseFulfilledByInterception() { + return responseFulfilledByInterception; + } } diff --git a/contrib/src/main/java/org/archive/net/chrome/ChromeWindow.java b/contrib/src/main/java/org/archive/net/chrome/ChromeWindow.java index a65fa8cc..e54b4583 100644 --- a/contrib/src/main/java/org/archive/net/chrome/ChromeWindow.java +++ b/contrib/src/main/java/org/archive/net/chrome/ChromeWindow.java @@ -27,8 +27,7 @@ import java.util.concurrent.*; import java.util.function.Consumer; import java.util.logging.Logger; -import static java.util.logging.Level.FINE; -import static java.util.logging.Level.WARNING; +import static java.util.logging.Level.*; /** * A browser window or tab. @@ -43,6 +42,7 @@ public class ChromeWindow implements Closeable { private CompletableFuture loadEventFuture; private final Map requestMap = new ConcurrentHashMap<>(); private Consumer requestConsumer; + private Consumer requestInterceptor; private final ExecutorService eventExecutor; public ChromeWindow(ChromeClient client, String targetId) { @@ -136,6 +136,22 @@ public class ChromeWindow implements Closeable { } } + private void handlePausedRequest(JSONObject params) { + String networkId = params.getString("networkId"); + ChromeRequest request = requestMap.computeIfAbsent(networkId, id -> new ChromeRequest(this, id)); + request.setRequestJson(params.getJSONObject("request")); + String id = params.getString("requestId"); + InterceptedRequest interceptedRequest = new InterceptedRequest(this, id, request); + try { + requestInterceptor.accept(interceptedRequest); + } catch (Exception e) { + logger.log(SEVERE, "Request interceptor threw", e); + } + if (!interceptedRequest.isHandled()) { + interceptedRequest.continueNormally(); + } + } + private void handleRequestWillBeSent(JSONObject params) { String requestId = params.getString("requestId"); ChromeRequest request = requestMap.computeIfAbsent(requestId, id -> new ChromeRequest(this, id)); @@ -198,15 +214,8 @@ public class ChromeWindow implements Closeable { call("Network.enable"); } - private void handlePausedRequest(JSONObject params) { - if (params.has("responseStatusCode")) { - String stream = call("Fetch.takeResponseBodyAsStream", "requestId", - params.getString("requestId")).getString("stream"); - System.out.println(call("IO.read", "handle", stream)); - System.out.println("stream " + stream); - call("IO.close", "handle", stream); - } else { - call("Fetch.continueRequest", "requestId", params.getString("requestId")); - } + public void interceptRequests(Consumer requestInterceptor) { + this.requestInterceptor = requestInterceptor; + call("Fetch.enable"); } } \ No newline at end of file diff --git a/contrib/src/main/java/org/archive/net/chrome/InterceptedRequest.java b/contrib/src/main/java/org/archive/net/chrome/InterceptedRequest.java new file mode 100644 index 00000000..39c69a5e --- /dev/null +++ b/contrib/src/main/java/org/archive/net/chrome/InterceptedRequest.java @@ -0,0 +1,78 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. + * + * The IA licenses this file to You 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 org.archive.net.chrome; + +import org.json.JSONArray; +import org.json.JSONObject; + +import java.util.Base64; +import java.util.Collection; +import java.util.Map; + +public class InterceptedRequest { + private final String id; + private final ChromeRequest request; + private final ChromeWindow window; + private boolean handled; + + public InterceptedRequest(ChromeWindow window, String id, ChromeRequest request) { + this.window = window; + this.id = id; + this.request = request; + } + + public ChromeRequest getRequest() { + return request; + } + + public void fulfill(int status, Collection> headers, byte[] body) { + setHandled(); + JSONArray headerArray = new JSONArray(); + for (Map.Entry entry : headers) { + JSONObject object = new JSONObject(); + object.put("name", entry.getKey()); + object.put("value", entry.getValue()); + headerArray.put(object); + } + String encodedBody = Base64.getEncoder().encodeToString(body); + request.setResponseFulfilledByInterception(true); + window.call("Fetch.fulfillRequest", + "requestId", id, + "responseCode", status, + "responseHeaders", headerArray, + "body", encodedBody); + } + + public void continueNormally() { + setHandled(); + window.call("Fetch.continueRequest", "requestId", id); + } + + public boolean isHandled() { + return handled; + } + + private void setHandled() { + if (handled) { + throw new IllegalStateException("intercepted request already handled"); + } + handled = true; + } +} diff --git a/contrib/src/test/java/org/archive/modules/extractor/ExtractorChromeTest.java b/contrib/src/test/java/org/archive/modules/extractor/ExtractorChromeTest.java index 17a9db14..6bdb16e6 100644 --- a/contrib/src/test/java/org/archive/modules/extractor/ExtractorChromeTest.java +++ b/contrib/src/test/java/org/archive/modules/extractor/ExtractorChromeTest.java @@ -22,10 +22,15 @@ package org.archive.modules.extractor; import org.archive.crawler.framework.CrawlController; import org.archive.crawler.framework.Frontier; import org.archive.crawler.reporting.CrawlerLoggerModule; +import org.archive.modules.CrawlMetadata; import org.archive.modules.CrawlURI; import org.archive.modules.DispositionChain; import org.archive.modules.Processor; +import org.archive.modules.fetcher.DefaultServerCache; +import org.archive.modules.fetcher.FetchHTTP; +import org.archive.modules.fetcher.SimpleCookieStore; import org.archive.net.UURIFactory; +import org.archive.util.Recorder; import org.eclipse.jetty.server.Request; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.handler.AbstractHandler; @@ -100,7 +105,7 @@ public class ExtractorChromeTest { public TemporaryFolder tempFolder = new TemporaryFolder(); @Test - public void test() throws IOException { + public void test() throws IOException, InterruptedException { List processedURIs = Collections.synchronizedList(new ArrayList<>()); CrawlController controller = new CrawlController(); DispositionChain dispositionChain = new DispositionChain(); @@ -136,22 +141,38 @@ public class ExtractorChromeTest { replay(frontier); controller.setFrontier(frontier); + FetchHTTP fetchHTTP = new FetchHTTP(); + fetchHTTP.setServerCache(new DefaultServerCache()); + fetchHTTP.setCookieStore(new SimpleCookieStore()); + fetchHTTP.setUserAgentProvider(new CrawlMetadata()); + fetchHTTP.start(); + ExtractorChrome extractor = new ExtractorChrome(controller, event -> { /* ignored */ }); try { extractor.start(); } catch (RuntimeException e) { assumeNoException("Unable to start Chrome", e); } + + Recorder recorder = new Recorder(tempFolder.newFile(), 1024, 1024); try { CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://127.0.0.1:7778/")); + + Recorder.setHttpRecorder(recorder); + curi.setRecorder(recorder); + fetchHTTP.process(curi); + extractor.innerExtract(curi); + List outLinks = curi.getOutLinks().stream().map(CrawlURI::toString).sorted().collect(toList()); assertEquals(Collections.singletonList("http://example.org/page2.html"), outLinks); } finally { extractor.stop(); + fetchHTTP.stop(); + recorder.cleanup(); } - assertEquals(3, processedURIs.size()); + assertEquals(2, processedURIs.size()); for (CrawlURI curi: processedURIs) { assertEquals(200, curi.getFetchStatus()); assertEquals(curi.getUURI().getPath().equals("/post") ? HTTP_POST : HTTP_GET, curi.getFetchType());