Streams 132

streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/AbstractRegexExtensionExtractor.java

@@ -25,6 +25,8 @@
 import org.apache.streams.core.StreamsDatum;
 import org.apache.streams.core.StreamsProcessor;
 import org.apache.streams.pojo.json.Activity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
 import java.util.List;
@@ -42,6 +44,9 @@ public abstract class AbstractRegexExtensionExtractor<T> implements StreamsProce
     private final String extensionKey;
     private final String defaultPattern;
 
+    private final static Logger LOGGER = LoggerFactory.getLogger(AbstractRegexExtensionExtractor.class);
+
+
     private String pattern;
 
     protected AbstractRegexExtensionExtractor(String patternConfigKey, String extensionKey, String defaultPattern) {
@@ -56,29 +61,29 @@ public String getPattern() {
 
     @Override
     public List<StreamsDatum> process(StreamsDatum entry) {
-        if(!(entry.getDocument() instanceof Activity)) {
+        if (!(entry.getDocument() instanceof Activity)) {
             return Lists.newArrayList();
         }
-        if(Strings.isNullOrEmpty(pattern)) {
+        if (Strings.isNullOrEmpty(pattern)) {
             prepare(null);
         }
-        Activity activity = (Activity)entry.getDocument();
+        Activity activity = (Activity) entry.getDocument();
         Map<String, List<Integer>> matches = RegexUtils.extractMatches(pattern, activity.getContent());
         Collection<T> entities = ensureTargetObject(activity);
-        for(String key : matches.keySet()) {
+        for (String key : matches.keySet()) {
             entities.add(prepareObject(key));
         }
         return Lists.newArrayList(entry);
     }
 
     @Override
     public void prepare(Object configurationObject) {
-        if(configurationObject instanceof Map) {
-            if(((Map)configurationObject).containsKey(patternConfigKey)) {
-                pattern = (String)((Map)configurationObject).get(patternConfigKey);
+        if (configurationObject instanceof Map) {
+            if (((Map) configurationObject).containsKey(patternConfigKey)) {
+                pattern = (String) ((Map) configurationObject).get(patternConfigKey);
             }
-        } else if(configurationObject instanceof String) {
-            pattern = (String)configurationObject;
+        } else if (configurationObject instanceof String) {
+            pattern = (String) configurationObject;
         } else {
             pattern = defaultPattern;
         }

streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUtils.java

@@ -21,6 +21,8 @@
 
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Map;
@@ -33,6 +35,7 @@
 public class RegexUtils {
 
     private static final Map<String, Pattern> patternCache = Maps.newConcurrentMap();
+    private final static Logger LOGGER = LoggerFactory.getLogger(RegexUtils.class);
 
     private RegexUtils() {}
 
@@ -58,22 +61,32 @@ public static Map<String, List<Integer>> extractWordMatches(String pattern, Stri
     }
 
     protected static Map<String, List<Integer>> getMatches(String pattern, String content, int capture) {
-        Matcher m = getPattern(pattern).matcher(content);
-        Map<String, List<Integer>> matches = Maps.newHashMap();
-        while(m.find()) {
-            String group = capture > 0 ? m.group(capture) : m.group();
-            if(group != null && !group.equals("")) {
-                List<Integer> indices;
-                if(matches.containsKey(group)) {
-                    indices = matches.get(group);
-                } else {
-                    indices = Lists.newArrayList();
-                    matches.put(group, indices);
+        try {
+            Map<String, List<Integer>> matches = Maps.newHashMap();
+            if(content == null) {
+                return matches;
+            }
+
+            Matcher m = getPattern(pattern).matcher(content);
+            while (m.find()) {
+                String group = capture > 0 ? m.group(capture) : m.group();
+                if (group != null && !group.equals("")) {
+                    List<Integer> indices;
+                    if (matches.containsKey(group)) {
+                        indices = matches.get(group);
+                    } else {
+                        indices = Lists.newArrayList();
+                        matches.put(group, indices);
+                    }
+                    indices.add(m.start());
                 }
-                indices.add(m.start());
             }
+            return matches;
+        } catch (Throwable e) {
+            LOGGER.error("Throwable process {}", e);
+            e.printStackTrace();
+            throw new RuntimeException(e);
         }
-        return matches;
     }
 
     private static Pattern getPattern(String pattern) {
@@ -86,6 +99,4 @@ private static Pattern getPattern(String pattern) {
         }
         return p;
     }
-
-
-}
+}