Make the plugin functional

.github/workflows/build.yml

@@ -0,0 +1,47 @@
+name: build
+
+env:
+  java_version: 17
+
+on:
+  push:
+    branches:
+      - '*'
+    tags-ignore:
+      - '*'
+  pull_request:
+    branches:
+      - '*'
+jobs:
+  build:
+    name: Build
+    if: "!contains(github.event.head_commit.message, '[ci skip]')"
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    strategy:
+      fail-fast: false
+
+    steps:
+      - name: Environment
+        run: env | sort
+
+      - name: Checkout
+        uses: actions/checkout@v1
+        with:
+          fetch-depth: 1
+          submodules: true
+
+      - name: Setup Java ${{ env.java_version }}
+        uses: actions/setup-java@v1
+        with:
+          java-version: ${{ env.java_version }}
+          architecture: x64
+          distribution: 'temurin'
+
+      - name: Compile
+        run: ./gradlew assemble
+
+      - name: Tests
+        run: ./gradlew check
+        env:
+          GRADLE_OPTS: '-Dorg.gradle.daemon=false'

.github/workflows/release.yml

@@ -0,0 +1,29 @@
+name: release
+
+env:
+  java_version: 17
+
+on:
+  push:
+    tags:
+      - "[0-9]+.[0-9]+.[0-9]+"
+
+permissions:
+  contents: write
+
+jobs:
+  publish-gpr:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up JDK ${{ env.java_version }} for x64
+        uses: actions/setup-java@v3
+        with:
+          java-version: ${{ env.java_version }}
+          distribution: "temurin"
+          architecture: x64
+
+      - name: Build and upload artifact and release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: ./gradlew :plugins:nf-kafka:upload

.gitignore

@@ -3,6 +3,9 @@
 .idea
 .nextflow
 
+# Ignore dev secrets
+gradle.properties
+
 # Ignore Gradle build output directory
 build
 work

README.md

@@ -17,25 +17,27 @@ Make sure to have Nextflow `22.10.0` or later. Add the following snippet to your
 
 ```
 plugins {
-  id 'nf-kafa@0.0.1'
+  id 'nf-kafka@0.0.2'
 }
 ```
 
 ## Configuration
 
 The plugin configuration is specified using the `kafka` scope: 
 
-| Config option 	               | Description 	                |
-|-------------------------------|---	                        |
-| `kafka.url`                   | The connection url. 
-| `kafka.group`                 | The group where the plugin will be attached.
+| Config option 	              | Description 	                                          | Mandatory |
+| ----------------------------- | ------------------------------------------------------- | --------- |
+| `kafka.url`                   | The connection url.                                     | yes       |
+| `kafka.group`                 | The group where the plugin will be attached.            | yes       |
+| `kafka.pollTimeout`           | The time that consumer will wait to consume from topic. | no        |
 
 For example:
 
 ```
 kafka {
   url = 'localhost:902'
   group = 'group'
+  pollTimeout = 2500 // ms
 }
 ```
 
@@ -124,7 +126,7 @@ services:
 
 ```
 plugins{
-   id '[email protected].1'
+   id '[email protected].2'
 }
 
 kafka{    

plugins/build.gradle

@@ -20,10 +20,10 @@ plugins {
     id "idea"
 }
 
-ext.github_organization = 'nextflow-io'
-ext.github_username = project.findProperty('github_username') ?: 'pditommaso'
+ext.github_owner = project.findProperty('github_owner') ?: System.getenv('GITHUB_REPOSITORY_OWNER')
+ext.github_user = project.findProperty('github_user') ?: System.getenv('GITHUB_TRIGGERING_ACTOR')
+ext.github_repository = project.findProperty('github_repository') ?: System.getenv('GITHUB_REPOSITORY')
 ext.github_access_token = project.findProperty('github_access_token') ?: System.getenv('GITHUB_TOKEN')
-ext.github_commit_email = project.findProperty('github_commit_email') ?: '[email protected]'
 
 jar.enabled = false
 
@@ -86,7 +86,7 @@ subprojects {
               {
                 "version": "${project.version}",
                 "date": "${timestamp}",
-                "url": "https://github.com/${github_organization}/${project.name}/releases/download/${project.version}/${project.name}-${project.version}.zip",
+                "url": "https://github.com/${github_repository}/releases/download/${project.version}/${project.name}-${project.version}.zip",
                 "requires": "${metaFromManifest('Plugin-Requires',file('src/resources/META-INF/MANIFEST.MF'))}",
                 "sha512sum": "${computeSha512(zip)}"
               }
@@ -128,8 +128,8 @@ subprojects {
                                       "$buildDir/libs/${project.name}-${project.version}-meta.json" ]}
         release = providers.provider { project.version }
         repo = providers.provider { project.name }
-        owner = github_organization
-        userName = github_username
+        owner = github_owner
+        userName = github_user
         authToken = github_access_token
         skipExisting = true
     }
@@ -143,11 +143,11 @@ classes.dependsOn subprojects.copyPluginLibs
 /*
  * Merge and publish the plugins index file
  */
-task publishIndex( type: io.nextflow.gradle.tasks.GithubRepositoryPublisher ) {
-    indexUrl = 'https://github.com/nextflow-io/plugins/main/plugins.json'
-    repos = ['nf-sqldb']
-    owner = github_organization
-    githubUser = github_username
-    githubEmail = github_commit_email
-    githubToken = github_access_token
-}
+// task publishIndex( type: io.nextflow.gradle.tasks.GithubRepositoryPublisher ) {
+//     indexUrl = 'https://github.com/nextflow-io/plugins/main/plugins.json'
+//     repos = 
+//     owner = 
+//     githubUser = 
+//     githubEmail = 
+//     githubToken = 
+// }

plugins/nf-kafka/build.gradle

@@ -63,7 +63,7 @@ dependencies {
     compileOnly 'org.slf4j:slf4j-api:1.7.10'
     compileOnly 'org.pf4j:pf4j:3.4.1'
 
-    implementation 'org.apache.kafka:kafka-clients:2.6.3'
+    implementation 'org.apache.kafka:kafka-clients:3.7.1'
 
     testImplementation "io.nextflow:nextflow:$nextflowVersion"
     testImplementation "org.codehaus.groovy:groovy:3.0.10"

plugins/nf-kafka/src/main/nextflow/events/ChannelKafkaExtension.groovy

@@ -55,13 +55,13 @@ class ChannelKafkaExtension extends PluginExtensionPoint {
     }
 
     @Factory
-    DataflowWriteChannel fromTopic(String topic, Duration duration=Duration.ofSeconds(1)) {
-        topicToChannel(topic, duration, false)
+    DataflowWriteChannel fromTopic(String topic) {
+        topicToChannel(topic, false)
     }
 
     @Factory
-    DataflowWriteChannel watchTopic(String topic, Duration duration=Duration.ofSeconds(1)) {
-        topicToChannel(topic, duration, true)
+    DataflowWriteChannel watchTopic(String topic, Map until) {
+        topicToChannel(topic, true, until)
     }
 
     @Operator
@@ -91,7 +91,7 @@ class ChannelKafkaExtension extends PluginExtensionPoint {
                 .publishMessage([key, message])
     }
 
-    private DataflowWriteChannel topicToChannel(String topic, Duration duration, boolean listening){
+    private DataflowWriteChannel topicToChannel(String topic, boolean listening, Map until=null){
         final channel = CH.create()
 
         final handler = new TopicHandler()
@@ -100,8 +100,9 @@ class ChannelKafkaExtension extends PluginExtensionPoint {
                 .withGroup(config.group)
                 .withTopic(topic)
                 .withListening(listening)
+                .withUntil(until)
                 .withTarget(channel)
-                .withDuration(duration)
+                .withDuration(Duration.ofMillis(config.pollTimeout))
         if(NF.dsl2) {
             session.addIgniter {-> handler.perform() }
         }

plugins/nf-kafka/src/main/nextflow/events/KafkaPlugin.groovy

@@ -34,15 +34,8 @@ class KafkaPlugin extends BasePlugin {
         super(wrapper)
     }
 
-    @Override
-    void start() {
-        super.start()
-    }
-
     @Override
     void stop() {
         ThreadFactory.instance.shutdownExecutors()
-        super.stop()
     }
-
 }

plugins/nf-kafka/src/main/nextflow/events/kafa/KafkaConfig.groovy

@@ -13,13 +13,16 @@ import groovy.transform.ToString
 @CompileStatic
 class KafkaConfig {
 
-    private String url
-    private String group
+    private String url // mandatory
+    private String group // mandatory
+    private Integer pollTimeout // ms
 
     KafkaConfig(Map map){
         def config = map ?: Collections.emptyMap()
-        url = config.url
-        group = config.group
+
+        url = config.url ?: KafkaConfig.missingConf("url")
+        group = config.group ?: KafkaConfig.missingConf("group")
+        pollTimeout = KafkaConfig.parsePollTimeout(config.pollTimeout)
     }
 
     String getUrl(){
@@ -29,5 +32,31 @@ class KafkaConfig {
     String getGroup(){
         group
     }
+
+    Integer getpollTimeout(){
+        pollTimeout
+    }
+
+    static private Integer parsePollTimeout(confPollTimeout) {
+        Integer pollTimeout = 1000 // default value
+        if (confPollTimeout) {
+            pollTimeout = confPollTimeout
+                            .toString()
+                            .toInteger()
+        } 
+        return pollTimeout
+    }
+
+    static private missingConf(String conf) {
+        throw new KafkaConfigException(conf);
+    }
 
 }
+
+class KafkaConfigException extends Exception {
+    public KafkaConfigException(String conf) {
+        super(
+            "the configuration of the $conf is mandatory, please configure it in the scope of the kafka plugin conf"
+        );
+    }
+}

plugins/nf-kafka/src/main/nextflow/events/kafa/TopicHandler.groovy

@@ -3,6 +3,7 @@ package nextflow.events.kafa
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import groovyx.gpars.dataflow.DataflowWriteChannel
+import nextflow.Channel
 import nextflow.Session
 import nextflow.events.KafkaPlugin
 import nextflow.util.ThreadPoolBuilder
@@ -31,6 +32,7 @@ class TopicHandler {
     private String topic
     private Duration duration
     private boolean listening
+    private Map until
     private Session session
 
     KafkaConsumer<String, String> consumer
@@ -60,6 +62,11 @@ class TopicHandler {
         this
     }
 
+    TopicHandler withUntil(Map until){
+        this.until = until
+        this
+    }
+
     TopicHandler withTarget(DataflowWriteChannel channel) {
         this.target = channel
         return this
@@ -77,6 +84,7 @@ class TopicHandler {
         }else{
             consume()
             closeConsumer()
+            this.target.bind(Channel.STOP)
         }
         return this
     }
@@ -105,8 +113,15 @@ class TopicHandler {
         try {
             final records = consumer.poll(duration)
             records.each {
-                target << [ it.key(), it.value()]
+                Map record = [ key:it.key(), value:it.value() ]
+                target << record
+
+                if (record == this.until) {
+                    Thread.currentThread().interrupt()
+                    target.bind(Channel.STOP)
+                }
             }
+
         }catch(Exception e){
             log.error "Exception reading kafka topic $topic",e
         }

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

@@ -2,5 +2,5 @@ Manifest-Version: 1.0
 Plugin-Class: nextflow.events.KafkaPlugin
 Plugin-Id: nf-kafka
 Plugin-Provider: Seqera Labs
-Plugin-Version: 0.0.1
+Plugin-Version: 0.0.2
 Plugin-Requires: >=22.10.0