@Test public void canAbortUnconditionalDownloadWhenTerminationThresholdTimeLimitExceeded() throws Exception { final SlaveDownloader slaveDownloader = new SlaveDownloader(); final HttpRetrieveResponse response = httpRetrieveResponseFactory.create(ResponseType.DISK_STORAGE, pathOnDisk); final ProcessingJobLimits limits = new ProcessingJobLimits( 10l , 5 * 1000l , 10 * 1000l , 10 , 100 * 1000l ); final RetrieveUrl task = new RetrieveUrl(image1GitHubUrl, limits, DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD, "jobid-1", "referenceid-1", Collections.<String, String>emptyMap(), new ProcessingJobTaskDocumentReference(DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD, "source-reference-1", Collections.<ProcessingJobSubTask>emptyList()), null,new ReferenceOwner("unknown","unknwon","unknown")); slaveDownloader.downloadAndStoreInHttpRetrieveResponse(response, task); assertEquals(RetrievingState.FINISHED_TIME_LIMIT, response.getState()); assertEquals(pathOnDisk, response.getAbsolutePath()); assertNotNull(response.getSourceIp()); assertEquals(response.getContentSizeInBytes().longValue(), 0); }
public HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task) { if ((task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) && (task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD)) { throw new IllegalArgumentException("The downloader can handle only condition & unconditional downloads. Cannot handle " + task.getDocumentReferenceTask().getTaskType()); } if (task.getUrl() == null || "".equalsIgnoreCase(task.getUrl())) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Retrieval of the media failed, URL is empty or null."); httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("Retrieval of the media failed, URL is empty or null."); return httpRetrieveResponse; } try { new URL(task.getUrl()); } catch (Exception e) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("An error has occurred:" + e.getMessage()); httpRetrieveResponse.setException(e); return httpRetrieveResponse; } final AsyncHttpClient asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder() .setMaxRedirects(task.getLimits().getRetrievalMaxNrOfRedirects()) .setFollowRedirect(true) .setConnectTimeout(100000) .setAcceptAnyCertificate(true) .setMaxRequestRetry(3) .build()); httpRetrieveResponse.setState(RetrievingState.PROCESSING); final long connectionSetupStartTimestamp = System.currentTimeMillis(); final ListenableFuture<Integer> downloadListener = asyncHttpClient.prepareGet(task.getUrl()).execute(new AsyncHandler<Integer>() { final TimeWindowCounter timeWindowCounter = new TimeWindowCounter(); @Override public STATE onStatusReceived(HttpResponseStatus status) throws Exception { final long connectionSetupDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setSocketConnectToDownloadStartDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setCheckingDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setUrl(new URL(task.getUrl())); httpRetrieveResponse.setSourceIp(Slave.URL_RESOLVER.resolveIpOfUrl(task.getUrl())); if (connectionSetupDurationInMillis > task.getLimits().getRetrievalConnectionTimeoutInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to initiate the connection (" + connectionSetupDurationInMillis + "ms, larger than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms)"); return STATE.ABORT; } httpRetrieveResponse.setHttpResponseCode(status.getStatusCode()); return STATE.CONTINUE; } @Override public STATE onHeadersReceived(HttpResponseHeaders downloadResponseHeaders) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); for (final Map.Entry<String, List<String>> entry : downloadResponseHeaders.getHeaders()) { final String header = entry.getKey(); if (!entry.getValue().isEmpty()) { final String firstValue = entry.getValue().get(0); httpRetrieveResponse.addHeader(header, firstValue); } } if (httpRetrieveResponse.getHttpResponseCode() >= 400) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("A HTTP error received (code >= 400), the download did not initiate."); return STATE.ABORT; } if (task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { final String existingContentLength = fetchContentLengthHeader(task.getHeaders()); final String downloadContentLength = downloadResponseHeaders.getHeaders().getFirstValue("Content-Length"); if (existingContentLength != null && downloadContentLength != null && existingContentLength.trim().equalsIgnoreCase(downloadContentLength.trim())) { httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setLog("Same file already downloaded before, skipping download (" + existingContentLength + "==" + downloadContentLength + ")"); return STATE.ABORT; } } timeWindowCounter.start(); return STATE.CONTINUE; } @Override public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + "ms)"); return STATE.ABORT; } if ((timeWindowCounter.previousTimeWindowRate() != -1) && (timeWindowCounter.previousTimeWindowRate() < task.getLimits().getRetrievalTerminationThresholdReadPerSecondInBytes())) { httpRetrieveResponse.setState(RetrievingState.FINISHED_RATE_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as the speed in which we could download this file was too low (" + timeWindowCounter.currentTimeWindowRate() + "bytes) for a period longer than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms"); return STATE.ABORT; } timeWindowCounter.incrementCount(bodyPart.length()); httpRetrieveResponse.addContent(bodyPart.getBodyPartBytes()); return STATE.CONTINUE; } @Override public Integer onCompleted() throws Exception { if (httpRetrieveResponse.getState() == RetrievingState.PROCESSING) httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setRetrievalDurationInMilliSecs(System.currentTimeMillis() - connectionSetupStartTimestamp); try { httpRetrieveResponse.close(); } catch (IOException e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Failed to close the download response. This might be a bug in harvester slave code.", e1); } return 0; } @Override public void onThrowable(Throwable e) { try { final File retrievalFileStorage = Paths.get(httpRetrieveResponse.getAbsolutePath()).toFile(); if (retrievalFileStorage.exists()) { retrievalFileStorage.delete(); } } catch (Exception e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Cleaning of the incomplete download result file failed in an unexpected way. This might be a bug in harvester slave code.", e1); } httpRetrieveResponse.setState(RetrievingState.ERROR); final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + " ms)"); } if (httpRetrieveResponse.getState() == RetrievingState.COMPLETED && task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); } else { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } } }); try { Integer r = downloadListener.get(1, TimeUnit.DAYS ); LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Download finished with status {}", r); } catch (Exception e) { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } finally { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); asyncHttpClient.close(); return httpRetrieveResponse; } }
SlaveDownloader { public HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task) { if ((task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) && (task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD)) { throw new IllegalArgumentException("The downloader can handle only condition & unconditional downloads. Cannot handle " + task.getDocumentReferenceTask().getTaskType()); } if (task.getUrl() == null || "".equalsIgnoreCase(task.getUrl())) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Retrieval of the media failed, URL is empty or null."); httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("Retrieval of the media failed, URL is empty or null."); return httpRetrieveResponse; } try { new URL(task.getUrl()); } catch (Exception e) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("An error has occurred:" + e.getMessage()); httpRetrieveResponse.setException(e); return httpRetrieveResponse; } final AsyncHttpClient asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder() .setMaxRedirects(task.getLimits().getRetrievalMaxNrOfRedirects()) .setFollowRedirect(true) .setConnectTimeout(100000) .setAcceptAnyCertificate(true) .setMaxRequestRetry(3) .build()); httpRetrieveResponse.setState(RetrievingState.PROCESSING); final long connectionSetupStartTimestamp = System.currentTimeMillis(); final ListenableFuture<Integer> downloadListener = asyncHttpClient.prepareGet(task.getUrl()).execute(new AsyncHandler<Integer>() { final TimeWindowCounter timeWindowCounter = new TimeWindowCounter(); @Override public STATE onStatusReceived(HttpResponseStatus status) throws Exception { final long connectionSetupDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setSocketConnectToDownloadStartDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setCheckingDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setUrl(new URL(task.getUrl())); httpRetrieveResponse.setSourceIp(Slave.URL_RESOLVER.resolveIpOfUrl(task.getUrl())); if (connectionSetupDurationInMillis > task.getLimits().getRetrievalConnectionTimeoutInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to initiate the connection (" + connectionSetupDurationInMillis + "ms, larger than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms)"); return STATE.ABORT; } httpRetrieveResponse.setHttpResponseCode(status.getStatusCode()); return STATE.CONTINUE; } @Override public STATE onHeadersReceived(HttpResponseHeaders downloadResponseHeaders) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); for (final Map.Entry<String, List<String>> entry : downloadResponseHeaders.getHeaders()) { final String header = entry.getKey(); if (!entry.getValue().isEmpty()) { final String firstValue = entry.getValue().get(0); httpRetrieveResponse.addHeader(header, firstValue); } } if (httpRetrieveResponse.getHttpResponseCode() >= 400) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("A HTTP error received (code >= 400), the download did not initiate."); return STATE.ABORT; } if (task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { final String existingContentLength = fetchContentLengthHeader(task.getHeaders()); final String downloadContentLength = downloadResponseHeaders.getHeaders().getFirstValue("Content-Length"); if (existingContentLength != null && downloadContentLength != null && existingContentLength.trim().equalsIgnoreCase(downloadContentLength.trim())) { httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setLog("Same file already downloaded before, skipping download (" + existingContentLength + "==" + downloadContentLength + ")"); return STATE.ABORT; } } timeWindowCounter.start(); return STATE.CONTINUE; } @Override public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + "ms)"); return STATE.ABORT; } if ((timeWindowCounter.previousTimeWindowRate() != -1) && (timeWindowCounter.previousTimeWindowRate() < task.getLimits().getRetrievalTerminationThresholdReadPerSecondInBytes())) { httpRetrieveResponse.setState(RetrievingState.FINISHED_RATE_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as the speed in which we could download this file was too low (" + timeWindowCounter.currentTimeWindowRate() + "bytes) for a period longer than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms"); return STATE.ABORT; } timeWindowCounter.incrementCount(bodyPart.length()); httpRetrieveResponse.addContent(bodyPart.getBodyPartBytes()); return STATE.CONTINUE; } @Override public Integer onCompleted() throws Exception { if (httpRetrieveResponse.getState() == RetrievingState.PROCESSING) httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setRetrievalDurationInMilliSecs(System.currentTimeMillis() - connectionSetupStartTimestamp); try { httpRetrieveResponse.close(); } catch (IOException e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Failed to close the download response. This might be a bug in harvester slave code.", e1); } return 0; } @Override public void onThrowable(Throwable e) { try { final File retrievalFileStorage = Paths.get(httpRetrieveResponse.getAbsolutePath()).toFile(); if (retrievalFileStorage.exists()) { retrievalFileStorage.delete(); } } catch (Exception e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Cleaning of the incomplete download result file failed in an unexpected way. This might be a bug in harvester slave code.", e1); } httpRetrieveResponse.setState(RetrievingState.ERROR); final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + " ms)"); } if (httpRetrieveResponse.getState() == RetrievingState.COMPLETED && task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); } else { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } } }); try { Integer r = downloadListener.get(1, TimeUnit.DAYS ); LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Download finished with status {}", r); } catch (Exception e) { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } finally { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); asyncHttpClient.close(); return httpRetrieveResponse; } } }
SlaveDownloader { public HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task) { if ((task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) && (task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD)) { throw new IllegalArgumentException("The downloader can handle only condition & unconditional downloads. Cannot handle " + task.getDocumentReferenceTask().getTaskType()); } if (task.getUrl() == null || "".equalsIgnoreCase(task.getUrl())) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Retrieval of the media failed, URL is empty or null."); httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("Retrieval of the media failed, URL is empty or null."); return httpRetrieveResponse; } try { new URL(task.getUrl()); } catch (Exception e) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("An error has occurred:" + e.getMessage()); httpRetrieveResponse.setException(e); return httpRetrieveResponse; } final AsyncHttpClient asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder() .setMaxRedirects(task.getLimits().getRetrievalMaxNrOfRedirects()) .setFollowRedirect(true) .setConnectTimeout(100000) .setAcceptAnyCertificate(true) .setMaxRequestRetry(3) .build()); httpRetrieveResponse.setState(RetrievingState.PROCESSING); final long connectionSetupStartTimestamp = System.currentTimeMillis(); final ListenableFuture<Integer> downloadListener = asyncHttpClient.prepareGet(task.getUrl()).execute(new AsyncHandler<Integer>() { final TimeWindowCounter timeWindowCounter = new TimeWindowCounter(); @Override public STATE onStatusReceived(HttpResponseStatus status) throws Exception { final long connectionSetupDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setSocketConnectToDownloadStartDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setCheckingDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setUrl(new URL(task.getUrl())); httpRetrieveResponse.setSourceIp(Slave.URL_RESOLVER.resolveIpOfUrl(task.getUrl())); if (connectionSetupDurationInMillis > task.getLimits().getRetrievalConnectionTimeoutInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to initiate the connection (" + connectionSetupDurationInMillis + "ms, larger than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms)"); return STATE.ABORT; } httpRetrieveResponse.setHttpResponseCode(status.getStatusCode()); return STATE.CONTINUE; } @Override public STATE onHeadersReceived(HttpResponseHeaders downloadResponseHeaders) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); for (final Map.Entry<String, List<String>> entry : downloadResponseHeaders.getHeaders()) { final String header = entry.getKey(); if (!entry.getValue().isEmpty()) { final String firstValue = entry.getValue().get(0); httpRetrieveResponse.addHeader(header, firstValue); } } if (httpRetrieveResponse.getHttpResponseCode() >= 400) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("A HTTP error received (code >= 400), the download did not initiate."); return STATE.ABORT; } if (task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { final String existingContentLength = fetchContentLengthHeader(task.getHeaders()); final String downloadContentLength = downloadResponseHeaders.getHeaders().getFirstValue("Content-Length"); if (existingContentLength != null && downloadContentLength != null && existingContentLength.trim().equalsIgnoreCase(downloadContentLength.trim())) { httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setLog("Same file already downloaded before, skipping download (" + existingContentLength + "==" + downloadContentLength + ")"); return STATE.ABORT; } } timeWindowCounter.start(); return STATE.CONTINUE; } @Override public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + "ms)"); return STATE.ABORT; } if ((timeWindowCounter.previousTimeWindowRate() != -1) && (timeWindowCounter.previousTimeWindowRate() < task.getLimits().getRetrievalTerminationThresholdReadPerSecondInBytes())) { httpRetrieveResponse.setState(RetrievingState.FINISHED_RATE_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as the speed in which we could download this file was too low (" + timeWindowCounter.currentTimeWindowRate() + "bytes) for a period longer than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms"); return STATE.ABORT; } timeWindowCounter.incrementCount(bodyPart.length()); httpRetrieveResponse.addContent(bodyPart.getBodyPartBytes()); return STATE.CONTINUE; } @Override public Integer onCompleted() throws Exception { if (httpRetrieveResponse.getState() == RetrievingState.PROCESSING) httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setRetrievalDurationInMilliSecs(System.currentTimeMillis() - connectionSetupStartTimestamp); try { httpRetrieveResponse.close(); } catch (IOException e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Failed to close the download response. This might be a bug in harvester slave code.", e1); } return 0; } @Override public void onThrowable(Throwable e) { try { final File retrievalFileStorage = Paths.get(httpRetrieveResponse.getAbsolutePath()).toFile(); if (retrievalFileStorage.exists()) { retrievalFileStorage.delete(); } } catch (Exception e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Cleaning of the incomplete download result file failed in an unexpected way. This might be a bug in harvester slave code.", e1); } httpRetrieveResponse.setState(RetrievingState.ERROR); final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + " ms)"); } if (httpRetrieveResponse.getState() == RetrievingState.COMPLETED && task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); } else { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } } }); try { Integer r = downloadListener.get(1, TimeUnit.DAYS ); LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Download finished with status {}", r); } catch (Exception e) { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } finally { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); asyncHttpClient.close(); return httpRetrieveResponse; } }  }
SlaveDownloader { public HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task) { if ((task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) && (task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD)) { throw new IllegalArgumentException("The downloader can handle only condition & unconditional downloads. Cannot handle " + task.getDocumentReferenceTask().getTaskType()); } if (task.getUrl() == null || "".equalsIgnoreCase(task.getUrl())) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Retrieval of the media failed, URL is empty or null."); httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("Retrieval of the media failed, URL is empty or null."); return httpRetrieveResponse; } try { new URL(task.getUrl()); } catch (Exception e) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("An error has occurred:" + e.getMessage()); httpRetrieveResponse.setException(e); return httpRetrieveResponse; } final AsyncHttpClient asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder() .setMaxRedirects(task.getLimits().getRetrievalMaxNrOfRedirects()) .setFollowRedirect(true) .setConnectTimeout(100000) .setAcceptAnyCertificate(true) .setMaxRequestRetry(3) .build()); httpRetrieveResponse.setState(RetrievingState.PROCESSING); final long connectionSetupStartTimestamp = System.currentTimeMillis(); final ListenableFuture<Integer> downloadListener = asyncHttpClient.prepareGet(task.getUrl()).execute(new AsyncHandler<Integer>() { final TimeWindowCounter timeWindowCounter = new TimeWindowCounter(); @Override public STATE onStatusReceived(HttpResponseStatus status) throws Exception { final long connectionSetupDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setSocketConnectToDownloadStartDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setCheckingDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setUrl(new URL(task.getUrl())); httpRetrieveResponse.setSourceIp(Slave.URL_RESOLVER.resolveIpOfUrl(task.getUrl())); if (connectionSetupDurationInMillis > task.getLimits().getRetrievalConnectionTimeoutInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to initiate the connection (" + connectionSetupDurationInMillis + "ms, larger than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms)"); return STATE.ABORT; } httpRetrieveResponse.setHttpResponseCode(status.getStatusCode()); return STATE.CONTINUE; } @Override public STATE onHeadersReceived(HttpResponseHeaders downloadResponseHeaders) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); for (final Map.Entry<String, List<String>> entry : downloadResponseHeaders.getHeaders()) { final String header = entry.getKey(); if (!entry.getValue().isEmpty()) { final String firstValue = entry.getValue().get(0); httpRetrieveResponse.addHeader(header, firstValue); } } if (httpRetrieveResponse.getHttpResponseCode() >= 400) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("A HTTP error received (code >= 400), the download did not initiate."); return STATE.ABORT; } if (task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { final String existingContentLength = fetchContentLengthHeader(task.getHeaders()); final String downloadContentLength = downloadResponseHeaders.getHeaders().getFirstValue("Content-Length"); if (existingContentLength != null && downloadContentLength != null && existingContentLength.trim().equalsIgnoreCase(downloadContentLength.trim())) { httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setLog("Same file already downloaded before, skipping download (" + existingContentLength + "==" + downloadContentLength + ")"); return STATE.ABORT; } } timeWindowCounter.start(); return STATE.CONTINUE; } @Override public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + "ms)"); return STATE.ABORT; } if ((timeWindowCounter.previousTimeWindowRate() != -1) && (timeWindowCounter.previousTimeWindowRate() < task.getLimits().getRetrievalTerminationThresholdReadPerSecondInBytes())) { httpRetrieveResponse.setState(RetrievingState.FINISHED_RATE_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as the speed in which we could download this file was too low (" + timeWindowCounter.currentTimeWindowRate() + "bytes) for a period longer than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms"); return STATE.ABORT; } timeWindowCounter.incrementCount(bodyPart.length()); httpRetrieveResponse.addContent(bodyPart.getBodyPartBytes()); return STATE.CONTINUE; } @Override public Integer onCompleted() throws Exception { if (httpRetrieveResponse.getState() == RetrievingState.PROCESSING) httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setRetrievalDurationInMilliSecs(System.currentTimeMillis() - connectionSetupStartTimestamp); try { httpRetrieveResponse.close(); } catch (IOException e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Failed to close the download response. This might be a bug in harvester slave code.", e1); } return 0; } @Override public void onThrowable(Throwable e) { try { final File retrievalFileStorage = Paths.get(httpRetrieveResponse.getAbsolutePath()).toFile(); if (retrievalFileStorage.exists()) { retrievalFileStorage.delete(); } } catch (Exception e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Cleaning of the incomplete download result file failed in an unexpected way. This might be a bug in harvester slave code.", e1); } httpRetrieveResponse.setState(RetrievingState.ERROR); final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + " ms)"); } if (httpRetrieveResponse.getState() == RetrievingState.COMPLETED && task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); } else { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } } }); try { Integer r = downloadListener.get(1, TimeUnit.DAYS ); LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Download finished with status {}", r); } catch (Exception e) { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } finally { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); asyncHttpClient.close(); return httpRetrieveResponse; } }  HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task); }
SlaveDownloader { public HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task) { if ((task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) && (task.getDocumentReferenceTask().getTaskType() != DocumentReferenceTaskType.UNCONDITIONAL_DOWNLOAD)) { throw new IllegalArgumentException("The downloader can handle only condition & unconditional downloads. Cannot handle " + task.getDocumentReferenceTask().getTaskType()); } if (task.getUrl() == null || "".equalsIgnoreCase(task.getUrl())) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Retrieval of the media failed, URL is empty or null."); httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("Retrieval of the media failed, URL is empty or null."); return httpRetrieveResponse; } try { new URL(task.getUrl()); } catch (Exception e) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("An error has occurred:" + e.getMessage()); httpRetrieveResponse.setException(e); return httpRetrieveResponse; } final AsyncHttpClient asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder() .setMaxRedirects(task.getLimits().getRetrievalMaxNrOfRedirects()) .setFollowRedirect(true) .setConnectTimeout(100000) .setAcceptAnyCertificate(true) .setMaxRequestRetry(3) .build()); httpRetrieveResponse.setState(RetrievingState.PROCESSING); final long connectionSetupStartTimestamp = System.currentTimeMillis(); final ListenableFuture<Integer> downloadListener = asyncHttpClient.prepareGet(task.getUrl()).execute(new AsyncHandler<Integer>() { final TimeWindowCounter timeWindowCounter = new TimeWindowCounter(); @Override public STATE onStatusReceived(HttpResponseStatus status) throws Exception { final long connectionSetupDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setSocketConnectToDownloadStartDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setCheckingDurationInMilliSecs(connectionSetupDurationInMillis); httpRetrieveResponse.setUrl(new URL(task.getUrl())); httpRetrieveResponse.setSourceIp(Slave.URL_RESOLVER.resolveIpOfUrl(task.getUrl())); if (connectionSetupDurationInMillis > task.getLimits().getRetrievalConnectionTimeoutInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to initiate the connection (" + connectionSetupDurationInMillis + "ms, larger than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms)"); return STATE.ABORT; } httpRetrieveResponse.setHttpResponseCode(status.getStatusCode()); return STATE.CONTINUE; } @Override public STATE onHeadersReceived(HttpResponseHeaders downloadResponseHeaders) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); for (final Map.Entry<String, List<String>> entry : downloadResponseHeaders.getHeaders()) { final String header = entry.getKey(); if (!entry.getValue().isEmpty()) { final String firstValue = entry.getValue().get(0); httpRetrieveResponse.addHeader(header, firstValue); } } if (httpRetrieveResponse.getHttpResponseCode() >= 400) { httpRetrieveResponse.setState(RetrievingState.ERROR); httpRetrieveResponse.setLog("A HTTP error received (code >= 400), the download did not initiate."); return STATE.ABORT; } if (task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { final String existingContentLength = fetchContentLengthHeader(task.getHeaders()); final String downloadContentLength = downloadResponseHeaders.getHeaders().getFirstValue("Content-Length"); if (existingContentLength != null && downloadContentLength != null && existingContentLength.trim().equalsIgnoreCase(downloadContentLength.trim())) { httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setLog("Same file already downloaded before, skipping download (" + existingContentLength + "==" + downloadContentLength + ")"); return STATE.ABORT; } } timeWindowCounter.start(); return STATE.CONTINUE; } @Override public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception { final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; httpRetrieveResponse.setRetrievalDurationInMilliSecs(downloadDurationInMillis); if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + "ms)"); return STATE.ABORT; } if ((timeWindowCounter.previousTimeWindowRate() != -1) && (timeWindowCounter.previousTimeWindowRate() < task.getLimits().getRetrievalTerminationThresholdReadPerSecondInBytes())) { httpRetrieveResponse.setState(RetrievingState.FINISHED_RATE_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as the speed in which we could download this file was too low (" + timeWindowCounter.currentTimeWindowRate() + "bytes) for a period longer than " + task.getLimits().getRetrievalConnectionTimeoutInMillis() + "ms"); return STATE.ABORT; } timeWindowCounter.incrementCount(bodyPart.length()); httpRetrieveResponse.addContent(bodyPart.getBodyPartBytes()); return STATE.CONTINUE; } @Override public Integer onCompleted() throws Exception { if (httpRetrieveResponse.getState() == RetrievingState.PROCESSING) httpRetrieveResponse.setState(RetrievingState.COMPLETED); httpRetrieveResponse.setRetrievalDurationInMilliSecs(System.currentTimeMillis() - connectionSetupStartTimestamp); try { httpRetrieveResponse.close(); } catch (IOException e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Failed to close the download response. This might be a bug in harvester slave code.", e1); } return 0; } @Override public void onThrowable(Throwable e) { try { final File retrievalFileStorage = Paths.get(httpRetrieveResponse.getAbsolutePath()).toFile(); if (retrievalFileStorage.exists()) { retrievalFileStorage.delete(); } } catch (Exception e1) { LOG.error(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Cleaning of the incomplete download result file failed in an unexpected way. This might be a bug in harvester slave code.", e1); } httpRetrieveResponse.setState(RetrievingState.ERROR); final long downloadDurationInMillis = System.currentTimeMillis() - connectionSetupStartTimestamp; if (downloadDurationInMillis > task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis()) { httpRetrieveResponse.setState(RetrievingState.FINISHED_TIME_LIMIT); httpRetrieveResponse.setLog("The download was aborted, as it took too long to download this file (" + downloadDurationInMillis + "ms longer than " + task.getLimits().getRetrievalTerminationThresholdTimeLimitInMillis() + " ms)"); } if (httpRetrieveResponse.getState() == RetrievingState.COMPLETED && task.getDocumentReferenceTask().getTaskType() == DocumentReferenceTaskType.CONDITIONAL_DOWNLOAD) { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); } else { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } } }); try { Integer r = downloadListener.get(1, TimeUnit.DAYS ); LOG.debug(LoggingComponent.appendAppFields(LoggingComponent.Slave.SLAVE_RETRIEVAL, task.getJobId(), task.getUrl(), task.getReferenceOwner()), "Download finished with status {}", r); } catch (Exception e) { cleanup(httpRetrieveResponse, task, asyncHttpClient, e); } finally { cleanup(httpRetrieveResponse, task, asyncHttpClient, httpRetrieveResponse.getException()); asyncHttpClient.close(); return httpRetrieveResponse; } }  HttpRetrieveResponse downloadAndStoreInHttpRetrieveResponse(final HttpRetrieveResponse httpRetrieveResponse, final RetrieveUrl task);  }