config-loader/config-loader-impl/src/main/java/org/opendaylight/bgpcep/config/loader/impl/ConfigLoaderImpl.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.bgpcep.config.loader.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.base.Stopwatch;
  13 import java.io.File;
  14 import java.io.FileInputStream;
  15 import java.io.IOException;
  16 import java.io.InputStream;
  17 import java.io.RandomAccessFile;
  18 import java.net.URISyntaxException;
  19 import java.nio.channels.FileChannel;
  20 import java.nio.channels.FileLock;
  21 import java.nio.file.ClosedWatchServiceException;
  22 import java.nio.file.WatchKey;
  23 import java.nio.file.WatchService;
  24 import java.util.ArrayList;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.concurrent.TimeUnit;
  29 import java.util.regex.Pattern;
  30 import java.util.stream.Collectors;
  31 import javax.xml.stream.XMLInputFactory;
  32 import javax.xml.stream.XMLStreamException;
  33 import javax.xml.stream.XMLStreamReader;
  34 import org.checkerframework.checker.lock.qual.GuardedBy;
  35 import org.opendaylight.bgpcep.config.loader.spi.ConfigFileProcessor;
  36 import org.opendaylight.bgpcep.config.loader.spi.ConfigLoader;
  37 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  38 import org.opendaylight.yangtools.concepts.AbstractRegistration;
  39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  40 import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter;
  41 import org.opendaylight.yangtools.yang.data.codec.xml.XmlParserStream;
  42 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNormalizedNodeStreamWriter;
  43 import org.opendaylight.yangtools.yang.data.impl.schema.NormalizedNodeResult;
  44 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  45 import org.opendaylight.yangtools.yang.model.api.SchemaNode;
  46 import org.opendaylight.yangtools.yang.model.util.SchemaContextUtil;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49 import org.xml.sax.SAXException;
  50
  51 public final class ConfigLoaderImpl implements ConfigLoader, AutoCloseable {
  52     private static final Logger LOG = LoggerFactory.getLogger(ConfigLoaderImpl.class);
  53     private static final String INTERRUPTED = "InterruptedException";
  54     private static final String EXTENSION = "-.*\\.xml";
  55     private static final String INITIAL = "^";
  56     private static final String READ = "rw";
  57     private static final long TIMEOUT_SECONDS = 5;
  58     @GuardedBy("this")
  59     private final Map<String, ConfigFileProcessor> configServices = new HashMap<>();
  60     private final EffectiveModelContext schemaContext;
  61     private final BindingNormalizedNodeSerializer bindingSerializer;
  62     private final String path;
  63     private final Thread watcherThread;
  64     private final File file;
  65     @GuardedBy("this")
  66     private boolean closed = false;
  67
  68     public ConfigLoaderImpl(final EffectiveModelContext schemaContext,
  69             final BindingNormalizedNodeSerializer bindingSerializer, final FileWatcher fileWatcher) {
  70         this.schemaContext = requireNonNull(schemaContext);
  71         this.bindingSerializer = requireNonNull(bindingSerializer);
  72         this.path = requireNonNull(fileWatcher.getPathFile());
  73         this.file = new File(this.path);
  74         this.watcherThread = new Thread(new ConfigLoaderImplRunnable(requireNonNull(fileWatcher.getWatchService())));
  75     }
  76
  77     public void init() {
  78         this.watcherThread.start();
  79         LOG.info("Config Loader service initiated");
  80     }
  81
  82     private synchronized void handleConfigFile(final ConfigFileProcessor config, final String filename) {
  83         final NormalizedNode<?, ?> dto;
  84         try {
  85             dto = parseDefaultConfigFile(config, filename);
  86         } catch (final IOException | XMLStreamException e) {
  87             LOG.warn("Failed to parse config file {}", filename, e);
  88             return;
  89         }
  90         LOG.info("Loading initial config {}", filename);
  91         config.loadConfiguration(dto);
  92     }
  93
  94     private NormalizedNode<?, ?> parseDefaultConfigFile(final ConfigFileProcessor config, final String filename)
  95             throws IOException, XMLStreamException {
  96         final NormalizedNodeResult result = new NormalizedNodeResult();
  97         final NormalizedNodeStreamWriter streamWriter = ImmutableNormalizedNodeStreamWriter.from(result);
  98
  99         final File newFile = new File(this.path, filename);
 100         FileChannel channel = new RandomAccessFile(newFile, READ).getChannel();
 101
 102         FileLock lock = null;
 103         final Stopwatch stopwatch = Stopwatch.createStarted();
 104         while (lock == null || stopwatch.elapsed(TimeUnit.SECONDS) > TIMEOUT_SECONDS) {
 105             try {
 106                 lock = channel.tryLock();
 107             } catch (final IllegalStateException e) {
 108                 //Ignore
 109             }
 110             if (lock == null) {
 111                 try {
 112                     Thread.sleep(100L);
 113                 } catch (InterruptedException e) {
 114                     LOG.warn("Failed to lock xml", e);
 115                 }
 116             }
 117         }
 118
 119         try (InputStream resourceAsStream = new FileInputStream(newFile)) {
 120             final XMLInputFactory factory = XMLInputFactory.newInstance();
 121             final XMLStreamReader reader = factory.createXMLStreamReader(resourceAsStream);
 122
 123             final SchemaNode schemaNode = SchemaContextUtil
 124                     .findDataSchemaNode(this.schemaContext, config.getSchemaPath());
 125             try (XmlParserStream xmlParser = XmlParserStream.create(streamWriter, this.schemaContext, schemaNode)) {
 126                 xmlParser.parse(reader);
 127             } catch (final URISyntaxException | XMLStreamException | IOException | SAXException e) {
 128                 LOG.warn("Failed to parse xml", e);
 129             } finally {
 130                 reader.close();
 131                 channel.close();
 132             }
 133         }
 134
 135         return result.getResult();
 136     }
 137
 138     @Override
 139     public synchronized AbstractRegistration registerConfigFile(final ConfigFileProcessor config) {
 140         final String pattern = INITIAL + config.getSchemaPath().getLastComponent().getLocalName() + EXTENSION;
 141         this.configServices.put(pattern, config);
 142
 143         final File[] fList = this.file.listFiles();
 144         final List<String> newFiles = new ArrayList<>();
 145         if (fList != null) {
 146             for (final File newFile : fList) {
 147                 if (newFile.isFile()) {
 148                     final String filename = newFile.getName();
 149                     if (Pattern.matches(pattern, filename)) {
 150                         newFiles.add(filename);
 151                     }
 152                 }
 153             }
 154         }
 155         for (final String filename : newFiles) {
 156             handleConfigFile(config, filename);
 157         }
 158         return new AbstractRegistration() {
 159             @Override
 160             protected void removeRegistration() {
 161                 synchronized (ConfigLoaderImpl.this) {
 162                     ConfigLoaderImpl.this.configServices.remove(pattern);
 163                 }
 164             }
 165         };
 166     }
 167
 168     @Override
 169     public BindingNormalizedNodeSerializer getBindingNormalizedNodeSerializer() {
 170         return this.bindingSerializer;
 171     }
 172
 173
 174     @Override
 175     public synchronized void close() {
 176         LOG.info("Config Loader service closed");
 177         this.closed = true;
 178         this.watcherThread.interrupt();
 179     }
 180
 181     private class ConfigLoaderImplRunnable implements Runnable {
 182         @GuardedBy("this")
 183         private final WatchService watchService;
 184
 185         ConfigLoaderImplRunnable(final WatchService watchService) {
 186             this.watchService = watchService;
 187         }
 188
 189         @Override
 190         public void run() {
 191             while (!Thread.currentThread().isInterrupted()) {
 192                 handleChanges();
 193             }
 194         }
 195
 196         private synchronized void handleChanges() {
 197             final WatchKey key;
 198             try {
 199                 key = this.watchService.take();
 200             } catch (final InterruptedException | ClosedWatchServiceException e) {
 201                 if (!ConfigLoaderImpl.this.closed) {
 202                     LOG.warn(INTERRUPTED, e);
 203                     Thread.currentThread().interrupt();
 204                 }
 205                 return;
 206             }
 207
 208             if (key != null) {
 209                 final List<String> fileNames = key.pollEvents()
 210                         .stream().map(event -> event.context().toString())
 211                         .collect(Collectors.toList());
 212                 fileNames.forEach(this::handleEvent);
 213
 214                 final boolean reset = key.reset();
 215                 if (!reset) {
 216                     LOG.warn("Could not reset the watch key.");
 217                 }
 218             }
 219         }
 220
 221         private synchronized void handleEvent(final String filename) {
 222             ConfigLoaderImpl.this.configServices.entrySet().stream()
 223                     .filter(entry -> Pattern.matches(entry.getKey(), filename))
 224                     .forEach(entry -> handleConfigFile(entry.getValue(), filename));
 225         }
 226     }
 227 }