CAMEL-9517: Validator Endpoint- clearCachedSchema added

camel-core/src/main/java/org/apache/camel/component/validator/ValidatorEndpoint.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.validator;
 
+import java.io.IOException;
 import java.io.InputStream;
 import javax.xml.XMLConstants;
 import javax.xml.validation.SchemaFactory;
@@ -26,9 +27,12 @@
 import org.apache.camel.Consumer;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
+import org.apache.camel.api.management.ManagedOperation;
+import org.apache.camel.api.management.ManagedResource;
 import org.apache.camel.converter.IOConverter;
 import org.apache.camel.impl.DefaultEndpoint;
 import org.apache.camel.processor.validation.DefaultValidationErrorHandler;
+import org.apache.camel.processor.validation.SchemaReader;
 import org.apache.camel.processor.validation.ValidatingProcessor;
 import org.apache.camel.processor.validation.ValidatorErrorHandler;
 import org.apache.camel.spi.Metadata;
@@ -43,6 +47,7 @@
 /**
  * Validates the payload of a message using XML Schema and JAXP Validation.
  */
+@ManagedResource(description = "Managed ValidatorEndpoint")
 @UriEndpoint(scheme = "validator", title = "Validator", syntax = "validator:resourceUri", producerOnly = true, label = "core,validation")
 public class ValidatorEndpoint extends DefaultEndpoint {
 
@@ -73,6 +78,14 @@ public class ValidatorEndpoint extends DefaultEndpoint {
     @UriParam(description = "To validate against a header instead of the message body.")
     private String headerName;
 
+    /**
+     * We need a one-to-one relation between endpoint and schema reader in order
+     * to be able to clear the cached schema in the schema reader. See method
+     * {@link #clearCachedSchema}.
+     */
+    private final SchemaReader schemaReader = new SchemaReader();
+    private volatile boolean schemaReaderConfigured;
+
     public ValidatorEndpoint() {
     }
 
@@ -81,28 +94,58 @@ public ValidatorEndpoint(String endpointUri, Component component, String resourc
         this.resourceUri = resourceUri;
     }
 
+    @ManagedOperation(description = "Clears the cached schema, forcing to re-load the schema on next request")
+    public void clearCachedSchema() throws Exception {
+        LOG.debug("{} rereading schema resource: {}", this, resourceUri);
+        byte[] bytes = readSchemaResource();
+        schemaReader.setSchemaAsByteArray(bytes);
+
+        schemaReader.setSchema(null); // will cause to reload the schema from
+                                      // the set byte-array on next request
+    }
+
     @Override
     public Producer createProducer() throws Exception {
-        ValidatingProcessor validator = new ValidatingProcessor();
 
+        if (!schemaReaderConfigured) {
+            if (resourceResolver != null) {
+                schemaReader.setResourceResolver(resourceResolver);
+            } else {
+                schemaReader.setResourceResolver(new DefaultLSResourceResolver(getCamelContext(), resourceUri));
+            }
+            schemaReader.setSchemaLanguage(getSchemaLanguage());
+            schemaReader.setSchemaFactory(getSchemaFactory());
+
+            byte[] bytes = readSchemaResource();
+            schemaReader.setSchemaAsByteArray(bytes);
+            LOG.debug("{} using schema resource: {}", this, resourceUri);
+
+            // force loading of schema at create time otherwise concurrent
+            // processing could cause thread safe issues for the
+            // javax.xml.validation.SchemaFactory
+            schemaReader.loadSchema();
+
+            // configure only once
+            schemaReaderConfigured = true;
+        }
+
+        ValidatingProcessor validator = new ValidatingProcessor(schemaReader);
+        configureValidator(validator);
+
+        return new ValidatorProducer(this, validator);
+    }
+
+    protected byte[] readSchemaResource() throws IOException {
         InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(getCamelContext(), resourceUri);
         byte[] bytes = null;
         try {
             bytes = IOConverter.toBytes(is);
         } finally {
-            // and make sure to close the input stream after the schema has been loaded
+            // and make sure to close the input stream after the schema has been
+            // loaded
             IOHelper.close(is);
         }
-
-        validator.setSchemaAsByteArray(bytes);
-        LOG.debug("{} using schema resource: {}", this, resourceUri);
-        configureValidator(validator);
-
-        // force loading of schema at create time otherwise concurrent
-        // processing could cause thread safe issues for the javax.xml.validation.SchemaFactory
-        validator.loadSchema();
-
-        return new ValidatorProducer(this, validator);
+        return bytes;
     }
 
     @Override
@@ -116,13 +159,6 @@ public boolean isSingleton() {
     }
 
     protected void configureValidator(ValidatingProcessor validator) throws Exception {
-        if (resourceResolver != null) {
-            validator.setResourceResolver(resourceResolver);
-        } else {
-            validator.setResourceResolver(new DefaultLSResourceResolver(getCamelContext(), resourceUri));
-        }
-        validator.setSchemaLanguage(getSchemaLanguage());
-        validator.setSchemaFactory(getSchemaFactory());
         validator.setErrorHandler(getErrorHandler());
         validator.setUseDom(isUseDom());
         validator.setUseSharedSchema(isUseSharedSchema());

camel-core/src/main/java/org/apache/camel/processor/validation/SchemaReader.java

@@ -0,0 +1,181 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.camel.processor.validation;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+
+import javax.xml.XMLConstants;
+import javax.xml.transform.Source;
+import javax.xml.transform.stream.StreamSource;
+import javax.xml.validation.Schema;
+import javax.xml.validation.SchemaFactory;
+
+import org.w3c.dom.ls.LSResourceResolver;
+import org.xml.sax.SAXException;
+
+/**
+ * Reads the schema used in the processor {@link ValidatingProcessor}. Contains
+ * the method {@link clearCachedSchema()} to force re-reading the schema.
+ */
+public class SchemaReader {
+
+    private String schemaLanguage = XMLConstants.W3C_XML_SCHEMA_NS_URI;
+    // must be volatile because is accessed from different threads see ValidatorEndpoint.clearCachedSchema
+    private volatile Schema schema;
+    private Source schemaSource;
+    // must be volatile because is accessed from different threads see ValidatorEndpoint.clearCachedSchema
+    private volatile SchemaFactory schemaFactory;
+    private URL schemaUrl;
+    private File schemaFile;
+    private volatile byte[] schemaAsByteArray;
+    private LSResourceResolver resourceResolver;
+
+    public void loadSchema() throws Exception {
+        // force loading of schema
+        schema = createSchema();
+    }
+
+    // Properties
+    // -----------------------------------------------------------------------
+
+    public Schema getSchema() throws IOException, SAXException {
+        if (schema == null) {
+            synchronized (this) {
+                if (schema == null) {
+                    schema = createSchema();
+                }
+            }
+        }
+        return schema;
+    }
+
+    public void setSchema(Schema schema) {
+        this.schema = schema;
+    }
+
+    public String getSchemaLanguage() {
+        return schemaLanguage;
+    }
+
+    public void setSchemaLanguage(String schemaLanguage) {
+        this.schemaLanguage = schemaLanguage;
+    }
+
+    public Source getSchemaSource() throws IOException {
+        if (schemaSource == null) {
+            schemaSource = createSchemaSource();
+        }
+        return schemaSource;
+    }
+
+    public void setSchemaSource(Source schemaSource) {
+        this.schemaSource = schemaSource;
+    }
+
+    public URL getSchemaUrl() {
+        return schemaUrl;
+    }
+
+    public void setSchemaUrl(URL schemaUrl) {
+        this.schemaUrl = schemaUrl;
+    }
+
+    public File getSchemaFile() {
+        return schemaFile;
+    }
+
+    public void setSchemaFile(File schemaFile) {
+        this.schemaFile = schemaFile;
+    }
+
+    public byte[] getSchemaAsByteArray() {
+        return schemaAsByteArray;
+    }
+
+    public void setSchemaAsByteArray(byte[] schemaAsByteArray) {
+        this.schemaAsByteArray = schemaAsByteArray;
+    }
+
+    public SchemaFactory getSchemaFactory() {
+        if (schemaFactory == null) {
+            synchronized (this) {
+                if (schemaFactory == null) {
+                    schemaFactory = createSchemaFactory();
+                }
+            }
+        }
+        return schemaFactory;
+    }
+
+    public void setSchemaFactory(SchemaFactory schemaFactory) {
+        this.schemaFactory = schemaFactory;
+    }
+
+    public LSResourceResolver getResourceResolver() {
+        return resourceResolver;
+    }
+
+    public void setResourceResolver(LSResourceResolver resourceResolver) {
+        this.resourceResolver = resourceResolver;
+    }
+
+    protected SchemaFactory createSchemaFactory() {
+        SchemaFactory factory = SchemaFactory.newInstance(schemaLanguage);
+        if (getResourceResolver() != null) {
+            factory.setResourceResolver(getResourceResolver());
+        }
+        return factory;
+    }
+
+    protected Source createSchemaSource() throws IOException {
+        throw new IllegalArgumentException("You must specify either a schema, schemaFile, schemaSource or schemaUrl property");
+    }
+
+    protected Schema createSchema() throws SAXException, IOException {
+        SchemaFactory factory = getSchemaFactory();
+
+        URL url = getSchemaUrl();
+        if (url != null) {
+            synchronized (this) {
+                return factory.newSchema(url);
+            }
+        }
+
+        File file = getSchemaFile();
+        if (file != null) {
+            synchronized (this) {
+                return factory.newSchema(file);
+            }
+        }
+
+        byte[] bytes = getSchemaAsByteArray();
+        if (bytes != null) {
+            synchronized (this) {
+                return factory.newSchema(new StreamSource(new ByteArrayInputStream(schemaAsByteArray)));
+            }
+        }
+
+        Source source = getSchemaSource();
+        synchronized (this) {
+            return factory.newSchema(source);
+        }
+    }
+
+}