added #getPropertyNames to c.f.FlowProcess

cwensel · Feb 16, 2012 · b5bd117 · b5bd117
1 parent 5eecc58
commit b5bd117
Show file tree

Hide file tree

Showing 7 changed files with 81 additions and 0 deletions.
diff --git a/src/core/cascading/flow/FlowProcess.java b/src/core/cascading/flow/FlowProcess.java
@@ -21,6 +21,8 @@
 package cascading.flow;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 
 import cascading.tap.Tap;
@@ -65,6 +67,12 @@ public Object getProperty( String key )
       return null;
       }
 
+    @Override
+    public Collection<String> getPropertyKeys()
+      {
+      return Collections.EMPTY_SET;
+      }
+
     public void keepAlive()
       {
       }
@@ -201,6 +209,13 @@ public void setCurrentSession( FlowSession currentSession )
    */
   public abstract Object getProperty( String key );
 
+  /**
+   * Method getPropertyKeys returns an immutable collection of all available property key values.
+   *
+   * @return a Collection<String>
+   */
+  public abstract Collection<String> getPropertyKeys();
+
   /**
    * Method keepAlive notifies the system that the current process is still alive. Use this method if a particular
    * {@link cascading.operation.Operation} takes some moments to complete. Each system is different, so calling

diff --git a/src/core/cascading/flow/FlowProcessWrapper.java b/src/core/cascading/flow/FlowProcessWrapper.java
@@ -21,6 +21,7 @@
 package cascading.flow;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 
 import cascading.tap.Tap;
@@ -94,6 +95,12 @@ public Object getProperty( String key )
     return delegate.getProperty( key );
     }
 
+  @Override
+  public Collection<String> getPropertyKeys()
+    {
+    return delegate.getPropertyKeys();
+    }
+
   @Override
   public void keepAlive()
     {

diff --git a/src/core/cascading/flow/stream/PipeFlowProcess.java b/src/core/cascading/flow/stream/PipeFlowProcess.java
@@ -20,6 +20,11 @@
 
 package cascading.flow.stream;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
 import cascading.flow.FlowProcess;
 import cascading.flow.FlowProcessWrapper;
 import cascading.pipe.ConfigDef;
@@ -71,4 +76,15 @@ public Object getProperty( String key )
     {
     return configDef.apply( key, getter );
     }
+
+  @Override
+  public Collection<String> getPropertyKeys()
+    {
+    Set<String> keys = new HashSet<String>();
+
+    keys.addAll( getDelegate().getPropertyKeys() );
+    keys.addAll( configDef.getAllKeys() );
+
+    return Collections.unmodifiableSet( keys );
+    }
   }
diff --git a/src/core/cascading/pipe/ConfigDef.java b/src/core/cascading/pipe/ConfigDef.java
@@ -21,9 +21,12 @@
 package cascading.pipe;
 
 import java.io.Serializable;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * The ConfigDef class allows for the creation of a configuration properties template to be applied to an existing
@@ -169,4 +172,14 @@ public void apply( Mode mode, Setter setter )
         }
       }
     }
+
+  public Collection<String> getAllKeys()
+    {
+    Set<String> keys = new HashSet<String>();
+
+    for( Map<String, String> map : config.values() )
+      keys.addAll( map.keySet() );
+
+    return Collections.unmodifiableSet( keys );
+    }
   }
diff --git a/src/hadoop/cascading/flow/hadoop/HadoopFlowProcess.java b/src/hadoop/cascading/flow/hadoop/HadoopFlowProcess.java
@@ -21,7 +21,11 @@
 package cascading.flow.hadoop;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 import cascading.CascadingException;
 import cascading.flow.FlowProcess;
@@ -211,6 +215,17 @@ public Object getProperty( String key )
     return jobConf.get( key );
     }
 
+  @Override
+  public Collection<String> getPropertyKeys()
+    {
+    Set<String> keys = new HashSet<String>();
+
+    for( Map.Entry<String, String> entry : jobConf )
+      keys.add( entry.getKey() );
+
+    return Collections.unmodifiableSet( keys );
+    }
+
   @Override
   public void keepAlive()
     {

diff --git a/src/local/cascading/flow/local/LocalFlowProcess.java b/src/local/cascading/flow/local/LocalFlowProcess.java
@@ -21,6 +21,8 @@
 package cascading.flow.local;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Properties;
 
@@ -77,6 +79,12 @@ public Object getProperty( String key )
     return config.getProperty( key );
     }
 
+  @Override
+  public Collection<String> getPropertyKeys()
+    {
+    return Collections.unmodifiableSet( config.stringPropertyNames() );
+    }
+
   @Override
   public void keepAlive()
     {

diff --git a/src/test/cascading/operation/filter/LimitFilterTest.java b/src/test/cascading/operation/filter/LimitFilterTest.java
@@ -21,6 +21,7 @@
 package cascading.operation.filter;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 
 import cascading.CascadingTestCase;
@@ -95,6 +96,12 @@ public Object getProperty( String key )
       return null;
       }
 
+    @Override
+    public Collection<String> getPropertyKeys()
+      {
+      return null;
+      }
+
     @Override
     public void keepAlive()
       {