Merge pull request #5 from mswiderski/sync-monitor

added monitor capabilities to support configuration update in clustered ...

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/config/Added.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012 JBoss Inc
+ *
+ * Licensed 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.uberfire.backend.server.config;
+
+import static java.lang.annotation.ElementType.*;
+import static java.lang.annotation.RetentionPolicy.*;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import javax.inject.Qualifier;
+
+@Qualifier
+@Target({PARAMETER, FIELD})
+@Retention(RUNTIME)
+public @interface Added {
+
+}

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/config/ConfigurationServiceImpl.java

@@ -3,13 +3,18 @@
 import java.io.OutputStream;
 import java.net.URI;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import javax.enterprise.context.ApplicationScoped;
 import javax.enterprise.context.ContextNotActiveException;
+import javax.enterprise.event.Event;
 import javax.inject.Inject;
 import javax.inject.Named;
 
@@ -27,6 +32,11 @@
 
 @ApplicationScoped
 public class ConfigurationServiceImpl implements ConfigurationService {
+    private static final String LAST_MODIFIED_MARKER_FILE = ".lastmodified";
+    private static final String MONITOR_DISABLED = "org.kie.sys.repo.monitor.disabled";
+    private static final String MONITOR_CHECK_INTERVAL = "org.kie.sys.repo.monitor.interval";
+    // mainly for windows as *NIX is based on POSIX but escape always to keep it consistent
+    private static final String INVALID_FILENAME_CHARS = "[\\,/,:,*,?,\",<,>,|]";
 
     @Inject
     @Named("system")
@@ -38,21 +48,41 @@ public class ConfigurationServiceImpl implements ConfigurationService {
     @Inject
     private Identity identity;
 
+
     //Cache of ConfigGroups to avoid reloading them from file
-    private final Map<ConfigType, List<ConfigGroup>> configuration = new HashMap<ConfigType, List<ConfigGroup>>();
+    private final Map<ConfigType, List<ConfigGroup>> configuration = new ConcurrentHashMap<ConfigType, List<ConfigGroup>>();
+    private long localLastModifiedValue = -1;
 
     @Inject
     @Named("ioStrategy")
     private IOService ioService;
 
+    // monitor capabilities
+    @Inject
+    private Event<SystemRepositoryChangedEvent> changedEvent;
+    private ExecutorService executorService;
+
     @PostConstruct
     public void setup() {
         try {
             ioService.newFileSystem( URI.create( systemRepository.getUri() ),
                                      systemRepository.getEnvironment(),
                                      FileSystemType.Bootstrap.BOOTSTRAP_INSTANCE );
+            updateLastModified();
         } catch ( FileSystemAlreadyExistsException e ) {
         }
+        // enable monitor by default
+        if (System.getProperty(MONITOR_DISABLED) == null) {
+            executorService = Executors.newSingleThreadExecutor();
+            executorService.execute(new CheckConfigurationUpdates());
+        }
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        if (this.executorService != null) {
+            this.executorService.shutdownNow();
+        }
     }
 
     @Override
@@ -88,7 +118,14 @@ public boolean accept( final Path entry ) throws IOException {
     @Override
     public boolean addConfiguration( final ConfigGroup configGroup ) {
         try {
-            final Path filePath = ioService.get( systemRepository.getUri() ).resolve( configGroup.getName() + configGroup.getType().getExt() );
+            String filename = configGroup.getName().replaceAll(INVALID_FILENAME_CHARS, "_");
+
+            final Path filePath = ioService.get( systemRepository.getUri() ).resolve( filename + configGroup.getType().getExt() );
+            // avoid duplicated writes to not cause cyclic cluster sync
+            if (ioService.exists(filePath)) {
+                return true;
+            }
+
             final CommentedOption commentedOption = new CommentedOption( getIdentityName(),
                                                                          "Created config " + filePath.getFileName() );
             final OutputStream outputStream = ioService.newOutputStream( filePath,
@@ -101,6 +138,7 @@ public boolean addConfiguration( final ConfigGroup configGroup ) {
 
             //Invalidate cache if a new item has been created; otherwise cached value is stale
             configuration.remove( configGroup.getType() );
+            updateLastModified();
 
             return true;
 
@@ -111,10 +149,23 @@ public boolean addConfiguration( final ConfigGroup configGroup ) {
 
     @Override
     public boolean removeConfiguration( final ConfigGroup configGroup ) {
+
         //Invalidate cache if an item has been removed; otherwise cached value is stale
         configuration.remove( configGroup.getType() );
+        String filename = configGroup.getName().replaceAll(INVALID_FILENAME_CHARS, "_");
+        final Path filePath = ioService.get( systemRepository.getUri() ).resolve( filename + configGroup.getType().getExt() );
+
+        // avoid duplicated writes to not cause cyclic cluster sync
+        if (!ioService.exists(filePath)) {
+            return true;
+        }
+        boolean result = ioService.deleteIfExists( filePath );
+
+        if (result) {
+            updateLastModified();
+        }
 
-        return ioService.deleteIfExists( ioService.get( systemRepository.getUri() ).resolve( configGroup.getName() + configGroup.getType().getExt() ) );
+        return result;
     }
 
     protected String getIdentityName() {
@@ -124,4 +175,58 @@ protected String getIdentityName() {
             return "unknown";
         }
     }
+
+    protected long getLastModified() {
+        final Path lastModifiedPath = ioService.get( systemRepository.getUri() ).resolve(LAST_MODIFIED_MARKER_FILE);
+
+        return ioService.getLastModifiedTime(lastModifiedPath).toMillis();
+    }
+
+    protected void updateLastModified() {
+        try {
+            final Path lastModifiedPath = ioService.get( systemRepository.getUri() ).resolve(LAST_MODIFIED_MARKER_FILE);
+            final CommentedOption commentedOption = new CommentedOption( "system", "system repo updated" );
+            final OutputStream outputStream = ioService.newOutputStream(lastModifiedPath,
+                    StandardOpenOption.TRUNCATE_EXISTING,
+                    commentedOption );
+
+            outputStream.write(new Date().toString().getBytes());
+            outputStream.close();
+            // update the last value to avoid to be retriggered byt the monitor
+            localLastModifiedValue = getLastModified();
+
+        } catch ( java.io.IOException e ) {
+            throw new RuntimeException( "Error when updating system repository", e );
+        }
+    }
+
+    private class CheckConfigurationUpdates implements Runnable {
+
+        private boolean active = true;
+
+        @Override
+        public void run() {
+
+            while (active) {
+                try {
+                    long currentValue = getLastModified();
+                    if (currentValue > localLastModifiedValue) {
+                        localLastModifiedValue = currentValue;
+                        // invalidate cached values as system repo has changed - for now only for deployments
+                        configuration.remove(ConfigType.DEPLOYMENT);
+                        changedEvent.fire(new SystemRepositoryChangedEvent());
+
+                    }
+
+                    Thread.sleep(Long.parseLong(System.getProperty(MONITOR_CHECK_INTERVAL, "2000")));
+                } catch (Exception e) {
+
+                }
+            }
+        }
+
+        public void deactivate() {
+            this.active = false;
+        }
+    }
 }

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/config/Removed.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2012 JBoss Inc
+ *
+ * Licensed 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.uberfire.backend.server.config;
+
+import static java.lang.annotation.ElementType.*;
+import static java.lang.annotation.RetentionPolicy.*;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import javax.inject.Qualifier;
+
+@Qualifier
+@Target({PARAMETER, FIELD})
+@Retention(RUNTIME)
+public @interface Removed {
+
+}

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/config/SystemRepositoryChangedEvent.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012 JBoss Inc
+ *
+ * Licensed 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.uberfire.backend.server.config;
+
+public class SystemRepositoryChangedEvent {
+
+}

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/deployment/DeploymentConfigChangedEvent.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2012 JBoss Inc
+ *
+ * Licensed 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.uberfire.backend.server.deployment;
+
+public class DeploymentConfigChangedEvent {
+
+    public Object getDeploymentUnit() {
+        return deploymentUnit;
+    }
+
+    public void setDeploymentUnit(Object deploymentUnit) {
+        this.deploymentUnit = deploymentUnit;
+    }
+
+    private Object deploymentUnit;
+
+    public DeploymentConfigChangedEvent() {
+
+    }
+
+    public DeploymentConfigChangedEvent(Object deploymentUnit) {
+        this.deploymentUnit = deploymentUnit;
+    }
+}

uberfire-backend/uberfire-backend-server/src/main/java/org/uberfire/backend/server/deployment/DeploymentConfigServiceImpl.java

@@ -1,20 +1,27 @@
 package org.uberfire.backend.server.deployment;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
-
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import javax.annotation.PostConstruct;
 import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.event.Event;
+import javax.enterprise.event.Observes;
 import javax.inject.Inject;
 
 import org.uberfire.backend.deployment.DeploymentConfig;
 import org.uberfire.backend.deployment.DeploymentConfigService;
+import org.uberfire.backend.server.config.Added;
 import org.uberfire.backend.server.config.ConfigGroup;
 import org.uberfire.backend.server.config.ConfigType;
 import org.uberfire.backend.server.config.ConfigurationFactory;
 import org.uberfire.backend.server.config.ConfigurationService;
+import org.uberfire.backend.server.config.Removed;
+import org.uberfire.backend.server.config.SystemRepositoryChangedEvent;
 
 @ApplicationScoped
 public class DeploymentConfigServiceImpl implements DeploymentConfigService {
@@ -27,8 +34,15 @@ public class DeploymentConfigServiceImpl implements DeploymentConfigService {
 
     @Inject
     private DeploymentConfigFactory deploymentFactory;
+    @Inject
+    @Added
+    private Event<DeploymentConfigChangedEvent> addedDeploymentEvent;
+    @Inject
+    @Removed
+    private Event<DeploymentConfigChangedEvent> removedDeploymentEvent;
 
-    private Map<String, DeploymentConfig> registeredDeployments = new HashMap<String, DeploymentConfig>();
+
+    private Map<String, DeploymentConfig> registeredDeployments = new ConcurrentHashMap<String, DeploymentConfig>();
 
     @PostConstruct
     public void loadGroups() {
@@ -70,4 +84,36 @@ public DeploymentConfig getDeployment(String identifier) {
     public Collection<DeploymentConfig> getDeployments() {
         return Collections.unmodifiableCollection(registeredDeployments.values());
     }
+
+    public void updateRegisteredDeployments(@Observes SystemRepositoryChangedEvent changedEvent) {
+        Collection<ConfigGroup> deployments = configurationService.getConfiguration(ConfigType.DEPLOYMENT);
+        if (deployments != null) {
+            List<String> processedDeployments = new ArrayList<String>();
+            for (ConfigGroup deploymentConfig : deployments) {
+                String name = deploymentConfig.getName();
+
+                if (!this.registeredDeployments.containsKey(name)) {
+                    // add it to registered deployments
+                    DeploymentConfig deployment = deploymentFactory.newDeployment(deploymentConfig);
+                    // trigger deployment of new element
+                    addedDeploymentEvent.fire(new DeploymentConfigChangedEvent(deployment.getDeploymentUnit()));
+                    registeredDeployments.put(deployment.getIdentifier(), deployment);
+                }
+
+                processedDeployments.add(name);
+
+            }
+
+            Set<String> registeredDeploymedIds = registeredDeployments.keySet();
+            // process undeploy
+            for (String identifier : registeredDeploymedIds) {
+                if (!processedDeployments.contains(identifier)) {
+                    DeploymentConfig deployment = registeredDeployments.remove(identifier);
+
+                    // trigger undeloyment as it was removed
+                    removedDeploymentEvent.fire(new DeploymentConfigChangedEvent(deployment.getDeploymentUnit()));
+                }
+            }
+        }
+    }
 }