diff --git a/src/jvm/backtype/storm/task/GeneralTopologyContext.java b/src/jvm/backtype/storm/task/GeneralTopologyContext.java
new file mode 100644
index 0000000..da8d3a7
--- /dev/null
+++ b/src/jvm/backtype/storm/task/GeneralTopologyContext.java
@@ -0,0 +1,193 @@
+package backtype.storm.task;
+
+import backtype.storm.Config;
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.generated.Grouping;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.generated.StreamInfo;
+import backtype.storm.tuple.Fields;
+import backtype.storm.utils.ThriftTopologyUtils;
+import backtype.storm.utils.Utils;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.json.simple.JSONValue;
+import org.json.simple.JSONAware;
+
+/**
+ * A TopologyContext is given to bolts and spouts in their "prepare" and "open"
+ * methods, respectively. This object provides information about the component's
+ * place within the topology, such as task ids, inputs and outputs, etc.
+ *
+ * <p>The TopologyContext is also used to declare ISubscribedState objects to
+ * synchronize state with StateSpouts this object is subscribed to.</p>
+ */
+public class GeneralTopologyContext implements JSONAware {
+    private StormTopology _topology;
+    private Map<Integer, String> _taskToComponent;
+    private Map<String, List<Integer>> _componentToTasks;
+    private String _stormId;
+    protected Map _stormConf;
+    
+    public GeneralTopologyContext(StormTopology topology, Map stormConf,
+            Map<Integer, String> taskToComponent, String stormId) {
+        _topology = topology;
+        _stormConf = stormConf;
+        _taskToComponent = taskToComponent;
+        _stormId = stormId;
+        _componentToTasks = new HashMap<String, List<Integer>>();
+        for(Integer task: taskToComponent.keySet()) {
+            String component = taskToComponent.get(task);
+            List<Integer> curr = _componentToTasks.get(component);
+            if(curr==null) curr = new ArrayList<Integer>();
+            curr.add(task);
+            _componentToTasks.put(component, curr);
+        }
+        for(String component: _componentToTasks.keySet()) {
+            List<Integer> tasks = _componentToTasks.get(component);
+            Collections.sort(tasks);
+        }
+    }
+
+    /**
+     * Gets the unique id assigned to this topology. The id is the storm name with a
+     * unique nonce appended to it.
+     * @return the storm id
+     */
+    public String getStormId() {
+        return _stormId;
+    }
+
+    /**
+     * Gets the Thrift object representing the topology.
+     * 
+     * @return the Thrift definition representing the topology
+     */
+    public StormTopology getRawTopology() {
+        return _topology;
+    }
+
+    /**
+     * Gets the component id for the specified task id. The component id maps
+     * to a component id specified for a Spout or Bolt in the topology definition.
+     *
+     * @param taskId the task id
+     * @return the component id for the input task id
+     */
+    public String getComponentId(int taskId) {
+        return _taskToComponent.get(taskId);
+    }
+
+    /**
+     * Gets the set of streams declared for the specified component.
+     */
+    public Set<String> getComponentStreams(String componentId) {
+        return getComponentCommon(componentId).get_streams().keySet();
+    }
+
+    /**
+     * Gets the task ids allocated for the given component id. The task ids are
+     * always returned in ascending order.
+     */
+    public List<Integer> getComponentTasks(String componentId) {
+        List<Integer> ret = _componentToTasks.get(componentId);
+        if(ret==null) return new ArrayList<Integer>();
+        else return new ArrayList<Integer>(ret);
+    }
+
+    /**
+     * Gets the declared output fields for the specified component/stream.
+     */
+    public Fields getComponentOutputFields(String componentId, String streamId) {
+        StreamInfo streamInfo = getComponentCommon(componentId).get_streams().get(streamId);
+        if(streamInfo==null) {
+            throw new IllegalArgumentException("No output fields defined for component:stream " + componentId + ":" + streamId);
+        }
+        return new Fields(streamInfo.get_output_fields());
+    }
+
+    /**
+     * Gets the declared output fields for the specified global stream id.
+     */
+    public Fields getComponentOutputFields(GlobalStreamId id) {
+        return getComponentOutputFields(id.get_componentId(), id.get_streamId());
+    }    
+    
+    /**
+     * Gets the declared inputs to the specified component.
+     *
+     * @return A map from subscribed component/stream to the grouping subscribed with.
+     */
+    public Map<GlobalStreamId, Grouping> getSources(String componentId) {
+        return getComponentCommon(componentId).get_inputs();
+    }
+
+    /**
+     * Gets information about who is consuming the outputs of the specified component,
+     * and how.
+     *
+     * @return Map from stream id to component id to the Grouping used.
+     */
+    public Map<String, Map<String, Grouping>> getTargets(String componentId) {
+        Map<String, Map<String, Grouping>> ret = new HashMap<String, Map<String, Grouping>>();
+        for(String otherComponentId: getComponentIds()) {
+            Map<GlobalStreamId, Grouping> inputs = getComponentCommon(otherComponentId).get_inputs();
+            for(GlobalStreamId id: inputs.keySet()) {
+                if(id.get_componentId().equals(componentId)) {
+                    Map<String, Grouping> curr = ret.get(id.get_streamId());
+                    if(curr==null) curr = new HashMap<String, Grouping>();
+                    curr.put(otherComponentId, inputs.get(id));
+                    ret.put(id.get_streamId(), curr);
+                }
+            }
+        }
+        return ret;
+    }
+
+    @Override
+    public String toJSONString() {
+        Map obj = new HashMap();
+        obj.put("task->component", _taskToComponent);
+        // TODO: jsonify StormTopology
+        // at the minimum should send source info
+        return JSONValue.toJSONString(obj);
+    }
+
+    /**
+     * Gets a map from task id to component id.
+     */
+    public Map<Integer, String> getTaskToComponent() {
+        return _taskToComponent;
+    }
+    
+    /**
+     * Gets a list of all component ids in this topology
+     */
+    public Set<String> getComponentIds() {
+        return ThriftTopologyUtils.getComponentIds(getRawTopology());
+    }
+
+    public ComponentCommon getComponentCommon(String componentId) {
+        return ThriftTopologyUtils.getComponentCommon(getRawTopology(), componentId);
+    }
+    
+    public int maxTopologyMessageTimeout() {
+        Integer max = Utils.getInt(_stormConf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS));
+        for(String spout: getRawTopology().get_spouts().keySet()) {
+            ComponentCommon common = getComponentCommon(spout);
+            String jsonConf = common.get_json_conf();
+            if(jsonConf!=null) {
+                Map conf = (Map) JSONValue.parse(jsonConf);
+                Object comp = conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS);
+                if(comp!=null) {
+                    max = Math.max(Utils.getInt(comp), max);
+                }
+            }
+        }
+        return max;
+    }
+}
\ No newline at end of file
diff --git a/src/jvm/backtype/storm/task/TopologyContext.java b/src/jvm/backtype/storm/task/TopologyContext.java
index 7cfe9b5..8778cb2 100644
--- a/src/jvm/backtype/storm/task/TopologyContext.java
+++ b/src/jvm/backtype/storm/task/TopologyContext.java
@@ -20,7 +20,6 @@
 import java.util.Set;
 import org.apache.commons.lang.NotImplementedException;
 import org.json.simple.JSONValue;
-import org.json.simple.JSONAware;
 
 /**
  * A TopologyContext is given to bolts and spouts in their "prepare" and "open"
@@ -30,17 +29,12 @@
  * <p>The TopologyContext is also used to declare ISubscribedState objects to
  * synchronize state with StateSpouts this object is subscribed to.</p>
  */
-public class TopologyContext implements JSONAware {
-    private StormTopology _topology;
-    private Map<Integer, String> _taskToComponent;
+public class TopologyContext extends GeneralTopologyContext {
     private Integer _taskId;
-    private Map<String, List<Integer>> _componentToTasks;
     private String _codeDir;
     private String _pidDir;
-    private String _stormId;
     private Object _taskData = null;
     private List<ITaskHook> _hooks = new ArrayList<ITaskHook>();
-    private Map _stormConf;
     private Integer _workerPort;
     private List<Integer> _workerTasks;
     
@@ -48,28 +42,13 @@
             Map<Integer, String> taskToComponent, String stormId,
             String codeDir, String pidDir, Integer taskId,
             Integer workerPort, List<Integer> workerTasks) {
-        _topology = topology;
-        _stormConf = stormConf;
+        super(topology, stormConf, taskToComponent, stormId);
         _workerPort = workerPort;
-        _taskToComponent = taskToComponent;
-        _stormId = stormId;
         _taskId = taskId;
-        _componentToTasks = new HashMap<String, List<Integer>>();
         _pidDir = pidDir;
         _codeDir = codeDir;
         _workerTasks = new ArrayList<Integer>(workerTasks);
         Collections.sort(_workerTasks);
-        for(Integer task: taskToComponent.keySet()) {
-            String component = taskToComponent.get(task);
-            List<Integer> curr = _componentToTasks.get(component);
-            if(curr==null) curr = new ArrayList<Integer>();
-            curr.add(task);
-            _componentToTasks.put(component, curr);
-        }
-        for(String component: _componentToTasks.keySet()) {
-            List<Integer> tasks = _componentToTasks.get(component);
-            Collections.sort(tasks);
-        }
     }
 
     /**
@@ -128,15 +107,6 @@
     }
 
     /**
-     * Gets the unique id assigned to this topology. The id is the storm name with a
-     * unique nonce appended to it.
-     * @return the storm id
-     */
-    public String getStormId() {
-        return _stormId;
-    }
-
-    /**
      * Gets the task id of this task.
      * 
      * @return the task id
@@ -146,26 +116,6 @@
     }
 
     /**
-     * Gets the Thrift object representing the topology.
-     * 
-     * @return the Thrift definition representing the topology
-     */
-    public StormTopology getRawTopology() {
-        return _topology;
-    }
-
-    /**
-     * Gets the component id for the specified task id. The component id maps
-     * to a component id specified for a Spout or Bolt in the topology definition.
-     *
-     * @param taskId the task id
-     * @return the component id for the input task id
-     */
-    public String getComponentId(int taskId) {
-        return _taskToComponent.get(taskId);
-    }
-
-    /**
      * Gets the component id for this task. The component id maps
      * to a component id specified for a Spout or Bolt in the topology definition.
      * @return
@@ -198,23 +148,6 @@
     }
 
     /**
-     * Gets the set of streams declared for the specified component.
-     */
-    public Set<String> getComponentStreams(String componentId) {
-        return getComponentCommon(componentId).get_streams().keySet();
-    }
-
-    /**
-     * Gets the task ids allocated for the given component id. The task ids are
-     * always returned in ascending order.
-     */
-    public List<Integer> getComponentTasks(String componentId) {
-        List<Integer> ret = _componentToTasks.get(componentId);
-        if(ret==null) return new ArrayList<Integer>();
-        else return new ArrayList<Integer>(ret);
-    }
-
-    /**
      * Gets the index of this task id in getComponentTasks(getThisComponentId()).
      * An example use case for this method is determining which task
      * accesses which resource in a distributed resource to ensure an even distribution.
@@ -229,24 +162,6 @@
         }
         throw new RuntimeException("Fatal: could not find this task id in this component");
     }
-
-    /**
-     * Gets the declared output fields for the specified component/stream.
-     */
-    public Fields getComponentOutputFields(String componentId, String streamId) {
-        StreamInfo streamInfo = getComponentCommon(componentId).get_streams().get(streamId);
-        if(streamInfo==null) {
-            throw new IllegalArgumentException("No output fields defined for component:stream " + componentId + ":" + streamId);
-        }
-        return new Fields(streamInfo.get_output_fields());
-    }
-
-    /**
-     * Gets the declared output fields for the specified global stream id.
-     */
-    public Fields getComponentOutputFields(GlobalStreamId id) {
-        return getComponentOutputFields(id.get_componentId(), id.get_streamId());
-    }    
     
     /**
      * Gets the declared inputs to this component.
@@ -256,15 +171,6 @@
     public Map<GlobalStreamId, Grouping> getThisSources() {
         return getSources(getThisComponentId());
     }
-    
-    /**
-     * Gets the declared inputs to the specified component.
-     *
-     * @return A map from subscribed component/stream to the grouping subscribed with.
-     */
-    public Map<GlobalStreamId, Grouping> getSources(String componentId) {
-        return getComponentCommon(componentId).get_inputs();
-    }
 
     /**
      * Gets information about who is consuming the outputs of this component, and how.
@@ -276,37 +182,6 @@
     }
 
     /**
-     * Gets information about who is consuming the outputs of the specified component,
-     * and how.
-     *
-     * @return Map from stream id to component id to the Grouping used.
-     */
-    public Map<String, Map<String, Grouping>> getTargets(String componentId) {
-        Map<String, Map<String, Grouping>> ret = new HashMap<String, Map<String, Grouping>>();
-        for(String otherComponentId: getComponentIds()) {
-            Map<GlobalStreamId, Grouping> inputs = getComponentCommon(otherComponentId).get_inputs();
-            for(GlobalStreamId id: inputs.keySet()) {
-                if(id.get_componentId().equals(componentId)) {
-                    Map<String, Grouping> curr = ret.get(id.get_streamId());
-                    if(curr==null) curr = new HashMap<String, Grouping>();
-                    curr.put(otherComponentId, inputs.get(id));
-                    ret.put(id.get_streamId(), curr);
-                }
-            }
-        }
-        return ret;
-    }
-
-    public String toJSONString() {
-        Map obj = new HashMap();
-        obj.put("taskid", _taskId);
-        obj.put("task->component", _taskToComponent);
-        // TODO: jsonify StormTopology
-        // at the minimum should send source info
-        return JSONValue.toJSONString(obj);
-    }
-
-    /**
      * Gets the location of the external resources for this worker on the
      * local filesystem. These external resources typically include bolts implemented
      * in other languages, such as Ruby or Python.
@@ -323,24 +198,6 @@
     public String getPIDDir() {
         return _pidDir;
     }
-
-    /**
-     * Gets a map from task id to component id.
-     */
-    public Map<Integer, String> getTaskToComponent() {
-        return _taskToComponent;
-    }
-    
-    /**
-     * Gets a list of all component ids in this topology
-     */
-    public Set<String> getComponentIds() {
-        return ThriftTopologyUtils.getComponentIds(getRawTopology());
-    }
-
-    public ComponentCommon getComponentCommon(String componentId) {
-        return ThriftTopologyUtils.getComponentCommon(getRawTopology(), componentId);
-    }
     
     public void setTaskData(Object data) {
         _taskData = data;
@@ -350,22 +207,6 @@
         return _taskData;
     }
     
-    public int maxTopologyMessageTimeout() {
-        Integer max = Utils.getInt(_stormConf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS));
-        for(String spout: getRawTopology().get_spouts().keySet()) {
-            ComponentCommon common = getComponentCommon(spout);
-            String jsonConf = common.get_json_conf();
-            if(jsonConf!=null) {
-                Map conf = (Map) JSONValue.parse(jsonConf);
-                Object comp = conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS);
-                if(comp!=null) {
-                    max = Math.max(Utils.getInt(comp), max);
-                }
-            }
-        }
-        return max;
-    }
-    
     public Integer getThisWorkerPort() {
         return _workerPort;
     }