Merge pull request #4 from nextflow-io/3-create-a-channel-extension-point

feature: create a hello channel extension point

Author: pditommaso

README.md

@@ -3,6 +3,9 @@
 This project shows how to implement a simple Nextflow plugin named `nf-hello` that intercepts 
 workflow execution events to print a message when the execution starts and on workflow completion.
 
+Also, this plugin enriches the `channel` with a `producer` a `consumer` methods (`sayHello` and `goodbye`)
+allowing to include them into the script 
+
 ## Plugin assets 
                     
 - `settings.gradle`
@@ -38,6 +41,18 @@ workflow execution events to print a message when the execution starts and on wo
 
     The plugin unit tests. 
 
+## ExtensionPointS
+
+ExtensionPoint is the basic interface who use nextflow-core to integrate plugins into it.
+It's only a basic interface and serves as starting point for more specialized extensions. 
+
+Among others, nextflow-core integrate following sub ExtensionPointS:
+
+- `TraceObserverFactory` to provide a list of TraceObserverS 
+- `ChannelExtensionPoint` to enrich the channel with custom methods
+
+In this plugin you can find examples for both of them
+
 ## Compile & run unit tests 
 
 Run the following command in the project root directory (ie. where the file `settings.gradle` is located):

plugins/nf-hello/build.gradle

@@ -53,15 +53,15 @@ sourceSets {
 
 dependencies {
     // This dependency is exported to consumers, that is to say found on their compile classpath.
-    compileOnly 'io.nextflow:nextflow:21.04.0'
+    compileOnly 'io.nextflow:nextflow:22.04.0'
     compileOnly 'org.slf4j:slf4j-api:1.7.10'
     compileOnly 'org.pf4j:pf4j:3.4.1'
     // add here plugins depepencies
 
     // test configuration
     testImplementation "org.codehaus.groovy:groovy:3.0.8"
     testImplementation "org.codehaus.groovy:groovy-nio:3.0.8"
-    testImplementation 'io.nextflow:nextflow:21.04.0'
+    testImplementation 'io.nextflow:nextflow:22.04.0'
     testImplementation ("org.codehaus.groovy:groovy-test:3.0.8") { exclude group: 'org.codehaus.groovy' }
     testImplementation ("cglib:cglib-nodep:3.3.0")
     testImplementation ("org.objenesis:objenesis:3.1")

plugins/nf-hello/src/main/nextflow/hello/HelloExtension.groovy

@@ -0,0 +1,119 @@
+package nextflow.hello
+
+import groovy.util.logging.Slf4j
+import groovyx.gpars.dataflow.DataflowReadChannel
+import groovyx.gpars.dataflow.DataflowWriteChannel
+import groovyx.gpars.dataflow.expression.DataflowExpression
+import nextflow.Channel
+import nextflow.Global
+import nextflow.Session
+import nextflow.extension.ChannelExtensionPoint
+import nextflow.extension.CH
+import nextflow.NF
+import nextflow.extension.DataflowHelper
+import nextflow.plugin.Scoped
+
+import java.util.concurrent.CompletableFuture
+
+/**
+ * @author : jorge <[email protected]>
+ *
+ */
+@Slf4j
+@Scoped('hello')
+class HelloExtension extends ChannelExtensionPoint{
+
+    /*
+     * A session hold information about current execution of the script
+     */
+    private Session session
+
+    /*
+     * nf-core initializes the plugin once loaded and session is ready
+     * @param session
+     */
+    @Override
+    protected void init(Session session) {
+        this.session = session
+    }
+
+    /*
+     * reverse is a `producer` method and will be available to the script because:
+     *
+     * - it's public
+     * - it returns a DataflowWriteChannel
+     *
+     * nf-core will inspect the extension class and allow the script to call all these kind of methods
+     *
+     * the method can require arguments but it's not mandatory, it depends of the business logic of the method
+     *
+     * business logic can write into the channel once ready and values will be consumed from it
+     */
+    DataflowWriteChannel reverse(String message) {
+        createReverseChannel(message)
+    }
+
+    static String goodbyeMessage
+
+    /*
+    * goodbye is a `consumer` method as it receives values from a channel to perform some logic.
+    *
+    * Consumer methods are introspected by nextflow-core and include into the DSL if the method:
+    *
+    * - it's public
+    * - it returns a DataflowWriteChannel
+    * - it has only one arguments of DataflowReadChannel class
+    *
+    * a consumer method needs to proporcionate 2 closures:
+    * - a closure to consume items (one by one)
+    * - a finalizer closure
+    *
+    * in this case `goodbye` will consume a message and will store it as an upper case
+    */
+    DataflowWriteChannel goodbye(DataflowReadChannel source) {
+        final target = CH.createBy(source)
+        final next = {
+            goodbyeMessage = "$it".toString().toUpperCase()
+            target.bind(it)
+        }
+        final done = {
+            target.bind(Channel.STOP)
+        }
+        DataflowHelper.subscribeImpl(source, [onNext: next, onComplete: done])
+        target
+    }
+
+    protected DataflowWriteChannel createReverseChannel(final String message){
+        final channel = CH.create()
+        if( NF.isDsl2() ){
+            session.addIgniter { ->
+                businessLogicHere(channel, message)
+            }
+        }else{
+            businessLogicHere(channel, message)
+        }
+        channel
+    }
+
+    /*
+    * businessLogicHere will send, across the channel, the message reversed
+    * and after will send an STOP signal to let know the channel it has been finished
+    */
+    protected static businessLogicHere(final DataflowWriteChannel channel, final String message){
+        def future = CompletableFuture.runAsync({
+            channel.bind(message.reverse())
+            channel.bind(Channel.STOP)
+        })
+        future.exceptionally(this.&handlerException)
+    }
+
+    /*
+    * an util class to trace exceptions
+    */
+    static private void handlerException(Throwable e) {
+        final error = e.cause ?: e
+        log.error(error.message, error)
+        final session = Global.session as Session
+        session?.abort(error)
+    }
+}

plugins/nf-hello/src/main/nextflow/hello/HelloPlugin.groovy

@@ -18,6 +18,7 @@ package nextflow.hello
 
 import groovy.transform.CompileStatic
 import nextflow.plugin.BasePlugin
+import nextflow.plugin.Scoped
 import org.pf4j.PluginWrapper
 
 /**

plugins/nf-hello/src/resources/META-INF/MANIFEST.MF

@@ -1,6 +1,6 @@
 Manifest-Version: 1.0
 Plugin-Id: nf-hello
-Plugin-Version: 0.1.0
+Plugin-Version: 0.2.0
 Plugin-Class: nextflow.hello.HelloPlugin
 Plugin-Provider: nextflow
-Plugin-Requires: >=21.04.0
+Plugin-Requires: >=22.04.0

plugins/nf-hello/src/resources/META-INF/extensions.idx

@@ -1 +1,2 @@
 nextflow.hello.HelloFactory
+nextflow.hello.HelloExtension

plugins/nf-hello/src/test/nextflow/hello/ChannelExtensionHelloTest.groovy

@@ -0,0 +1,53 @@
+package nextflow.hello
+
+import groovyx.gpars.dataflow.DataflowQueue
+import nextflow.Channel
+import nextflow.Session
+import nextflow.extension.ChannelExtensionDelegate
+import spock.lang.Specification
+
+
+/**
+ * @author : jorge <[email protected]>
+ *
+ */
+class ChannelExtensionHelloTest extends Specification{
+
+    def "should create a channel from hello"(){
+
+        given:
+        def session = Mock(Session)
+
+        and:
+        def helloExtension = new HelloExtension(); helloExtension.init(session)
+
+        when:
+        def result = helloExtension.reverse("Hi")
+
+        then:
+        result.val == 'iH'
+        result.val == Channel.STOP
+    }
+
+    def "should consume a message from script"(){
+
+        given:
+        def session = Mock(Session)
+
+        and:
+        def helloExtension = new HelloExtension(); helloExtension.init(session)
+
+        and:
+        def ch = new DataflowQueue()
+        ch.bind('Goodbye folks')
+        ch.bind( Channel.STOP )
+
+        when:
+        def result = helloExtension.goodbye(ch)
+
+        then:
+        result.val == 'Goodbye folks'
+        result.val == Channel.STOP
+        helloExtension.goodbyeMessage == 'Goodbye folks'.toUpperCase()
+    }
+}

plugins/nf-hello/src/test/nextflow/hello/HelloDslTest.groovy

@@ -0,0 +1,49 @@
+package nextflow.hello
+
+import nextflow.Channel
+import nextflow.extension.ChannelExtensionDelegate
+import nextflow.plugin.Plugins
+import spock.lang.Specification
+import spock.lang.Timeout
+
+
+/**
+ * @author : jorge <[email protected]>
+ *
+ */
+@Timeout(10)
+class HelloDslTest extends Specification{
+
+    def setup () {
+        ChannelExtensionDelegate.reloadExtensionPoints()
+    }
+
+    def 'should perform a hi and create a channel' () {
+        when:
+        def SCRIPT = '''
+            channel.hello.reverse('hi!') 
+            '''
+        and:
+        def result = new MockScriptRunner([:]).setScript(SCRIPT).execute()
+        then:
+        result.val == '!ih'
+        result.val == Channel.STOP
+    }
+
+    def 'should store a goodbye' () {
+        when:
+        def SCRIPT = '''
+            channel
+                .of('Bye bye folks')
+                .goodbye() 
+            '''
+        and:
+        def result = new MockScriptRunner([:]).setScript(SCRIPT).execute()
+        then:
+        result.val == 'Bye bye folks'
+        result.val == Channel.STOP
+
+        and:
+        HelloExtension.goodbyeMessage == 'Bye bye folks'.toUpperCase()
+    }
+}