IGNITE-22530 Refactor AbstractIgniteCdcStreamer for use with CdcRegexManager

Author: Andrei Nadyktov

modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractIgniteCdcStreamer.java

@@ -17,16 +17,10 @@
 
 package org.apache.ignite.cdc;
 
-import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.HashSet;
 import java.util.Iterator;
-import java.util.List;
-import java.util.Optional;
 import java.util.Set;
-import java.util.regex.Pattern;
-import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
 
 import org.apache.ignite.IgniteCheckedException;
@@ -44,8 +38,6 @@
 import org.apache.ignite.metric.MetricRegistry;
 import org.apache.ignite.resources.LoggerResource;
 
-import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
-import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
 import static org.apache.ignite.cdc.kafka.IgniteToKafkaCdcStreamer.DFLT_IS_ONLY_PRIMARY;
 
 /**
@@ -78,33 +70,21 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     /** */
     public static final String LAST_EVT_SENT_TIME_DESC = "Timestamp of last applied event to destination cluster";
 
-    /** File with saved names of caches added by cache masks. */
-    private static final String SAVED_CACHES_FILE = "caches";
-
-    /** Temporary file with saved names of caches added by cache masks. */
-    private static final String SAVED_CACHES_TMP_FILE = "caches_tmp";
-
-    /** CDC directory path. */
-    private Path cdcDir;
-
     /** Handle only primary entry flag. */
     private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
 
     /** Cache names. */
     private Set<String> caches;
 
+    /** Regexp manager. */
+    private CdcRegexManager regexManager;
+
     /** Include regex templates for cache names. */
     private Set<String> includeTemplates = new HashSet<>();
 
-    /** Compiled include regex patterns for cache names. */
-    private Set<Pattern> includeFilters;
-
     /** Exclude regex templates for cache names. */
     private Set<String> excludeTemplates = new HashSet<>();
 
-    /** Compiled exclude regex patterns for cache names. */
-    private Set<Pattern> excludeFilters;
-
     /** Cache IDs. */
     protected Set<Integer> cachesIds;
 
@@ -139,24 +119,18 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     @Override public void start(MetricRegistry reg, Path cdcDir) {
         A.notEmpty(caches, "caches");
 
-        this.cdcDir = cdcDir;
+        regexManager = new CdcRegexManager(cdcDir, log);
 
         cachesIds = caches.stream()
             .mapToInt(CU::cacheId)
             .boxed()
             .collect(Collectors.toSet());
 
-        prepareRegexFilters();
+        regexManager.compileRegexp(includeTemplates, excludeTemplates);
 
-        try {
-            loadCaches().stream()
-                .filter(this::matchesFilters)
-                .map(CU::cacheId)
-                .forEach(cachesIds::add);
-        }
-        catch (IOException e) {
-            throw new IgniteException(e);
-        }
+        regexManager.getSavedCaches().stream()
+            .map(CU::cacheId)
+            .forEach(cachesIds::add);
 
         MetricRegistryImpl mreg = (MetricRegistryImpl)reg;
 
@@ -195,147 +169,26 @@ public abstract class AbstractIgniteCdcStreamer implements CdcConsumerEx {
     /** {@inheritDoc} */
     @Override public void onCacheChange(Iterator<CdcCacheEvent> cacheEvents) {
         cacheEvents.forEachRemaining(e -> {
-            matchWithRegexTemplates(e.configuration().getName());
+            matchWithRegex(e.configuration().getName());
         });
     }
 
     /**
      * Finds match between cache name and user's regex templates.
-     * If match is found, adds this cache's id to id's list and saves cache name to file.
+     * If match is found, adds this cache's id to id's list.
      *
      * @param cacheName Cache name.
      */
-    private void matchWithRegexTemplates(String cacheName) {
+    private void matchWithRegex(String cacheName) {
         int cacheId = CU.cacheId(cacheName);
 
-        if (!cachesIds.contains(cacheId) && matchesFilters(cacheName)) {
+        if (!cachesIds.contains(cacheId) && regexManager.match(cacheName))
             cachesIds.add(cacheId);
-
-            try {
-                List<String> caches = loadCaches();
-
-                caches.add(cacheName);
-
-                save(caches);
-            }
-            catch (IOException e) {
-                throw new IgniteException(e);
-            }
-
-            if (log.isInfoEnabled())
-                log.info("Cache has been added to replication [cacheName=" + cacheName + "]");
-        }
-    }
-
-    /**
-     * Writes caches list to file
-     *
-     * @param caches Caches list.
-     */
-    private void save(List<String> caches) throws IOException {
-        if (cdcDir == null) {
-            throw new IgniteException("Can't write to '" + SAVED_CACHES_FILE + "' file. Cdc directory is null");
-        }
-        Path savedCachesPath = cdcDir.resolve(SAVED_CACHES_FILE);
-        Path tmpSavedCachesPath = cdcDir.resolve(SAVED_CACHES_TMP_FILE);
-
-        StringBuilder cacheList = new StringBuilder();
-
-        for (String cache : caches) {
-            cacheList.append(cache);
-
-            cacheList.append('\n');
-        }
-
-        Files.write(tmpSavedCachesPath, cacheList.toString().getBytes());
-
-        Files.move(tmpSavedCachesPath, savedCachesPath, ATOMIC_MOVE, REPLACE_EXISTING);
-    }
-
-    /**
-     * Loads saved caches from file.
-     *
-     * @return List of saved caches names.
-     */
-    private List<String> loadCaches() throws IOException {
-        if (cdcDir == null) {
-            throw new IgniteException("Can't load '" + SAVED_CACHES_FILE + "' file. Cdc directory is null");
-        }
-        Path savedCachesPath = cdcDir.resolve(SAVED_CACHES_FILE);
-
-        if (Files.notExists(savedCachesPath)) {
-            Files.createFile(savedCachesPath);
-
-            if (log.isInfoEnabled())
-                log.info("Cache list created: " + savedCachesPath);
-        }
-
-        return Files.readAllLines(savedCachesPath);
-    }
-
-    /**
-     * Compiles regex patterns from user templates.
-     *
-     * @throws PatternSyntaxException If the template's syntax is invalid
-     */
-    private void prepareRegexFilters() {
-        includeFilters = includeTemplates.stream()
-            .map(Pattern::compile)
-            .collect(Collectors.toSet());
-
-        excludeFilters = excludeTemplates.stream()
-            .map(Pattern::compile)
-            .collect(Collectors.toSet());
-    }
-
-    /**
-     * Matches cache name with compiled regex patterns.
-     *
-     * @param cacheName Cache name.
-     * @return True if cache name match include patterns and don't match exclude patterns.
-     */
-    private boolean matchesFilters(String cacheName) {
-        boolean matchesInclude = includeFilters.stream()
-            .anyMatch(pattern -> pattern.matcher(cacheName).matches());
-
-        boolean notMatchesExclude = excludeFilters.stream()
-            .noneMatch(pattern -> pattern.matcher(cacheName).matches());
-
-        return matchesInclude && notMatchesExclude;
     }
 
     /** {@inheritDoc} */
     @Override public void onCacheDestroy(Iterator<Integer> caches) {
-        caches.forEachRemaining(this::deleteRegexpCacheIfPresent);
-    }
-
-    /**
-     * Removes cache added by regexp from cache list, if this cache is present in file, to prevent file size overflow.
-     *
-     * @param cacheId Cache id.
-     */
-    private void deleteRegexpCacheIfPresent(Integer cacheId) {
-        try {
-            List<String> caches = loadCaches();
-
-            Optional<String> cacheName = caches.stream()
-                .filter(name -> CU.cacheId(name) == cacheId)
-                .findAny();
-
-            if (cacheName.isPresent()) {
-                String name = cacheName.get();
-
-                caches.remove(name);
-
-                save(caches);
-
-                if (log.isInfoEnabled())
-                    log.info("Cache has been removed from replication [cacheName=" + name + ']');
-            }
-        }
-        catch (IOException e) {
-            throw new IgniteException(e);
-        }
+        caches.forEachRemaining(regexManager::deleteRegexpCacheIfPresent);
     }
 
     /** {@inheritDoc} */