/**
 * Copyright (c) 2011, Cloudera, Inc. All Rights Reserved.
 *
 * Cloudera, Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"). You may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied. See the License for
 * the specific language governing permissions and limitations under the
 * License.
 */
package com.cloudera.crunch.tool;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.util.Tool;

import com.cloudera.crunch.PCollection;
import com.cloudera.crunch.PTable;
import com.cloudera.crunch.Pipeline;
import com.cloudera.crunch.Source;
import com.cloudera.crunch.TableSource;
import com.cloudera.crunch.Target;
import com.cloudera.crunch.impl.mr.MRPipeline;
import com.cloudera.crunch.io.At;
import com.cloudera.crunch.io.From;
import com.cloudera.crunch.io.To;

/**
 * An extension of the {@code Tool} interface that creates a {@code Pipeline}
 * instance and provides methods for working with the Pipeline from inside of
 * the Tool's run method.
 *
 */
public abstract class CrunchTool extends Configured implements Tool {

  protected static final From from = new From();
  protected static final To to = new To();
  protected static final At at = new At();
  
  private Pipeline pipeline;

  public CrunchTool() throws IOException {
	this.pipeline = new MRPipeline(getClass());
  }
  
  @Override
  public void setConf(Configuration conf) {
	super.setConf(conf);
	pipeline.setConfiguration(conf);
  }
  
  @Override
  public Configuration getConf() {
	return pipeline.getConfiguration();
  }
  
  protected <T> PCollection<T> read(Source<T> source) {
	return pipeline.read(source);
  }
  
  protected <K, V> PTable<K, V> read(TableSource<K, V> tableSource) {
	return pipeline.read(tableSource);
  }
  
  protected PCollection<String> readTextFile(String pathName) {
	return pipeline.readTextFile(pathName);
  }
  
  protected void write(PCollection<?> pcollection, Target target) {
	pipeline.write(pcollection, target);
  }
  
  protected void writeTextFile(PCollection<?> pcollection, String pathName) {
	pipeline.writeTextFile(pcollection, pathName);
  }
  
  protected void run() {
	pipeline.run();
  }
  
  protected void done() {
	pipeline.done();
  }
}

@@ -973,16 +973,65 @@
 Target;%0A
+import com.cloudera.crunch.impl.mem.MemPipeline;%0A
 import c
@@ -1658,19 +1658,141 @@
 this
-.pipeline =
+(false);%0A  %7D%0A  %0A  public CrunchTool(boolean inMemory) throws IOException %7B%0A    this.pipeline = inMemory ? MemPipeline.getInstance() :
  new
@@ -1815,16 +1815,18 @@
 lass());
+  
 %0A  %7D%0A  %0A
@@ -2038,24 +2038,86 @@
 %7D%0A  %0A  p
-rotected
+ublic void enableDebug() %7B%0A    pipeline.enableDebug();%0A  %7D%0A  %0A  public
  %3CT%3E PCo
@@ -2194,24 +2194,21 @@
 %7D%0A  %0A  p
-rotected
+ublic
  %3CK, V%3E 
@@ -2304,24 +2304,21 @@
 %7D%0A  %0A  p
-rotected
+ublic
  PCollec
@@ -2409,32 +2409,29 @@
 ;%0A  %7D%0A  %0A  p
-rotected
+ublic
  void write(
@@ -2519,32 +2519,29 @@
 ;%0A  %7D%0A  %0A  p
-rotected
+ublic
  void writeT
@@ -2649,32 +2649,29 @@
 ;%0A  %7D%0A  %0A  p
-rotected
+ublic
  void run() 
@@ -2699,24 +2699,21 @@
 %7D%0A  %0A  p
-rotected
+ublic
  void do

Add in-memory option to CrunchTool