Adding an 'args' parameter to the plugin framework.

adam-cli/src/main/scala/org/bdgenomics/adam/cli/PluginExecutor.scala

@@ -54,6 +54,9 @@ class PluginExecutorArgs extends Args4jBase with SparkArgs with ParquetArgs {
 
   @Args4jOption(name = "-access_control", usage = "Class for access control")
   var accessControl: String = "org.bdgenomics.adam.plugins.EmptyAccessControl"
+
+  @Args4jOption(name = "-plugin_args", usage = "Arguments for the plugin")
+  var pluginArgs: String = ""
 }
 
 class PluginExecutor(protected val args: PluginExecutorArgs) extends ADAMSparkCommand[PluginExecutorArgs] {
@@ -111,7 +114,7 @@ class PluginExecutor(protected val args: PluginExecutorArgs) extends ADAMSparkCo
       case Some(filterFunc) => firstRdd.filter(filterFunc)
     }
 
-    val results = plugin.run(sc, input)
+    val results = plugin.run(sc, input, args.pluginArgs)
 
     output(sc, results)
   }

adam-cli/src/test/scala/org/bdgenomics/adam/cli/PluginExecutorSuite.scala

@@ -42,4 +42,29 @@ class PluginExecutorSuite extends FunSuite {
     // Make sure that 10 lines are output
     assert(outputString.lines.size === 10)
   }
+
+  test("takeN works correctly on example SAM with arg of '3'") {
+
+    val args = new PluginExecutorArgs()
+    args.plugin = "org.bdgenomics.adam.plugins.TakeNPlugin"
+    args.pluginArgs = "3"
+
+    val stream = Thread.currentThread().getContextClassLoader.getResourceAsStream("reads12.sam")
+    val file = File.createTempFile("reads12", ".sam")
+    val os = new FileOutputStream(file)
+    val bytes = new Array[Byte](stream.available())
+    stream.read(bytes)
+    os.write(bytes)
+    args.input = file.getAbsolutePath
+
+    val pluginExecutor = new PluginExecutor(args)
+
+    val bytesWritten = new ByteArrayOutputStream()
+    scala.Console.withOut(bytesWritten)(pluginExecutor.run())
+
+    val outputString = bytesWritten.toString
+
+    // Make sure that 3 lines are output
+    assert(outputString.lines.size === 3)
+  }
 }

adam-core/src/main/scala/org/bdgenomics/adam/plugins/ADAMPlugin.scala

@@ -42,5 +42,5 @@ trait ADAMPlugin[Input, Output] {
   /**
    * Method to create the transformations on the RDD.
    */
-  def run(sc: SparkContext, recs: RDD[Input]): RDD[Output]
+  def run(sc: SparkContext, recs: RDD[Input], args: String): RDD[Output]
 }

adam-core/src/test/scala/org/bdgenomics/adam/plugins/Take10Plugin.scala

@@ -24,7 +24,7 @@ class Take10Plugin extends ADAMPlugin[ADAMRecord, ADAMRecord] with Serializable
   override def projection: Option[Schema] = None
   override def predicate: Option[(ADAMRecord) => Boolean] = None
 
-  override def run(sc: SparkContext, recs: RDD[ADAMRecord]): RDD[ADAMRecord] = {
+  override def run(sc: SparkContext, recs: RDD[ADAMRecord], args: String): RDD[ADAMRecord] = {
     sc.parallelize(recs.take(10))
   }
 }

adam-core/src/test/scala/org/bdgenomics/adam/plugins/TakeNPlugin.scala

@@ -0,0 +1,31 @@
+/**
+ * Copyright 2014 Genome Bridge LLC
+ *
+ * Licensed 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 org.bdgenomics.adam.plugins
+
+import org.bdgenomics.adam.avro.ADAMRecord
+import org.apache.avro.Schema
+import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkContext
+
+class TakeNPlugin extends ADAMPlugin[ADAMRecord, ADAMRecord] with Serializable {
+  override def projection: Option[Schema] = None
+  override def predicate: Option[(ADAMRecord) => Boolean] = None
+
+  override def run(sc: SparkContext, recs: RDD[ADAMRecord], args: String): RDD[ADAMRecord] = {
+    val n = args.toInt
+    sc.parallelize(recs.take(n))
+  }
+}