Remove RecordMapPseudonymizer since pseudonymization can be done usin…

…g PseudoFuncs directly. Simplified StreamProcessor.
statisticsnorway · Feb 6, 2024 · cdfa6b6 · cdfa6b6
1 parent fbc5ec2
commit cdfa6b6
Show file tree

Hide file tree

Showing 13 changed files with 72 additions and 221 deletions.
diff --git a/pom.xml b/pom.xml
@@ -14,7 +14,7 @@
     <github.repository>statisticsnorway/dapla-dlp-pseudo-core</github.repository>
 
     <!-- Dependency versions -->
-    <dapla-dlp-pseudo-func.version>1.2.8</dapla-dlp-pseudo-func.version>
+    <dapla-dlp-pseudo-func.version>1.3.0-SNAPSHOT</dapla-dlp-pseudo-func.version>
     <guava.version>32.0.0-jre</guava.version>
     <jsonassert.version>1.5.1</jsonassert.version>
     <logback.version>1.4.6</logback.version>

diff --git a/src/main/java/no/ssb/dlp/pseudo/core/StreamProcessor.java b/src/main/java/no/ssb/dlp/pseudo/core/StreamProcessor.java
@@ -8,7 +8,7 @@
 import java.util.Map;
 
 public interface StreamProcessor {
-    <T> Completable init(InputStream is, RecordMapSerializer<T> serializer);
+    <T> Completable init(InputStream is);
     <T> Flowable<T> process(InputStream is, RecordMapSerializer<T> serializer);
     @FunctionalInterface
     public interface ItemProcessor {

diff --git a/src/main/java/no/ssb/dlp/pseudo/core/csv/CsvStreamProcessor.java b/src/main/java/no/ssb/dlp/pseudo/core/csv/CsvStreamProcessor.java
@@ -12,6 +12,7 @@
 import no.ssb.dlp.pseudo.core.StreamProcessor;
 import no.ssb.dlp.pseudo.core.map.RecordMapProcessor;
 import no.ssb.dlp.pseudo.core.map.RecordMapSerializer;
+import no.ssb.dlp.pseudo.core.map.RecordMapSerializerFactory;
 
 import java.io.InputStream;
 import java.util.LinkedHashMap;
@@ -22,20 +23,20 @@
 @Slf4j
 public class CsvStreamProcessor implements StreamProcessor {
 
-    private final RecordMapProcessor recordMapProcessor;
+    private final RecordMapProcessor<String> recordMapProcessor;
 
     @Override
-    public <T> Completable init(InputStream is, RecordMapSerializer<T> serializer) {
+    public <T> Completable init(InputStream is) {
         if (recordMapProcessor.hasPreprocessors()) {
-            return Completable.fromPublisher(processStream(is, serializer, (map) -> recordMapProcessor.init(map)));
+            return Completable.fromPublisher(processStream(is, RecordMapSerializerFactory.emptySerializer(), recordMapProcessor::init));
         } else {
             return Completable.complete();
         }
     }
 
     @Override
     public <T> Flowable<T> process(InputStream is, RecordMapSerializer<T> serializer) {
-        return processStream(is, serializer, (map) -> recordMapProcessor.process(map));
+        return processStream(is, serializer, recordMapProcessor::process);
     }
 
     <T> CsvProcessorContext<T> initCsvProcessorContext(InputStream is, RecordMapSerializer<T> serializer) {

diff --git a/src/main/java/no/ssb/dlp/pseudo/core/csv/CsvStreamPseudonymizer.java b/src/main/java/no/ssb/dlp/pseudo/core/csv/CsvStreamPseudonymizer.java
diff --git a/src/main/java/no/ssb/dlp/pseudo/core/field/FieldPseudonymizer.java b/src/main/java/no/ssb/dlp/pseudo/core/field/FieldPseudonymizer.java
@@ -17,6 +17,7 @@
 import static no.ssb.dlp.pseudo.core.PseudoOperation.DEPSEUDONYMIZE;
 import static no.ssb.dlp.pseudo.core.PseudoOperation.PSEUDONYMIZE;
 
+@Deprecated(forRemoval = true) // Should use PseudoFuncs instead
 public class FieldPseudonymizer {
 
     private final PseudoFuncs pseudoFuncs;
@@ -25,43 +26,41 @@ private FieldPseudonymizer(PseudoFuncs pseudoFuncs) {
         this.pseudoFuncs = pseudoFuncs;
     }
 
-    public String pseudonymize(FieldDescriptor field, String varValue) {
+    public PseudoFuncOutput pseudonymize(FieldDescriptor field, String varValue) {
         return process(PSEUDONYMIZE, field, varValue);
     }
 
-    public String depseudonymize(FieldDescriptor field, String varValue) {
+    public PseudoFuncOutput depseudonymize(FieldDescriptor field, String varValue) {
         return process(DEPSEUDONYMIZE, field, varValue);
     }
 
     public Optional<PseudoFuncRuleMatch> match(FieldDescriptor field) {
         return pseudoFuncs.findPseudoFunc(field);
     }
 
-    public String init(FieldDescriptor field, String varValue) {
+    public void init(FieldDescriptor field, String varValue) {
         Optional<PseudoFuncRuleMatch> match = pseudoFuncs.findPseudoFunc(field);
         if (match.isPresent()) {
             match.get().getFunc().init(PseudoFuncInput.of(varValue));
         }
-        return varValue;
     }
 
-    private String process(PseudoOperation operation, FieldDescriptor field, String varValue) {
+    private PseudoFuncOutput process(PseudoOperation operation, FieldDescriptor field, String varValue) {
 
         // TODO: This check is function type specific (e.g. only applies for FPE?)
         if (varValue == null || varValue.length() <= 2) {
-            return varValue;
+            return PseudoFuncOutput.of(varValue);
         }
 
         PseudoFuncRuleMatch match = pseudoFuncs.findPseudoFunc(field).orElse(null);
         try {
             if (match == null) {
-                return varValue;
+                return PseudoFuncOutput.of(varValue);
             }
 
-            PseudoFuncOutput res = (operation == PSEUDONYMIZE)
+            return (operation == PSEUDONYMIZE)
               ? match.getFunc().apply(PseudoFuncInput.of(varValue))
               : match.getFunc().restore(PseudoFuncInput.of(varValue));
-            return (String) res.getFirstValue();
         }
         catch (Exception e) {
             throw new PseudoException(operation + " error - field='" + field.getPath() + "', originalValue='" + varValue  + "'", e);
@@ -74,6 +73,8 @@ public static class Builder {
 
         private Collection<PseudoKeyset> keysets;
 
+        private String correlationId;
+
         public Builder secrets(Collection<PseudoSecret> secrets) {
             this.secrets = secrets;
             return this;
@@ -89,10 +90,15 @@ public Builder keysets(Collection<PseudoKeyset> keysets) {
             return this;
         }
 
+        public Builder correlationId(String correlationId) {
+            this.correlationId = correlationId;
+            return this;
+        }
+
         public FieldPseudonymizer build() {
             Objects.requireNonNull(secrets, "PseudoSecrets can't be null");
             Objects.requireNonNull(rules, "PseudoFuncRule collection can't be null");
-            return new FieldPseudonymizer(new PseudoFuncs(rules, secrets, keysets));
+            return new FieldPseudonymizer(new PseudoFuncs(rules, secrets, keysets, correlationId));
         }
     }
 }
diff --git a/src/main/java/no/ssb/dlp/pseudo/core/func/PseudoFuncs.java b/src/main/java/no/ssb/dlp/pseudo/core/func/PseudoFuncs.java
@@ -30,14 +30,19 @@ public class PseudoFuncs {
 
     private final Map<PseudoFuncRule, PseudoFunc> ruleToFuncMap = new LinkedHashMap<>();
 
+
     //TODO: Validate that all required secrets are available
-    public PseudoFuncs(Collection<PseudoFuncRule> rules, Collection<PseudoSecret> pseudoSecrets, Collection<PseudoKeyset> keysets) {
-        Map<PseudoFuncRule, PseudoFuncConfig> ruleToPseudoFuncConfigs = initPseudoFuncConfigs(rules, pseudoSecrets, keysets);
+    public PseudoFuncs(Collection<PseudoFuncRule> rules, Collection<PseudoSecret> pseudoSecrets,
+                       Collection<PseudoKeyset> keysets, String correlationId) {
+        Map<PseudoFuncRule, PseudoFuncConfig> ruleToPseudoFuncConfigs = initPseudoFuncConfigs(rules, pseudoSecrets, keysets, correlationId);
         rules.forEach(rule -> ruleToFuncMap.put(rule, PseudoFuncFactory.create(ruleToPseudoFuncConfigs.get(rule))));
     }
 
     // TODO: Move these init functions elsewhere?
-    static Map<PseudoFuncRule, PseudoFuncConfig> initPseudoFuncConfigs(Collection<PseudoFuncRule> pseudoRules, Collection<PseudoSecret> pseudoSecrets, Collection<PseudoKeyset> pseudoKeysets) {
+    static Map<PseudoFuncRule, PseudoFuncConfig> initPseudoFuncConfigs(Collection<PseudoFuncRule> pseudoRules,
+                                                                       Collection<PseudoSecret> pseudoSecrets,
+                                                                       Collection<PseudoKeyset> pseudoKeysets,
+                                                                       String correlationId) {
 
         Map<String, PseudoSecret> pseudoSecretsMap = pseudoSecrets.stream().collect(
           Collectors.toMap(PseudoSecret::getName, Function.identity()));
@@ -49,6 +54,7 @@ static Map<PseudoFuncRule, PseudoFuncConfig> initPseudoFuncConfigs(Collection<Ps
           Function.identity(),
           rule -> {
               PseudoFuncConfig funcConfig = PseudoFuncConfigFactory.get(rule.getFunc());
+              funcConfig.add(Param.CORRELATION_ID, correlationId);
 
               if (FpeFunc.class.getName().equals(funcConfig.getFuncImpl())) {
                   enrichLegacyFpeFuncConfig(funcConfig, pseudoSecretsMap);
@@ -168,4 +174,7 @@ public PseudoFuncConfigException(String message, Exception e) {
         }
     }
 
+    public static final class Param {
+        public static final String CORRELATION_ID = "correlationId";
+    }
 }
diff --git a/src/main/java/no/ssb/dlp/pseudo/core/json/JsonStreamProcessor.java b/src/main/java/no/ssb/dlp/pseudo/core/json/JsonStreamProcessor.java
@@ -15,6 +15,7 @@
 import no.ssb.dlp.pseudo.core.map.RecordMap;
 import no.ssb.dlp.pseudo.core.map.RecordMapProcessor;
 import no.ssb.dlp.pseudo.core.map.RecordMapSerializer;
+import no.ssb.dlp.pseudo.core.map.RecordMapSerializerFactory;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -36,9 +37,9 @@ public class JsonStreamProcessor implements StreamProcessor {
     private final RecordMapProcessor recordMapProcessor;
 
     @Override
-    public <T> Completable init(InputStream is, RecordMapSerializer<T> serializer) {
+    public <T> Completable init(InputStream is) {
         if (recordMapProcessor.hasPreprocessors()) {
-            return Completable.fromPublisher(processStream(is, serializer, (map) -> recordMapProcessor.init(map)));
+            return Completable.fromPublisher(processStream(is, RecordMapSerializerFactory.emptySerializer(), recordMapProcessor::init));
         } else {
             return Completable.complete();
         }

diff --git a/src/main/java/no/ssb/dlp/pseudo/core/json/JsonStreamPseudonymizer.java b/src/main/java/no/ssb/dlp/pseudo/core/json/JsonStreamPseudonymizer.java
diff --git a/src/main/java/no/ssb/dlp/pseudo/core/map/RecordMapProcessor.java b/src/main/java/no/ssb/dlp/pseudo/core/map/RecordMapProcessor.java
@@ -1,14 +1,22 @@
 package no.ssb.dlp.pseudo.core.map;
 
+import io.reactivex.processors.FlowableProcessor;
+import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import no.ssb.dlp.pseudo.core.field.ValueInterceptorChain;
 
 import java.util.Map;
 
 @RequiredArgsConstructor
-public class RecordMapProcessor {
+public class RecordMapProcessor<T> {
     private final ValueInterceptorChain valueInterceptorChain;
-
+    @Getter
+    private final MetadataProcessor<T> metadataProcessor;
+    @FunctionalInterface
+    public interface MetadataProcessor<T> {
+        // The MetadataProcessor is used to publish/subscribe to events related to the processing of each RecordMap
+        FlowableProcessor<T> toFlowableProcessor();
+    }
     public Map<String, Object> init(Map<String, Object> r) {
         return MapTraverser.traverse(r, valueInterceptorChain::init);
     }