<commit_before>package nextflow.extension

import groovyx.gpars.dataflow.DataflowQueue
import nextflow.Channel
import spock.lang.Specification


/**
 *
 * @author <NAME> <<EMAIL>>
 */
class DataflowExtensionsTest extends Specification {

    def testHandlerNames() {

        when:
        DataflowExtensions.checkSubscribeHandlers( [:] )
        then:
        thrown(IllegalArgumentException)

        when:
        DataflowExtensions.checkSubscribeHandlers( [ onNext:{}] )
        then:
        true

        when:
        DataflowExtensions.checkSubscribeHandlers( [ onNext:{}, xxx:{}] )
        then:
        thrown(IllegalArgumentException)

        when:
        DataflowExtensions.checkSubscribeHandlers( [ xxx:{}] )
        then:
        thrown(IllegalArgumentException)


    }

    def testGrep() {

        when:
        def channel = new DataflowQueue()
        channel << 'Hello' << 'Halo' << 'Ciao' << 'Hola' << 'Bonjour'
        def result = channel.grep( ~/^H.*/ )
        then:
        result.val == 'Hello'
        result.val == 'Halo'
        result.val == 'Hola'


        when:
        def channel2 = new DataflowQueue()
        channel2 << 'Hello' << 1 << 'Ciao' << 2 << 'Bonjour'
        def result2 = channel2.grep(Number)
        then:
        result2.val == 1
        result2.val == 2

    }

    def testSubscribe() {

        when:
        def channel = Channel.create()
        int count = 0
        channel.subscribe { count++; } << 1 << 2 << 3
        sleep(100)

        then:
        count == 3

    }

    def testSubscribe1() {

        when:
        def count = 0
        def done = false
        Channel.from(1,2,3).subscribe onNext:  { count++ }, onComplete: { done = true }
        sleep 100
        then:
        done
        count == 3

    }

    def testSubscribe2() {

        when:
        def count = 0
        def done = false
        Channel.just(1).subscribe onNext:  { count++ }, onComplete: { done = true }
        sleep 100
        then:
        done
        count == 1

    }

    def testSubscribeError() {

        when:
        int next=0
        int error=0
        int complete=0
        Channel
                .from( 2,1,0,3,3 )
                .subscribe onNext: { println it/it; next++ }, onError: { error++ }, onComplete: { complete++ }
        sleep 100

        then:
        // on the third iteration raise an exception
        // next have to be equals to two
        next == 2
        // error have to be invoked one time
        error == 1
        // complete never
        complete == 0

    }


    def testDoFinally() {

        when:
        def flag
        Channel.from(1,2,3).doFinally { flag = 1 }
        sleep 100
        then:
        flag == 1

    }

    def testMap() {
        when:
        def result = Channel.from(1,2,3).map { "Hello $it" }
        then:
        result.val == 'Hello 1'
        result.val == 'Hello 2'
        result.val == 'Hello 3'
        result.val == Channel.STOP
    }

    def testSkip() {

        when:
        def result = Channel.from(1,2,3).map { it == 2 ? Channel.VOID : "Hello $it" }
        then:
        result.val == 'Hello 1'
        result.val == 'Hello 3'
        result.val == Channel.STOP

    }

    def testMapWithIndex() {
        when:
        def result = Channel.from('a','b','c').mapWithIndex { item, index -> [item, index]}
        then:
        result.val == [ 'a', 0 ]
        result.val == [ 'b', 1 ]
        result.val == [ 'c', 2 ]
        result.val == Channel.STOP
    }

    def testMapMany () {

        when:
        def result = Channel.from(1,2,3).mapMany { it -> [it, it*2] }
        then:
        result.val == 1
        result.val == 2
        result.val == 2
        result.val == 4
        result.val == 3
        result.val == 6
        result.val == Channel.STOP
    }

    def testMapManyWithHashArray () {

        when:
        def result = Channel.from(1,2,3).mapMany { it -> [ k: it, v: it*2] }
        then:
        result.val == new MapEntry('k',1)
        result.val == new MapEntry('v',2)
        result.val == new MapEntry('k',2)
        result.val == new MapEntry('v',4)
        result.val == new MapEntry('k',3)
        result.val == new MapEntry('v',6)
        result.val == Channel.STOP

    }



    def testReduce() {

        when:
        def channel = Channel.create()
        def result = channel.reduce { a, e -> a += e }
        channel << 1 << 2 << 3 << 4 << 5 << Channel.STOP
        then:
        result.getVal() == 15

        when:
        channel = Channel.create()
        result = channel.reduce { a, e -> a += e }
        channel << 99 << Channel.STOP
        then:
        result.getVal() == 99

        when:
        channel = Channel.create()
        result = channel.reduce { a, e -> a += e }
        channel << Channel.STOP
        then:<commit_msg>Improved test<commit_after>package nextflow.extension

import groovyx.gpars.dataflow.DataflowQueue
import nextflow.Channel
import spock.lang.Specification


/**
 *
 * @author <NAME> <<EMAIL>>
 */
class DataflowExtensionsTest extends Specification {

    def testHandlerNames() {

        when:
        DataflowExtensions.checkSubscribeHandlers( [:] )
        then:
        thrown(IllegalArgumentException)

        when:
        DataflowExtensions.checkSubscribeHandlers( [ onNext:{}] )
        then:
        true

        when:
        DataflowExtensions.checkSubscribeHandlers( [ onNext:{}, xxx:{}] )
        then:
        thrown(IllegalArgumentException)

        when:
        DataflowExtensions.checkSubscribeHandlers( [ xxx:{}] )
        then:
        thrown(IllegalArgumentException)


    }

    def testGrep() {

        when:
        def channel = new DataflowQueue()
        channel << 'Hello' << 'Halo' << 'Ciao' << 'Hola' << 'Bonjour'
        def result = channel.grep( ~/^H.*/ )
        then:
        result.val == 'Hello'
        result.val == 'Halo'
        result.val == 'Hola'


        when:
        def channel2 = new DataflowQueue()
        channel2 << 'Hello' << 1 << 'Ciao' << 2 << 'Bonjour'
        def result2 = channel2.grep(Number)
        then:
        result2.val == 1
        result2.val == 2

    }

    def testSubscribe() {

        when:
        def channel = Channel.create()
        int count = 0
        channel.subscribe { count++; } << 1 << 2 << 3
        sleep(100)
        then:
        count == 3

        when:
        count = 0
        channel = Channel.from(1,2,3,4)
        channel.subscribe { count++; }
        sleep(100)
        then:
        count == 4


        when:
        int a=0
        int b=0
        def channel3 = Channel.from(1,2,3,4,5)
        channel3.subscribe { if(it%2 != 0) a++ }
        channel3.subscribe { if(it%2 == 0) b++; println ">> $it"  }
        //channel3 << 1 << 2 << 3 << 4 << 5
        sleep(100)
        then:
        a == 3
        b == 2


    }

    def testSubscribe1() {

        when:
        def count = 0
        def done = false
        Channel.from(1,2,3).subscribe onNext:  { count++ }, onComplete: { done = true }
        sleep 100
        then:
        done
        count == 3

    }

    def testSubscribe2() {

        when:
        def count = 0
        def done = false
        Channel.just(1).subscribe onNext:  { count++ }, onComplete: { done = true }
        sleep 100
        then:
        done
        count == 1

    }

    def testSubscribeError() {

        when:
        int next=0
        int error=0
        int complete=0
        Channel
                .from( 2,1,0,3,3 )
                .subscribe onNext: { println it/it; next++ }, onError: { error++ }, onComplete: { complete++ }
        sleep 100

        then:
        // on the third iteration raise an exception
        // next have to be equals to two
        next == 2
        // error have to be invoked one time
        error == 1
        // complete never
        complete == 0

    }


    def testDoFinally() {

        when:
        def flag
        Channel.from(1,2,3).doFinally { flag = 1 }
        sleep 100
        then:
        flag == 1

    }

    def testMap() {
        when:
        def result = Channel.from(1,2,3).map { "Hello $it" }
        then:
        result.val == 'Hello 1'
        result.val == 'Hello 2'
        result.val == 'Hello 3'
        result.val == Channel.STOP
    }

    def testSkip() {

        when:
        def result = Channel.from(1,2,3).map { it == 2 ? Channel.VOID : "Hello $it" }
        then:
        result.val == 'Hello 1'
        result.val == 'Hello 3'
        result.val == Channel.STOP

    }

    def testMapWithIndex() {
        when:
        def result = Channel.from('a','b','c').mapWithIndex { item, index -> [item, index]}
        then:
        result.val == [ 'a', 0 ]
        result.val == [ 'b', 1 ]
        result.val == [ 'c', 2 ]
        result.val == Channel.STOP
    }

    def testMapMany () {

        when:
        def result = Channel.from(1,2,3).mapMany { it -> [it, it*2] }
        then:
        result.val == 1
        result.val == 2
        result.val == 2
        result.val == 4
        result.val == 3
        result.val == 6
        result.val == Channel.STOP
    }

    def testMapManyWithHashArray () {

        when:
        def result = Channel.from(1,2,3).mapMany { it -> [ k: it, v: it*2] }
        then:
        result.val == new MapEntry('k',1)
        result.val == new MapEntry('v',2)
        result.val == new MapEntry('k',2)
        result.val == new MapEntry('v',4)
        result.val == new MapEntry('k',3)
        result.val == new MapEntry('v',6)
        result.val == Channel.STOP

    }



    def testReduce() {

        when:
        def channel = Channel.create()
        def result = channel.reduce { a, e -> a += e }
        channel << 1 << 2 << 3 << 4 << 5 << Channel.STOP
        then:
        result.getVal() == 15


        when:
        channel = Channel.from(1,2,3,4,5)
        result = channel.reduce { a, e -> a += e }
        then:
        result.getVal() == 15

        when:
        channel = Channel.create()
        result = channel.reduce { a, e -> a += e }
        channel << 99 << Channel.STOP
        then:
        result.getVal() == 99

        when:
        channel = Channel.create()
        result = channel.reduce { a, e -> a += e }
        channel << Channel.STOP
        then:
{'fraction_non_alphanumeric': 0.13227307990690457, 'fraction_numerical': 0.023758727695888285, 'mean_word_length': 1.5382722126507506, 'pattern_counts': {'":': 0, '<': 107, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3227514', 'n_tokens_mistral': 3349, 'n_tokens_neox': 3075, 'n_words': 801}