/* * To change this license header, choose License Headers in Project Properties. * To change this template file, choose Tools | Templates * and open the template in the editor. */ package eu.dnetlib.espas.sos.client; import eu.dnetlib.espas.pep.AuthenticationPEP; import eu.dnetlib.espas.pep.PEPResponseMap; import eu.dnetlib.espas.sos.client.SOSRequestStatus.RequestStatus; import eu.dnetlib.espas.sos.client.jaxb.DataProviderType; import eu.dnetlib.espas.sos.client.utils.QuotaMonitor; import eu.dnetlib.espas.sos.client.utils.QuotaMonitor.RequestQuotaStatus; import eu.dnetlib.espas.sos.client.utils.RequestQuotaException; import eu.dnetlib.espas.sos.client.utils.SOSRequestStatusListenerIF; import eu.dnetlib.espas.sos.client.utils.SOSResultParser; import eu.dnetlib.espas.sos.client.utils.SOSResultSWEGenerator; import java.io.BufferedInputStream; import java.io.ByteArrayInputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.StringWriter; import java.net.HttpURLConnection; import java.net.URISyntaxException; import java.net.URL; import java.net.URLConnection; import java.util.LinkedList; import java.util.List; import java.util.ListIterator; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.CountDownLatch; import java.util.logging.Level; import org.apache.commons.io.IOUtils; import org.apache.http.client.utils.URIBuilder; import org.apache.log4j.Logger; /** * * @author gathanas */ public class SOSProviderHandler implements Runnable { private static final Logger _logger = Logger.getLogger(SOSProviderHandler.class); private static final String[] SOS_REQ_SERVICE_KVP = new String[]{"service", "SOS"}; private static final String[] SOS_REQ_VERSION_KVP = new String[]{"version", "2.0.0"}; private static final String[] SOS_REQ_GET_RESULT_REQUEST_KVP = new String[]{"request", "GetResult"}; private static final String[] SOS_REQ_GET_RESULT_TEMP_REQUEST_KVP = new String[]{"request", "GetResultTemplate"}; private final SOSProviderRequestIF providerRequestInfo; private final URL providerSOSEndpoint; private SOSRequestStatusListenerIF statusListener = null; private AuthenticationPEP authenticationPEP; private DataProviderType providerNode; private QuotaMonitor requestQuotaMonitor; private CountDownLatch providerLatch; private boolean processUnderQuotaFailure; public SOSProviderHandler(SOSRequestInfo providerRequestInfo, URL providerSOSEndpoint, AuthenticationPEP authenticationPEP) { this.providerRequestInfo = providerRequestInfo; this.providerSOSEndpoint = providerSOSEndpoint; this.authenticationPEP = authenticationPEP; } @Override public void run() { try { if (isProviderUp()) processRequest(); else if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.FAILED, "Provider is not responding to requests. " + "The provider service may be down or a network problem has occured."); statusListener.reportSOSRequestProviderStatus(providerRequestInfo.getProviderId(), reqStatus); } } catch (RequestQuotaException ex) { if (!processUnderQuotaFailure) { SOSRequestStatus requestStatus = new SOSRequestStatus(this.providerRequestInfo.getRequestId(), RequestStatus.FAILED, "The assigned quota for serving the given data download request has been surpassed. This download request will be dropped as a result."); statusListener.reportSOSRequestStatus(requestStatus); try { requestQuotaMonitor.cleanupTempRequestSpace(providerRequestInfo.getRequestId()); } catch (IOException ex1) { _logger.warn("Failed to cleanup temporary serving folder for request " + providerRequestInfo.getRequestId(), ex1); } } } catch (Exception ex) { _logger.error("Unexpected error occured while processing provider :" + providerRequestInfo.getProviderId() + " sos service responses", ex); if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.FAILED, "Unexpected exception occurred while processing provider request"); statusListener.reportSOSRequestProviderStatus(providerRequestInfo.getProviderId(), reqStatus); } } finally { if (providerLatch != null) providerLatch.countDown(); } } public void setRequestQuotaMonitor(QuotaMonitor requestQuotaMonitor) { this.requestQuotaMonitor = requestQuotaMonitor; } public void setLatch(CountDownLatch providerSyncLatch) { this.providerLatch = providerSyncLatch; } public void setStatusListener(SOSRequestStatusListenerIF statusListener) { this.statusListener = statusListener; } public DataProviderType getProviderNode() { return this.providerNode; } public void setProcessUnderQuotaFailure(boolean processUnderQuotaFailure) { this.processUnderQuotaFailure = processUnderQuotaFailure; } ////////////////////////////////// // /** * Test the connection to the provider in order to see if the service is up and running. */ private boolean isProviderUp() { return true; } private void processRequest() throws RequestQuotaException { if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.RUNNING, ""); statusListener.reportSOSRequestProviderStatus(providerRequestInfo.getProviderId(), reqStatus); } int failedRequestsCounter = 0; Set performedSOSRequests = new TreeSet(); int totalTemporalEncodings = 0; for (String observationId : providerRequestInfo.getObservationIDList()) { List temporalEncodingList = this.providerRequestInfo.getTemporalFilterEncodings(observationId); totalTemporalEncodings += temporalEncodingList.size(); ListIterator temporalConsIterator = temporalEncodingList.listIterator(); while (temporalConsIterator.hasNext()) { String datetimerange = temporalConsIterator.next(); for (String propertyId : providerRequestInfo.getPropertyIDList()) for (String _offering : providerRequestInfo.getOfferingFor(observationId, propertyId)) if (this.policyEnabled(observationId, propertyId, datetimerange) && !performedSOSRequests.contains(/*observationId + "@@" +*/_offering + "@@" + propertyId + "@@" + datetimerange)) try { if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.PENDING, ""); statusListener.reportSOSObsRequestStatus(_offering, propertyId, datetimerange, providerRequestInfo.getProviderId(), reqStatus); } URLConnection providerConnection = this.createGetResultTemplateKVPHTTPRequest(_offering, propertyId); SOSResultParser parser = this.processGetResultTemplateRespone(providerConnection); // if this is a supported combination i.e. observedProperty - offering, then we will have a template and a parser as a result if (parser != null) { // report reuquest processing start if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.RUNNING, "Template Retrieved and parser has been instantiated!"); statusListener.reportSOSObsRequestStatus(_offering, propertyId, datetimerange, providerRequestInfo.getProviderId(), reqStatus); } // temporal filter is an optional field in the getResult operation. providerConnection = this.createGetResultKVPHTTPRequest(_offering, propertyId, datetimerange); processGetResultResponseStream(_offering, propertyId, datetimerange, providerConnection, parser); // report the completion of the specific request if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.COMPLETED, "Result retrieved and parsed successfully"); statusListener.reportSOSObsRequestStatus(_offering, propertyId, datetimerange, providerRequestInfo.getProviderId(), reqStatus); } } performedSOSRequests.add((/*observationId + "@@" +*/_offering + "@@" + propertyId + "@@" + datetimerange)); } catch (RequestQuotaException ex) { if (ex.getProblem() == RequestQuotaStatus.RequestQuotaOverflow) { setRequestStatus(RequestStatus.FAILED, "The assigned quota for this data download request has been reached. Related requests will be dropped as a result", performedSOSRequests); throw ex; } else { setRequestStatus(RequestStatus.PENDING, "The quota assinged for serving data download requests by the server has been reached. Waiting for free space to emerge so as to continue serving this request.", performedSOSRequests); synchronized (QuotaMonitor.pendingThreadsLock) { try { QuotaMonitor.pendingThreadsLock.wait(); } catch (InterruptedException ex1) { _logger.warn("Interrupt exception raised while waiting for free space to emerge", ex1); } } // set the temporal constraint to the previous position and re-run the processing step for this request temporalConsIterator.previous(); } } catch (Exception ex) { _logger.error("Exception while processing request to " + this.providerRequestInfo.getProviderId() + " for observation id " + observationId + " and property :" + propertyId, ex); // System.out.println("\n\n"+ex.getLocalizedMessage()); // request failed due to exception; increase counter so as to check the total provider status failedRequestsCounter++; if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.FAILED, ex.getMessage()); statusListener.reportSOSObsRequestStatus(_offering, propertyId, datetimerange, providerRequestInfo.getProviderId(), reqStatus); // set this offering # property combination in the list of performed requests performedSOSRequests.add((observationId + "@@" + _offering + "@@" + propertyId + "@@" + datetimerange)); continue; } } else // request failed due to policy constraints failedRequestsCounter++; } } SOSRequestStatus reqStatus; if (statusListener != null && failedRequestsCounter < (providerRequestInfo.getObservationIDList().size() * providerRequestInfo.getPropertyIDList().size()) * totalTemporalEncodings) reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.COMPLETED, ""); else reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.FAILED, "All requests have been rejected by provider :" + providerRequestInfo.getProviderId()); statusListener.reportSOSRequestProviderStatus(providerRequestInfo.getProviderId(), reqStatus); } /** * Creates a URL encoded GetResultTemplate HTTP request for the specified service provider.The created HTTP GET request conforms to the guidelines specified * in the OGC SOS Service Interface specification v2.0 for the KVP mapping. */ private URLConnection createGetResultTemplateKVPHTTPRequest(String offeringId, String observedProperty) throws IOException, URISyntaxException { URIBuilder uRIBuilder = new URIBuilder(this.providerSOSEndpoint.toURI()); if (uRIBuilder != null) { uRIBuilder = uRIBuilder.addParameter(SOS_REQ_SERVICE_KVP[0], SOS_REQ_SERVICE_KVP[1]); uRIBuilder = uRIBuilder.addParameter(SOS_REQ_VERSION_KVP[0], SOS_REQ_VERSION_KVP[1]); uRIBuilder = uRIBuilder.addParameter(SOS_REQ_GET_RESULT_TEMP_REQUEST_KVP[0], SOS_REQ_GET_RESULT_TEMP_REQUEST_KVP[1]); uRIBuilder = uRIBuilder.addParameter("offering", offeringId); uRIBuilder = uRIBuilder.addParameter("observedProperty", observedProperty); } URL sosReqUrl = uRIBuilder.build().toURL(); _logger.info("Request url encoding is:" + sosReqUrl.toString()); URLConnection connection = sosReqUrl.openConnection(); if (connection instanceof HttpURLConnection) ((HttpURLConnection) connection).setRequestProperty("Content-Type", "text/xml"); return connection; } private SOSResultParser processGetResultTemplateRespone(URLConnection connection) throws IOException { SOSResultParser parser = null; connection.connect(); InputStream resultTemplateStream = new BufferedInputStream(connection.getInputStream()); if (resultTemplateStream != null) { StringWriter templateWritter = new StringWriter(); IOUtils.copy(resultTemplateStream, templateWritter); try { _logger.debug("Content from stream :" + templateWritter.getBuffer().toString()); parser = new SOSResultParser(new ByteArrayInputStream(templateWritter.getBuffer().toString().getBytes())); } catch (Exception ex) { _logger.error("exception while trying to retrieve result template or result parser.", ex); } } return parser; } /** * Creates a URL encoded GetResult Http request for the given service provider based on the provided attributes. The created HTTP GET request conforms to * the guidelines specified in the OGC SOS Service Interface specification v2.0 for the KVP mapping. */ private URLConnection createGetResultKVPHTTPRequest(String offeringId, String observedProperty, String temporalFilterEncoding) throws IOException, URISyntaxException { URIBuilder uRIBuilder = new URIBuilder(this.providerSOSEndpoint.toURI()); if (uRIBuilder != null) { uRIBuilder = uRIBuilder.addParameter(SOS_REQ_SERVICE_KVP[0], SOS_REQ_SERVICE_KVP[1]); uRIBuilder = uRIBuilder.addParameter(SOS_REQ_VERSION_KVP[0], SOS_REQ_VERSION_KVP[1]); uRIBuilder = uRIBuilder.addParameter(SOS_REQ_GET_RESULT_REQUEST_KVP[0], SOS_REQ_GET_RESULT_REQUEST_KVP[1]); uRIBuilder = uRIBuilder.addParameter("offering", offeringId); uRIBuilder = uRIBuilder.addParameter("observedProperty", observedProperty); uRIBuilder = uRIBuilder.addParameter("namespaces", providerRequestInfo.getRequestFilterNamespaces()); if (temporalFilterEncoding != null && !temporalFilterEncoding.isEmpty()) uRIBuilder = uRIBuilder.addParameter("temporalFilter", temporalFilterEncoding); } URL sosReqUrl = uRIBuilder.build().toURL(); _logger.info("GETResult request url encoding is:" + sosReqUrl.toString()); URLConnection connection = sosReqUrl.openConnection(); if (connection instanceof HttpURLConnection) ((HttpURLConnection) connection).setRequestProperty("Content-Type", "application/xml"); return connection; } /** * Is responsible for opening the Http connection to the provider's endpoint and processing the retrieved data stream. */ private void processGetResultResponseStream(String offeringId, String propertyId, String datetimerange, URLConnection connection, SOSResultParser parser) throws IOException, RequestQuotaException { //@todo add implementation code here connection.connect(); InputStream getResultStream = connection.getInputStream(); // if stream is ok copy the returned results to a temp file and start processing if (getResultStream != null) { _logger.info("Current meassurements: " + ((providerNode != null && providerNode.getMeasurements() != null) ? providerNode.getMeasurements().size() : 0) + ". Processing content stream for observation id :" + offeringId + " " + "and property :" + propertyId); // create the temp file for storing the getResult response message File resultStoreFile = getResultResponseFile(offeringId, propertyId, datetimerange); QuotaMonitor.RequestSpaceReport requestSpaceReport = requestQuotaMonitor.requestFreeSpace(this.providerRequestInfo.getRequestId()); if (requestSpaceReport.getSpaceStatus() == RequestQuotaStatus.RequestQuotaUnderflow) { long usedSpace = IOUtils.copyLarge(getResultStream, new FileOutputStream(resultStoreFile), 0, requestSpaceReport.getFreeSpace()); boolean quotaUpdated = requestQuotaMonitor.consumedRequestQuota(this.providerRequestInfo.getRequestId(), usedSpace); if (!quotaUpdated) throw new RequestQuotaException(this.providerRequestInfo.getRequestId(), RequestQuotaStatus.RequestQuotaOverflow, offeringId + " @@ " + propertyId + " @@ " + datetimerange, "Request volume quota limit has been reached. This request will be dropped!"); } else throw new RequestQuotaException(this.providerRequestInfo.getRequestId(), RequestQuotaStatus.ProviderQuotaOverflow, offeringId + " @@ " + propertyId + " @@ " + datetimerange, ""); if (getResultStream.available() > 0) throw new RequestQuotaException(this.providerRequestInfo.getRequestId(), RequestQuotaStatus.RequestQuotaOverflow, offeringId + " @@ " + propertyId + " @@ " + datetimerange, "Request volume quota limit has been reached. This request will be dropped!"); // @todo: report result retrieval to listener if (statusListener != null) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.PENDING, "Connection to getResult openned. Waiting for ANTLR processing!"); statusListener.reportSOSObsRequestStatus(offeringId, propertyId, datetimerange, providerRequestInfo.getProviderId(), reqStatus); } parser.setRootNode(providerNode); parser.setProviderId(providerRequestInfo.getProviderId()); parser.setObservedProperty(propertyId); parser.setOffering(offeringId); providerNode = parser.process((InputStream) new FileInputStream(resultStoreFile)); // transform the collected SOS result data and template to swe format SOSResultSWEGenerator sweGenerator = new SOSResultSWEGenerator(parser, (InputStream) new FileInputStream(resultStoreFile)); sweGenerator.printSweResult(new FileOutputStream(requestQuotaMonitor.getDmSpaceUtils().getRequestResultSWEArchiveFile(providerRequestInfo.getRequestId(), providerRequestInfo.getProviderId(), offeringId, propertyId, datetimerange))); } } private File getResultResponseFile(String observationId, String propertyId, String datetimerange) throws IOException { File tempFile = requestQuotaMonitor.getDmSpaceUtils().getTempRequestArchiveFile(providerRequestInfo.getRequestId(), providerRequestInfo.getProviderId(), observationId, propertyId, datetimerange); return tempFile; } private boolean policyEnabled(String observationId, String propertyId, String datetimerange) { boolean permitedFlag = false; String policyMessage = ""; if (authenticationPEP != null) try { List observationList = new LinkedList(); observationList.add(observationId); PEPResponseMap pepResponse = authenticationPEP.isPermitedRequest(observationList, this.providerRequestInfo.getUserId(), "access", null); permitedFlag = pepResponse.isResourcePermited(observationId); policyMessage = pepResponse.policyResponseMsg(observationId); } catch (Exception ex) { SOSRequestManager._logger.error("Exception while trying to query request policy constraints", ex); } if (!permitedFlag) for (String _offering : providerRequestInfo.getOfferingFor(observationId, propertyId)) this.statusListener.reportSOSObsRequestStatus(_offering, propertyId, datetimerange, this.providerRequestInfo.getProviderId(), new SOSRequestStatus(providerRequestInfo.getRequestId(), RequestStatus.FAILED, "Request not permited due to policy constraints. Policy advice :" + policyMessage)); return permitedFlag; } private void setRequestStatus(RequestStatus requestStatus, String message, Set performedSOSRequests) { SOSRequestStatus reqStatus = new SOSRequestStatus(providerRequestInfo.getRequestId(), requestStatus, message); // set the provider status statusListener.reportSOSRequestProviderStatus(this.providerRequestInfo.getProviderId(), reqStatus); // set the status of each observation request that has not been served up to now for (String observationId : providerRequestInfo.getObservationIDList()) for (String propertyId : this.providerRequestInfo.getPropertyIDList()) for (String datetimerange : providerRequestInfo.getTemporalFilterEncodings(observationId)) if (performedSOSRequests.contains(/*observationId + "@@" +*/providerRequestInfo.getOfferingFor(observationId, propertyId) + "@@" + propertyId + "@@" + datetimerange)) statusListener.reportSOSObsRequestStatus(observationId, propertyId, datetimerange, propertyId, reqStatus); } }