"org.codehaus.gpars:gpars:1.2.1"

"org.codehaus.gpars:gpars:1.3-SNAPSHOT"

<dependency>
    <groupId>org.codehaus.gpars</groupId>
    <artifactId>gpars</artifactId>
    <version>1.2.0</version>
</dependency>

"org.codehaus.jsr166-mirror:jsr166y:1.7.0"

<dependency>
    <groupId>org.codehaus.jsr166-mirror</groupId>
    <artifactId>jsr166y</artifactId>
    <version>1.7.0</version>
</dependency>

import static groovyx.gpars.actor.Actors.actor
/**
 * A demo showing two cooperating actors. The decryptor decrypts received messages
 * and replies them back.  The console actor sends a message to decrypt, prints out
 * the reply and terminates both actors.  The main thread waits on both actors to
 * finish using the join() method to prevent premature exit, since both actors use
 * the default actor group, which uses a daemon thread pool.
 * @author Dierk Koenig, Vaclav Pech
 */
def decryptor = actor {
    loop {
        react { message ->
            if (message instanceof String) reply message.reverse()
            else stop()
        }
    }
}
def console = actor {
    decryptor.send 'lellarap si yvoorG'
    react {
        println 'Decrypted message: ' + it
        decryptor.send false
    }
}
[decryptor, console]*.join()

import groovyx.gpars.MessagingRunnable;
import groovyx.gpars.actor.DynamicDispatchActor;
public class StatelessActorDemo {
    public static void main(String[] args) throws InterruptedException {
        final MyStatelessActor actor = new MyStatelessActor();
        actor.start();
        actor.send("Hello");
        actor.sendAndWait(10);
        actor.sendAndContinue(10.0, new MessagingRunnable<String>() {
            @Override protected void doRun(final String s) {
                System.out.println("Received a reply " + s);
            }
        });
    }
}
class MyStatelessActor extends DynamicDispatchActor {
    public void onMessage(final String msg) {
        System.out.println("Received " + msg);
        replyIfExists("Thank you");
    }
    public void onMessage(final Integer msg) {
        System.out.println("Received a number " + msg);
        replyIfExists("Thank you");
    }
    public void onMessage(final Object msg) {
        System.out.println("Received an object " + msg);
        replyIfExists("Thank you");
    }
}

myActor << 'message'
myAgent << {account -> account.add('5 USD')}
myDataflowVariable << 120332

myActor "message"
myAgent {house -> house.repair()}

myDataflowVariable >> {value -> doSomethingWith(value)}

import groovyx.gpars.MessagingRunnable;
 import groovyx.gpars.actor.DynamicDispatchActor;
 public class StatelessActorDemo {
     public static void main(String[] args) throws InterruptedException {
         final MyStatelessActor actor = new MyStatelessActor();
         actor.start();
         actor.send("Hello");
         actor.sendAndWait(10);
         actor.sendAndContinue(10.0, new MessagingRunnable<String>() {
             @Override protected void doRun(final String s) {
                 System.out.println("Received a reply " + s);
             }
         });
     }
 }
 class MyStatelessActor extends DynamicDispatchActor {
     public void onMessage(final String msg) {
         System.out.println("Received " + msg);
         replyIfExists("Thank you");
     }
     public void onMessage(final Integer msg) {
         System.out.println("Received a number " + msg);
         replyIfExists("Thank you");
     }
     public void onMessage(final Object msg) {
         System.out.println("Received an object " + msg);
         replyIfExists("Thank you");
     }
 }

import groovy.lang.Closure;
import groovyx.gpars.ReactorMessagingRunnable;
import groovyx.gpars.actor.Actor;
import groovyx.gpars.actor.ReactiveActor;
public class ReactorDemo {
    public static void main(final String[] args) throws InterruptedException {
        final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() {
            @Override protected Integer doRun(final Integer integer) {
                return integer * 2;
            }
        };
        final Actor actor = new ReactiveActor(handler);
        actor.start();
        System.out.println("Result: " +  actor.sendAndWait(1));
        System.out.println("Result: " +  actor.sendAndWait(2));
        System.out.println("Result: " +  actor.sendAndWait(3));
    }
}

import groovy.lang.Closure;
import groovyx.gpars.ReactorMessagingRunnable;
import groovyx.gpars.actor.Actor;
import groovyx.gpars.actor.Actors;
public class ReactorDemo {
    public static void main(final String[] args) throws InterruptedException {
        final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() {
            @Override protected Integer doRun(final Integer integer) {
                return integer * 2;
            }
        };
        final Actor actor = Actors.reactor(handler);
        System.out.println("Result: " +  actor.sendAndWait(1));
        System.out.println("Result: " +  actor.sendAndWait(2));
        System.out.println("Result: " +  actor.sendAndWait(3));
    }
}

import groovyx.gpars.MessagingRunnable;
 import groovyx.gpars.agent.Agent;
 public class AgentDemo {
     public static void main(final String[] args) throws InterruptedException {
         final Agent counter = new Agent<Integer>(0);
         counter.send(10);
         System.out.println("Current value: " + counter.getVal());
         counter.send(new MessagingRunnable<Integer>() {
             @Override protected void doRun(final Integer integer) {
                 counter.updateValue(integer + 1);
             }
         });
         System.out.println("Current value: " + counter.getVal());
     }
 }

import groovyx.gpars.MessagingRunnable;
import groovyx.gpars.dataflow.DataflowVariable;
import groovyx.gpars.group.DefaultPGroup;
import java.util.concurrent.Callable;
public class DataflowTaskDemo {
    public static void main(final String[] args) throws InterruptedException {
        final DefaultPGroup group = new DefaultPGroup(10);
        final DataflowVariable a = new DataflowVariable();
        group.task(new Runnable() {
            public void run() {
                a.bind(10);
            }
        });
        final Promise result = group.task(new Callable() {
            public Object call() throws Exception {
                return (Integer)a.getVal() + 10;
            }
        });
        result.whenBound(new MessagingRunnable<Integer>() {
            @Override protected void doRun(final Integer integer) {
                System.out.println("arguments = " + integer);
            }
        });
        System.out.println("result = " + result.getVal());
    }
}

import groovyx.gpars.DataflowMessagingRunnable;
import groovyx.gpars.dataflow.Dataflow;
import groovyx.gpars.dataflow.DataflowQueue;
import groovyx.gpars.dataflow.operator.DataflowProcessor;
import java.util.Arrays;
import java.util.List;
public class DataflowOperatorDemo {
    public static void main(final String[] args) throws InterruptedException {
        final DataflowQueue stream1 = new DataflowQueue();
        final DataflowQueue stream2 = new DataflowQueue();
        final DataflowQueue stream3 = new DataflowQueue();
        final DataflowQueue stream4 = new DataflowQueue();
        final DataflowProcessor op1 = Dataflow.selector(Arrays.asList(stream1), Arrays.asList(stream2), new DataflowMessagingRunnable(1) {
            @Override protected void doRun(final Object… objects) {
                getOwningProcessor().bindOutput(2*(Integer)objects[0]);
            }
        });
        final List secondOperatorInput = Arrays.asList(stream2, stream3);
        final DataflowProcessor op2 = Dataflow.operator(secondOperatorInput, Arrays.asList(stream4), new DataflowMessagingRunnable(2) {
            @Override protected void doRun(final Object… objects) {
                getOwningProcessor().bindOutput((Integer) objects[0] + (Integer) objects[1]);
            }
        });
        stream1.bind(1);
        stream1.bind(2);
        stream1.bind(3);
        stream3.bind(100);
        stream3.bind(100);
        stream3.bind(100);
        System.out.println("Result: " + stream4.getVal());
        System.out.println("Result: " + stream4.getVal());
        System.out.println("Result: " + stream4.getVal());
        op1.stop();
        op2.stop();
    }
}

def selfPortraits = images.findAllParallel{it.contains me}.collectParallel {it.resize()}

def smallestSelfPortrait = images.parallel.filter{it.contains me}.map{it.resize()}.min{it.sizeInMB}

//summarize numbers concurrently
 GParsPool.withPool {
     final AtomicInteger result = new AtomicInteger(0)
     [1, 2, 3, 4, 5].eachParallel {result.addAndGet(it)}
     assert 15 == result
 }
 //multiply numbers asynchronously
 GParsPool.withPool {
     final List result = [1, 2, 3, 4, 5].collectParallel {it * 2}
     assert ([2, 4, 6, 8, 10].equals(result))
 }

//check whether all elements within a collection meet certain criteria
 GParsPool.withPool(5) {ForkJoinPool pool ->
     assert [1, 2, 3, 4, 5].everyParallel {it > 0}
     assert ![1, 2, 3, 4, 5].everyParallel {it > 1}
 }

withPool(10) {...}
withPool(20, exceptionHandler) {...}

withPool {
     assert [1, 2, 3, 4, 5].everyParallel {it > 0}
     assert ![1, 2, 3, 4, 5].everyParallel {it > 1}
 }

import groovyx.gpars.ParallelEnhancer
def list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
ParallelEnhancer.enhanceInstance(list)
println list.collectParallel {it * 2 }
def animals = ['dog', 'ant', 'cat', 'whale']
ParallelEnhancer.enhanceInstance animals
println (animals.anyParallel {it ==~ /ant/} ? 'Found an ant' : 'No ants found')
println (animals.everyParallel {it.contains('a')} ? 'All animals contain a' : 'Some animals can live without an a')

GParsPool.withPool {
    //The selectImportantNames() will process the name collections concurrently
    assert ['ALICE', 'JASON'] == selectImportantNames(['Joe', 'Alice', 'Dave', 'Jason'].makeConcurrent())
}
/**
 * A function implemented using standard sequential collect() and findAll() methods.
 */
def selectImportantNames(names) {
    names.collect {it.toUpperCase()}.findAll{it.size() > 4}
}

import static groovyx.gpars.GParsPool.withPool
def list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
println 'Sequential: '
list.each { print it + ',' }
println()
withPool {
    println 'Sequential: '
    list.each { print it + ',' }
    println()
    list.makeConcurrent()
    println 'Concurrent: '
    list.each { print it + ',' }
    println()
    list.makeSequential()
    println 'Sequential: '
    list.each { print it + ',' }
    println()
}
println 'Sequential: '
list.each { print it + ',' }
println()

import static groovyx.gpars.GParsPool.withPool
def list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
println 'Sequential: '
list.each { print it + ',' }
println()
withPool {
    println 'Sequential: '
    list.each { print it + ',' }
    println()
    list.asConcurrent {
        println 'Concurrent: '
        list.each { print it + ',' }
        println()
    }
    println 'Sequential: '
    list.each { print it + ',' }
    println()
}
println 'Sequential: '
list.each { print it + ',' }
println()

/**
 * A function implemented using standard sequential collect() and findAll() methods.
 */
def selectImportantNames(names) {
    names.collect {it.toUpperCase()}.findAll{it.size() > 4}
}
def names = ['Joe', 'Alice', 'Dave', 'Jason']
ParallelEnhancer.enhanceInstance(names)
//The selectImportantNames() will process the name collections concurrently
assert ['ALICE', 'JASON'] == selectImportantNames(names.makeConcurrent())

import groovyx.gpars.ParallelEnhancer
def list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
println 'Sequential: '
list.each { print it + ',' }
println()
ParallelEnhancer.enhanceInstance(list)
println 'Sequential: '
list.each { print it + ',' }
println()
list.asConcurrent {
    println 'Concurrent: '
    list.each { print it + ',' }
    println()
}
list.makeSequential()
println 'Sequential: '
list.each { print it + ',' }
println()

def thumbnails = []
images.eachParallel {thumbnails << it.thumbnail}  //Concurrently accessing a not-thread-safe collection of thumbnails, don't do this!

//multiply numbers asynchronously
 GParsExecutorsPool.withPool {
     Collection<Future> result = [1, 2, 3, 4, 5].collectParallel{it * 10}
     assert new HashSet([10, 20, 30, 40, 50]) == new HashSet((Collection)result*.get())
 }
 //multiply numbers asynchronously using an asynchronous closure
 GParsExecutorsPool.withPool {
     def closure={it * 10}
     def asyncClosure=closure.async()
     Collection<Future> result = [1, 2, 3, 4, 5].collect(asyncClosure)
     assert new HashSet([10, 20, 30, 40, 50]) == new HashSet((Collection)result*.get())
 }

//find an element meeting specified criteria
 GParsExecutorsPool.withPool(5) {ExecutorService service ->
     service.submit({performLongCalculation()} as Runnable)
 }

withPool(10) {...}
withPool(20, threadFactory) {...}

withPool {
     def result = [1, 2, 3, 4, 5].findParallel{Number number -> number > 2}
     assert result in [3, 4, 5]
 }

import groovyx.gpars.GParsExecutorsPoolEnhancer
def list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
GParsExecutorsPoolEnhancer.enhanceInstance(list)
println list.collectParallel {it * 2 }
def animals = ['dog', 'ant', 'cat', 'whale']
GParsExecutorsPoolEnhancer.enhanceInstance animals
println (animals.anyParallel {it ==~ /ant/} ? 'Found an ant' : 'No ants found')
println (animals.allParallel {it.contains('a')} ? 'All animals contain a' : 'Some animals can live without an a')

GParsPool.withPool {
    def urls = ['http://www.dzone.com', 'http://www.theserverside.com', 'http://www.infoq.com']
    Closure download = {url ->
        println "Downloading $url"
        url.toURL().text.toUpperCase()
    }
    Closure cachingDownload = download.gmemoize()
    println 'Groovy sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GROOVY')}
    println 'Grails sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRAILS')}
    println 'Griffon sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRIFFON')}
    println 'Gradle sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRADLE')}
    println 'Concurrency sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('CONCURRENCY')}
    println 'GPars sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GPARS')}
}

Closure fib = {n -> n > 1 ? call(n - 1) + call(n - 2) : n}

Closure fib
fib = {n -> n > 1 ? fib(n - 1) + fib(n - 2) : n}.gmemoize()

Closure fib
fib = {n -> n > 1 ? fib(n - 1) + fib(n - 2) : n}.memoizeAtMost(2)

println 'Number of occurrences of the word GROOVY today: ' + urls.parallel
            .map {it.toURL().text.toUpperCase()}
            .filter {it.contains('GROOVY')}
            .map{it.split()}
            .map{it.findAll{word -> word.contains 'GROOVY'}.size()}
            .sum()

def myNumbers = (1..1000).parallel.filter{it % 2 == 0}.map{Math.sqrt it}.collection

import static groovyx.gpars.GParsPool.withPool
def words = "This is just a plain text to count words in"
print count(words)
def count(arg) {
  withPool {
    return arg.parallel
      .map{[it, 1]}
      .groupBy{it[0]}.getParallel()
      .map {it.value=it.value.size();it}
      .sort{-it.value}.collection
  }
}

def words = "This is just a plain text to count words in"
print count(words)
def count(arg) {
  withPool {
    return arg.parallel
      .map{[it, 1]}
      .combine(0) {sum, value -> sum + value}.getParallel()
      .sort{-it.value}.collection
  }
}

accumulator = {List acc, value -> acc << value} initialValue = []
accumulator = {List acc, value -> acc << value} initialValue = {-> []}
accumulator = {int sum, int value -> acc + value} initialValue = 0
accumulator = {int sum, int value -> sum + value} initialValue = {-> 0}
accumulator = {ShoppingCart cart, Item value -> cart.addItem(value)} initialValue = {-> new ShoppingCart()}

import groovy.transform.ToString
import groovy.transform.TupleConstructor
import static groovyx.gpars.GParsPool.withPool
TupleConstructor ToString
class PricedCar implements Cloneable {
    String model
    String color
    Double price
    boolean equals(final o) {
        if (this.is(o)) return true
        if (getClass() != o.class) return false
        final PricedCar pricedCar = (PricedCar) o
        if (color != pricedCar.color) return false
        if (model != pricedCar.model) return false
        return true
    }
    int hashCode() {
        int result
        result = (model != null ? model.hashCode() : 0)
        result = 31 * result + (color != null ? color.hashCode() : 0)
        return result
    }
    @Override
    protected Object clone() {
        return super.clone()
    }
}
def cars = [new PricedCar('F550', 'blue', 2342.223),
        new PricedCar('F550', 'red', 234.234),
        new PricedCar('Da', 'white', 2222.2),
        new PricedCar('Da', 'white', 1111.1)]
withPool {
    //Combine by model
    def result =
        cars.parallel.map {
            [it.model, it]
        }.combine(new PricedCar('', 'N/A', 0.0)) {sum, value ->
            sum.model = value.model
            sum.price += value.price
            sum
        }.values()
    println result

    //Combine by model and color (the PricedCar's equals and hashCode))
    result =
        cars.parallel.map {
            [it, it]
        }.combine(new PricedCar('', 'N/A', 0.0)) {sum, value ->
            sum.model = value.model
            sum.color = value.color
            sum.price += value.price
            sum
        }.values()
    println result
}

import groovyx.gpars.extra166y.Ops
groovyx.gpars.GParsPool.withPool {
    assert 15 == [1, 2, 3, 4, 5].parallelArray.reduce({a, b -> a + b} as Ops.Reducer, 0)                                        //summarize
    assert 55 == [1, 2, 3, 4, 5].parallelArray.withMapping({it ** 2} as Ops.Op).reduce({a, b -> a + b} as Ops.Reducer, 0)       //summarize squares
    assert 20 == [1, 2, 3, 4, 5].parallelArray.withFilter({it % 2 == 0} as Ops.Predicate)                                       //summarize squares of even numbers
            .withMapping({it ** 2} as Ops.Op)
            .reduce({a, b -> a + b} as Ops.Reducer, 0)
    assert 'aa:bb:cc:dd:ee' == 'abcde'.parallelArray                                                                            //concatenate duplicated characters with separator
            .withMapping({it * 2} as Ops.Op)
            .reduce({a, b -> "$a:$b"} as Ops.Reducer, "")

GParsPool.withPool() {
    Closure longLastingCalculation = {calculate()}
    Closure fastCalculation = longLastingCalculation.async()  //create a new closure, which starts the original closure on a thread pool
    Future result=fastCalculation()                           //returns almost immediately
    //do stuff while calculation performs …
    println result.get()
}

GParsPool.withPool() {
    /**
     * The callAsync() method is an asynchronous variant of the default call() method to invoke a closure.
     * It will return a Future for the result value.
     */
    assert 6 == {it * 2}.call(3)
    assert 6 == {it * 2}.callAsync(3).get()
}

{->
    while(true) {
        Thread.sleep 1000  //Simulate a bit of interesting calculation
        if (Thread.currentThread().isInterrupted()) break;  //We've been cancelled
    }
}.callTimeoutAsync(2000)

GParsExecutorsPool.withPool {ExecutorService executorService ->
    executorService << {println 'Inside parallel task'}
}

GParsPool.withPool {
    assert [10, 20] == GParsPool.executeAsyncAndWait({calculateA()}, {calculateB()}         //waits for results
    assert [10, 20] == GParsPool.executeAsync({calculateA()}, {calculateB()})*.get()  //returns Futures instead and doesn't wait for results to be calculated
}

def sum = (0..100000).inject(0, {a, b -> a + b})

def max = myNumbers.inject(0, {a, b -> a>b?a:b})

Closure download = {String url ->
    url.toURL().text
}
Closure loadFile = {String fileName ->
    …  //load the file here
}
Closure hash = {s -> s.hashCode()}
Closure compare = {int first, int second ->
    first == second
}
def result = compare(hash(download('http://www.gpars.org')), hash(loadFile('/coolStuff/gpars/website/index.html')))
println "The result of comparison: " + result

withPool {
    def maxPromise = numbers.inject(0, {a, b -> a>b?a:b}.asyncFun())
    println "Look Ma, I can talk to the user while the math is being done for me!"
    println maxPromise.get()
}

withPool {
    def sumPromise = (0..100000).inject(0, {a, b -> a + b}.asyncFun())
    println "Are we done yet? " + sumPromise.bound
    sumPromise.whenBound {sum -> println sum}
}

try {
    sumPromise.get()
} catch (MyCalculationException e) {
    println "Guess, things are not ideal today."
}

Closure download = {String url ->
        url.toURL().text
    }.asyncFun()
    Closure loadFile = {String fileName ->
        …  //load the file here
    }.asyncFun()
    Closure hash = {s -> s.hashCode()}.asyncFun()
    Closure compare = {int first, int second ->
        first == second
    }.asyncFun()
    def result = compare(hash(download('http://www.gpars.org')), hash(loadFile('/coolStuff/gpars/website/index.html')))
    println 'Allowed to do something else now'
    println "The result of comparison: " + result.get()

import static groovyx.gpars.GParsPool.withPool
  withPool {
      Closure plus = {Integer a, Integer b ->
          sleep 3000
          println 'Adding numbers'
          a + b
      }.asyncFun()
      Closure multiply = {Integer a, Integer b ->
          sleep 2000
          a * b
      }.asyncFun()
      Closure measureTime = {->
          sleep 3000
          4
      }.asyncFun()
      Closure distance = {Integer initialDistance, Integer velocity, Integer time ->
          plus(initialDistance, multiply(velocity, time))
      }.asyncFun()
      Closure chattyDistance = {Integer initialDistance, Integer velocity, Integer time ->
          println 'All parameters are now ready - starting'
          println 'About to call another asynchronous function'
          def innerResultPromise = plus(initialDistance, multiply(velocity, time))
          println 'Returning the promise for the inner calculation as my own result'
          return innerResultPromise
      }.asyncFun()
      println "Distance = " + distance(100, 20, measureTime()).get() + ' m'
      println "ChattyDistance = " + chattyDistance(100, 20, measureTime()).get() + ' m'
  }

class DownloadHelper {
    String download(String url) {
        url.toURL().text
    }
    int scanFor(String word, String text) {
        text.findAll(word).size()
    }
    String lower(s) {
        s.toLowerCase()
    }
}
//now we'll make the methods asynchronous
withPool {
    final DownloadHelper d = new DownloadHelper()
    Closure download = d.&download.asyncFun()
    Closure scanFor = d.&scanFor.asyncFun()
    Closure lower = d.&lower.asyncFun()
    //asynchronous processing
    def result = scanFor('groovy', lower(download('http://www.infoq.com')))
    println 'Allowed to do something else now'
    println result.get()
}

import static groovyx.gpars.GParsPool.withPool
import groovyx.gpars.AsyncFun
class DownloadingSearch {
    @AsyncFun Closure download = {String url ->
        url.toURL().text
    }
    @AsyncFun Closure scanFor = {String word, String text ->
        text.findAll(word).size()
    }
    @AsyncFun Closure lower = {s -> s.toLowerCase()}
    void scan() {
        def result = scanFor('groovy', lower(download('http://www.infoq.com')))  //synchronous processing
        println 'Allowed to do something else now'
        println result.get()
    }
}
withPool {
    new DownloadingSearch().scan()
}

@AsyncFun(GParsExecutorsPoolUtil) def sum6 = {a, b -> a + b }

@AsyncFun(blocking = true)
def sum = {a, b -> a + b }

Closure sPlus = {Integer a, Integer b ->
    a + b
}
Closure sMultiply = {Integer a, Integer b ->
    sleep 2000
    a * b
}
println "Synchronous result: " + sMultiply(sPlus(10, 30), 100)
final pool = new FJPool()
Closure aPlus = GParsPoolUtil.asyncFun(sPlus, pool)
Closure aMultiply = GParsPoolUtil.asyncFun(sMultiply, pool)
def result = aMultiply(aPlus(10, 30), 100)
println "Time to do something else while the calculation is running"
println "Asynchronous result: " + result.get()

Closure aPlus = GParsPoolUtil.asyncFun(sPlus)
Closure aMultiply = GParsPoolUtil.asyncFun(sMultiply)
withPool {
    def result = aMultiply(aPlus(10, 30), 100)
    println "Time to do something else while the calculation is running"
    println "Asynchronous result: " + result.get()
}

import static groovyx.gpars.GParsPool.runForkJoin
import static groovyx.gpars.GParsPool.withPool
withPool() {
    println """Number of files: ${
        runForkJoin(new File("./src")) {file ->
            long count = 0
            file.eachFile {
                if (it.isDirectory()) {
                    println "Forking a child task for $it"
                    forkOffChild(it)           //fork a child task
                } else {
                    count++
                }
            }
            return count + (childrenResults.sum(0))
            //use results of children tasks to calculate and store own result
        }
    }"""
}

def quicksort(numbers) {
    withPool {
        runForkJoin(0, numbers) {index, list ->
            def groups = list.groupBy {it <=> list[list.size().intdiv(2)]}
            if ((list.size() < 2) || (groups.size() == 1)) {
                return [index: index, list: list.clone()]
            }
            (-1..1).each {forkOffChild(it, groups[it] ?: [])}
            return [index: index, list: childrenResults.sort {it.index}.sum {it.list}]
        }.list
    }
}

public final class FileCounter extends AbstractForkJoinWorker<Long> {
    private final File file;
    def FileCounter(final File file) {
        this.file = file
    }
    @Override
    protected Long computeTask() {
        long count = 0;
        file.eachFile {
            if (it.isDirectory()) {
                println "Forking a thread for $it"
                forkOffChild(new FileCounter(it))           //fork a child task
            } else {
                count++
            }
        }
        return count + ((childrenResults)?.sum() ?: 0)  //use results of children tasks to calculate and store own result
    }
}
withPool(1) {pool ->  //feel free to experiment with the number of fork/join threads in the pool
    println "Number of files: ${runForkJoin(new FileCounter(new File("..")))}"
}

import static groovyx.gpars.GParsPool.runForkJoin
import static groovyx.gpars.GParsPool.withPool
/**
 * Splits a list of numbers in half
 */
def split(List<Integer> list) {
    int listSize = list.size()
    int middleIndex = listSize / 2
    def list1 = list[0..<middleIndex]
    def list2 = list[middleIndex..listSize - 1]
    return [list1, list2]
}
/**
 * Merges two sorted lists into one
 */
List<Integer> merge(List<Integer> a, List<Integer> b) {
    int i = 0, j = 0
    final int newSize = a.size() + b.size()
    List<Integer> result = new ArrayList<Integer>(newSize)
    while ((i < a.size()) && (j < b.size())) {
        if (a[i] <= b[j]) result << a[i++]
        else result << b[j++]
    }
    if (i < a.size()) result.addAll(a[i..-1])
    else result.addAll(b[j..-1])
    return result
}
final def numbers = [1, 5, 2, 4, 3, 8, 6, 7, 3, 4, 5, 2, 2, 9, 8, 7, 6, 7, 8, 1, 4, 1, 7, 5, 8, 2, 3, 9, 5, 7, 4, 3]
withPool(3) {  //feel free to experiment with the number of fork/join threads in the pool
    println """Sorted numbers: ${
        runForkJoin(numbers) {nums ->
            println "Thread ${Thread.currentThread().name[-1]}: Sorting $nums"
            switch (nums.size()) {
                case 0..1:
                    return nums                                   //store own result
                case 2:
                    if (nums[0] <= nums[1]) return nums     //store own result
                    else return nums[-1..0]                       //store own result
                default:
                    def splitList = split(nums)
                    [splitList[0], splitList[1]].each {forkOffChild it}  //fork a child task
                    return merge(* childrenResults)      //use results of children tasks to calculate and store own result
            }
        }
    }"""
}

public final class SortWorker extends AbstractForkJoinWorker<List<Integer>> {
    private final List numbers
    def SortWorker(final List<Integer> numbers) {
        this.numbers = numbers.asImmutable()
    }
    /**
     * Splits a list of numbers in half
     */
    def split(List<Integer> list) {
        int listSize = list.size()
        int middleIndex = listSize / 2
        def list1 = list[0..<middleIndex]
        def list2 = list[middleIndex..listSize - 1]
        return [list1, list2]
    }
    /**
     * Merges two sorted lists into one
     */
    List<Integer> merge(List<Integer> a, List<Integer> b) {
        int i = 0, j = 0
        final int newSize = a.size() + b.size()
        List<Integer> result = new ArrayList<Integer>(newSize)
        while ((i < a.size()) && (j < b.size())) {
            if (a[i] <= b[j]) result << a[i++]
            else result << b[j++]
        }
        if (i < a.size()) result.addAll(a[i..-1])
        else result.addAll(b[j..-1])
        return result
    }
    /**
     * Sorts a small list or delegates to two children, if the list contains more than two elements.
     */
    @Override
    protected List<Integer> computeTask() {
        println "Thread ${Thread.currentThread().name[-1]}: Sorting $numbers"
        switch (numbers.size()) {
            case 0..1:
                return numbers                                   //store own result
            case 2:
                if (numbers[0] <= numbers[1]) return numbers     //store own result
                else return numbers[-1..0]                       //store own result
            default:
                def splitList = split(numbers)
                [new SortWorker(splitList[0]), new SortWorker(splitList[1])].each{forkOffChild it}  //fork a child task
                return merge(* childrenResults)      //use results of children tasks to calculate and store own result
        }
    }
}
final def numbers = [1, 5, 2, 4, 3, 8, 6, 7, 3, 4, 5, 2, 2, 9, 8, 7, 6, 7, 8, 1, 4, 1, 7, 5, 8, 2, 3, 9, 5, 7, 4, 3]
withPool(1) {  //feel free to experiment with the number of fork/join threads in the pool
    println "Sorted numbers: ${runForkJoin(new SortWorker(numbers))}"
}

Closure fib = {number ->
            if (number <= 2) {
                return 1
            }
            forkOffChild(number - 1)                            //  This task will run asynchronously, probably in a different thread
            final def result = runChildDirectly(number - 2)     //  This task is run directly within the current thread
            return (Integer) getChildrenResults().sum() + result
        }
        withPool {
            assert 55 == runForkJoin(10, fib)
        }

def numbers = …
def quickSort = …
def mergeSort = …
def sortedNumbers = speculate(quickSort, mergeSort)

import static groovyx.gpars.GParsPool.speculate
import static groovyx.gpars.GParsPool.withPool
def alternative1 = {
    'http://www.dzone.com/links/index.html'.toURL().text
}
def alternative2 = {
    'http://www.dzone.com/'.toURL().text
}
def alternative3 = {
    'http://www.dzzzzzone.com/'.toURL().text  //wrong url
}
def alternative4 = {
    'http://dzone.com/'.toURL().text
}
withPool(4) {
    println speculate([alternative1, alternative2, alternative3, alternative4]).contains('groovy')
}

import groovyx.gpars.dataflow.DataflowQueue
import static groovyx.gpars.dataflow.Dataflow.task
def alternative1 = {
    'http://www.dzone.com/links/index.html'.toURL().text
}
def alternative2 = {
    'http://www.dzone.com/'.toURL().text
}
def alternative3 = {
    'http://www.dzzzzzone.com/'.toURL().text  //will fail due to wrong url
}
def alternative4 = {
    'http://dzone.com/'.toURL().text
}
//Pick either one of the following, both will work:
final def result = new DataflowQueue()
//  final def result = new DataflowVariable()
[alternative1, alternative2, alternative3, alternative4].each {code ->
    task {
        try {
            result << code()
        } catch (ignore) { }  //We deliberately ignore unsuccessful urls
    }
}
println result.val.contains('groovy')

import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
group = new DefaultPGroup(new ResizeablePool(true))
def t = group.task {
    println "I am a process"
}
t.join()

import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
group = new DefaultPGroup(new ResizeablePool(true))
class MyProcess implements Runnable {
    @Override
    void run() {
        println "I am a process"
    }
}
def t = group.task new MyProcess()
t.join()

import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
import java.util.concurrent.Callable
group = new DefaultPGroup(new ResizeablePool(true))
class MyProcess implements Callable<String> {
    @Override
    String call() {
        println "I am a process"
        return "CSP is great!"
    }
}
def t = group.task new MyProcess()
println t.get()

import groovy.transform.TupleConstructor
import groovyx.gpars.dataflow.DataflowReadChannel
import groovyx.gpars.dataflow.DataflowWriteChannel
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
import java.util.concurrent.Callable
import groovyx.gpars.dataflow.SyncDataflowQueue
group = new DefaultPGroup(new ResizeablePool(true))
@TupleConstructor
class Greeter implements Callable<String> {
    DataflowReadChannel names
    DataflowWriteChannel greetings
    @Override
    String call() {
        while(!Thread.currentThread().isInterrupted()) {
            String name = names.val
            greetings << "Hello " + name
        }
        return "CSP is great!"
    }
}
def a = new SyncDataflowQueue()
def b = new SyncDataflowQueue()
group.task new Greeter(a, b)
a << "Joe"
a << "Dave"
println b.val
println b.val

group = new DefaultPGroup(new ResizeablePool(true))
@TupleConstructor
class Formatter implements Callable<String> {
    DataflowReadChannel rawNames
    DataflowWriteChannel formattedNames
    @Override
    String call() {
        while(!Thread.currentThread().isInterrupted()) {
            String name = rawNames.val
            formattedNames << name.toUpperCase()
        }
    }
}
@TupleConstructor
class Greeter implements Callable<String> {
    DataflowReadChannel names
    DataflowWriteChannel greetings
    @Override
    String call() {
        while(!Thread.currentThread().isInterrupted()) {
            String name = names.val
            greetings << "Hello " + name
        }
    }
}
def a = new SyncDataflowQueue()
def b = new SyncDataflowQueue()
def c = new SyncDataflowQueue()
group.task new Formatter(a, b)
group.task new Greeter(b, c)
a << "Joe"
a << "Dave"
println c.val
println c.val

import groovy.transform.TupleConstructor
import groovyx.gpars.dataflow.SyncDataflowQueue
import groovyx.gpars.dataflow.DataflowReadChannel
import groovyx.gpars.dataflow.DataflowWriteChannel
import groovyx.gpars.dataflow.Select
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
import static groovyx.gpars.dataflow.Dataflow.select
group = new DefaultPGroup(new ResizeablePool(true))
@TupleConstructor
class Receptionist implements Runnable {
    DataflowReadChannel emails
    DataflowReadChannel phoneCalls
    DataflowReadChannel tweets
    DataflowWriteChannel forwardedMessages
    private final Select incomingRequests = select([phoneCalls, emails, tweets])  //prioritySelect() would give highest precedence to phone calls
    @Override
    void run() {
        while(!Thread.currentThread().isInterrupted()) {
            String msg = incomingRequests.select()
            forwardedMessages << msg.toUpperCase()
        }
    }
}
def a = new SyncDataflowQueue()
def b = new SyncDataflowQueue()
def c = new SyncDataflowQueue()
def d = new SyncDataflowQueue()
group.task new Receptionist(a, b, c, d)
a << "my email"
b << "my phone call"
c << "my tweet"
//The values come in random order since the process uses a Select to read its input
3.times{
    println d.val.value
}

final class Prefix implements Callable {
    private final DataflowChannel inChannel
    private final DataflowChannel outChannel
    private final def prefix
    def Prefix(final inChannel, final outChannel, final prefix) {
        this.inChannel = inChannel;
        this.outChannel = outChannel;
        this.prefix = prefix
    }
    public def call() {
        outChannel << prefix
        while (true) {
            sleep 200
            outChannel << inChannel.val
        }
    }
}

final class Copy implements Callable {
    private final DataflowChannel inChannel
    private final DataflowChannel outChannel1
    private final DataflowChannel outChannel2
    def Copy(final inChannel, final outChannel1, final outChannel2) {
        this.inChannel = inChannel;
        this.outChannel1 = outChannel1;
        this.outChannel2 = outChannel2;
    }
    public def call() {
        final PGroup group = Dataflow.retrieveCurrentDFPGroup()
        while (true) {
            def i = inChannel.val
            group.task {
                outChannel1 << i
                outChannel2 << i
            }.join()
        }
    }
}

import groovyx.gpars.dataflow.DataflowChannel
import groovyx.gpars.dataflow.SyncDataflowQueue
import groovyx.gpars.group.DefaultPGroup
group = new DefaultPGroup(6)
def fib(DataflowChannel out) {
    group.task {
        def a = new SyncDataflowQueue()
        def b = new SyncDataflowQueue()
        def c = new SyncDataflowQueue()
        def d = new SyncDataflowQueue()
        [new Prefix(d, a, 0L), new Prefix(c, d, 1L), new Copy(a, b, out), new StatePairs(b, c)].each { group.task it}
    }
}
final SyncDataflowQueue ch = new SyncDataflowQueue()
group.task new Print('Fibonacci numbers', ch)
fib(ch)
sleep 10000

import groovyx.gpars.actor.Actor
import groovyx.gpars.actor.DefaultActor
class GameMaster extends DefaultActor {
    int secretNum
    void afterStart() {
        secretNum = new Random().nextInt(10)
    }
    void act() {
        loop {
            react { int num ->
                if (num > secretNum)
                reply 'too large'
                else if (num < secretNum)
                reply 'too small'
                else {
                    reply 'you win'
                    terminate()
                }
            }
        }
    }
}
class Player extends DefaultActor {
    String name
    Actor server
    int myNum
    void act() {
        loop {
            myNum = new Random().nextInt(10)
            server.send myNum
            react {
                switch (it) {
                    case 'too large': println "$name: $myNum was too large"; break
                    case 'too small': println "$name: $myNum was too small"; break
                    case 'you win': println "$name: I won $myNum"; terminate(); break
                }
            }
        }
    }
}
def master = new GameMaster().start()
def player = new Player(name: 'Player', server: master).start()
//this forces main thread to live until both actors stop
[master, player]*.join()

def passiveActor = Actors.actor{
    loop {
        react { msg -> println "Received: $msg"; }
    }
}
passiveActor.send 'Message 1'
passiveActor << 'Message 2'    //using the << operator
passiveActor 'Message 3'       //using the implicit call() method

def replyingActor = Actors.actor{
    loop {
        react { msg ->
            println "Received: $msg";
            reply "I've got $msg"
        }
    }
}
def reply1 = replyingActor.sendAndWait('Message 4')
def reply2 = replyingActor.sendAndWait('Message 5', 10, TimeUnit.SECONDS)
use (TimeCategory) {
    def reply3 = replyingActor.sendAndWait('Message 6', 10.seconds)
}

friend.sendAndContinue 'I need money!', {money -> pocket money}
println 'I can continue while my friend is collecting money for me'

Promise loan = friend.sendAndPromise 'I need money!'
println 'I can continue while my friend is collecting money for me'
loan.whenBound {money -> pocket money}  //asynchronous waiting for a reply
println "Received ${loan.get()}"  //synchronous waiting for a reply

println 'Waiting for a gift'
react {gift ->
    if (myWife.likes gift) reply 'Thank you!'
}

Actors.actor {
    loop {
        println 'Waiting for a gift'
        react {gift ->
            if (myWife.likes gift) reply 'Thank you!'
            else {
                reply 'Try again, please'
                react {anotherGift ->
                    if (myChildren.like gift) reply 'Thank you!'
                }
                println 'Never reached'
            }
        }
        println 'Never reached'
    }
    println 'Never reached'
}

react {tweet ->
    if (isSpam(tweet)) ignoreTweetsFrom sender
    sender.send 'Never write me again!'
}

def decryptor = Actors.actor {
    react {message ->
        reply message.reverse()
//        sender.send message.reverse()    //An alternative way to send replies
    }
}
def console = Actors.actor {  //This actor will print out decrypted messages, since the replies are forwarded to it
    react {
        println 'Decrypted message: ' + it
    }
}
decryptor.send 'lellarap si yvoorG', console  //Specify an actor to send replies to
console.join()

def console = Actors.actor {
    loop {
        react {
            println it
        }
    }
}

final def decryptor = Actors.actor {
    loop {
        react {String message ->
            if ('stopService' == message) {
                println 'Stopping decryptor'
                stop()
            }
            else reply message.reverse()
        }
    }
}
Actors.actor {
    decryptor.send 'lellarap si yvoorG'
    react {
        println 'Decrypted message: ' + it
        decryptor.send 'stopService'
    }
}.join()

def friend = Actors.actor {
    react {
        //this doesn't reply -> caller won't receive any answer in time
        println it
        //reply 'Hello' //uncomment this to answer conversation
        react {
            println it
        }
    }
}
def me = Actors.actor {
    friend.send('Hi')
    //wait for answer 1sec
    react(1000) {msg ->
        if (msg == Actor.TIMEOUT) {
            friend.send('I see, busy as usual. Never mind.')
            stop()
        } else {
            //continue conversation
            println "Thank you for $msg"
        }
    }
}
me.join()

final DefaultActor me
me = Actors.actor {
    def message = 1
    message.metaClass.onDeliveryError = {->
        //send message back to the caller
        me << "Could not deliver $delegate"
    }
    def actor = Actors.actor {
        react {
            //wait 2sec in order next call in demo can be emitted
            Thread.sleep(2000)
            //stop actor after first message
            stop()
        }
    }
    actor << message
    actor << message
    react {
        //print whatever comes back
        println it
    }
}
me.join()

final DefaultActor me
me = Actors.actor {
    def message1 = 1
    def message2 = 2
    def actor = Actors.actor {
        react {
            //wait 2sec in order next call in demo can be emitted
            Thread.sleep(2000)
            //stop actor after first message
            stop()
        }
    }
    me.metaClass.onDeliveryError = {msg ->
        //callback on actor inaccessibility
        println "Could not deliver message $msg"
    }
    actor << message1
    actor << message2
    actor.join()
}
me.join()

class MyActor extends DefaultActor {
    public void onDeliveryError(msg) {
        println "Could not deliver message $msg"
    }
    …
}

def master = new GameMaster().start()
def player = new Player(name: 'Player', server: master).start()
[master, player]*.join()

final Actor actor = Actors.actor {
    def candidates = []
    def printResult = {-> println "The best offer is ${candidates.max()}"}
    loop(3, printResult) {
        react {
            candidates << it
        }
    }
}
actor 10
actor 30
actor 20
actor.join()

final Actor actor = Actors.actor {
    def candidates = []
    final Closure printResult = {-> println "Reached best offer - ${candidates.max()}"}
    loop({-> candidates.max() < 30}, printResult) {
        react {
            candidates << it
        }
    }
}
actor 10
actor 20
actor 25
actor 31
actor 20
actor.join()

import static groovyx.gpars.actor.Actors.actor
def console = actor {
    loop {
        react {
            println it
        }
    }

class CustomActor extends DefaultActor {
    @Override
    protected void act() {
        loop {
            react {
                println it
            }
        }
    }
}
def console=new CustomActor()
console.start()

console.send('Message')
console 'Message'
console.sendAndWait 'Message'                                                     //Wait for a reply
console.sendAndContinue 'Message', {reply -> println "I received reply: $reply"}  //Forward the reply to a function

import static groovyx.gpars.actor.Actors.actor
final def decryptor = actor {
    loop {
        react {String message->
            reply message.reverse()
        }
    }
}
def console = actor {
    decryptor.send 'lellarap si yvoorG'
    react {
        println 'Decrypted message: ' + it
    }
}
console.join()

def friend = Actors.actor {
    react {
        //this doesn't reply -> caller won't receive any answer in time
        println it
        //reply 'Hello' //uncomment this to answer conversation
        react {
            println it
        }
    }
}
def me = Actors.actor {
    friend.send('Hi')
    //wait for answer 1sec
    react(1000) {msg ->
        if (msg == Actor.TIMEOUT) {
            friend.send('I see, busy as usual. Never mind.')
            stop()
        } else {
            //continue conversation
            println "Thank you for $msg"
        }
    }
}
me.join()

def friend = Actors.actor {
    react {
        //this doesn't reply -> caller won't receive any answer in time
        println it
        //reply 'Hello' //uncomment this to answer conversation
        react {
            println it
        }
    }
}
def me = Actors.actor {
    friend.send('Hi')
    delegate.metaClass.onTimeout = {->
        friend.send('I see, busy as usual. Never mind.')
        stop()
    }
    //wait for answer 1sec
    react(1000) {msg ->
        if (msg != Actor.TIMEOUT) {
            //continue conversation
            println "Thank you for $msg"
        }
    }
}
me.join()

class MyActor extends DefaultActor {
    public void onTimeout() {
        …
    }
    protected void act() {
       …
    }
}

class MyCounterActor extends DefaultActor {
    private Integer counter = 0
    protected void act() {
        loop {
            react {
                counter++
            }
        }
    }
}

import groovyx.gpars.group.DefaultPGroup
//not necessary, just showing that a single-threaded pool can still handle multiple actors
def group = new DefaultPGroup(1);
final def console = group.actor {
    loop {
        react {
            println 'Result: ' + it
        }
    }
}
final def calculator = group.actor {
    react {a ->
        react {b ->
            console.send(a + b)
        }
    }
}
calculator.send 2
calculator.send 3
calculator.join()
group.shutdown()

import groovyx.gpars.group.DefaultPGroup
import static groovyx.gpars.actor.Actors.actor
Closure createMessageHandler(def parentActor) {
    return {
        react {List<Integer> message ->
            assert message != null
            switch (message.size()) {
                case 0..1:
                    parentActor.send(message)
                    break
                case 2:
                    if (message[0] <= message[1]) parentActor.send(message)
                    else parentActor.send(message[-1..0])
                    break
                default:
                    def splitList = split(message)
                    def child1 = actor(createMessageHandler(delegate))
                    def child2 = actor(createMessageHandler(delegate))
                    child1.send(splitList[0])
                    child2.send(splitList[1])
                    react {message1 ->
                        react {message2 ->
                            parentActor.send merge(message1, message2)
                        }
                    }
            }
        }
    }
}
def console = new DefaultPGroup(1).actor {
    react {
        println "Sorted array:t${it}"
        System.exit 0
    }
}
def sorter = actor(createMessageHandler(console))
sorter.send([1, 5, 2, 4, 3, 8, 6, 7, 3, 9, 5, 3])
console.join()
def split(List<Integer> list) {
    int listSize = list.size()
    int middleIndex = listSize / 2
    def list1 = list[0..<middleIndex]
    def list2 = list[middleIndex..listSize - 1]
    return [list1, list2]
}
List<Integer> merge(List<Integer> a, List<Integer> b) {
    int i = 0, j = 0
    final int newSize = a.size() + b.size()
    List<Integer> result = new ArrayList<Integer>(newSize)
    while ((i < a.size()) && (j < b.size())) {
        if (a[i] <= b[j]) result << a[i++]
        else result << b[j++]
    }
    if (i < a.size()) result.addAll(a[i..-1])
    else result.addAll(b[j..-1])
    return result
}

class MyActor extends DefaultActor {
    public void afterStart() {
        …
    }
    public void onTimeout() {
        …
    }
    protected void act() {
       …
    }
}

def myActor = actor {
    delegate.metaClass.onException = {
        log.error('Exception occurred', it)
    }
…
}

def myGroup = new DefaultPGroup()
def actor1 = myGroup.actor {
…
}
def actor2 = myGroup.actor {
…
}

class MyActor extends StaticDispatchActor<Integer> {
    private static PGroup group = new DefaultPGroup(100)
    MyActor(...) {
        this.parallelGroup = group
        …
    }
}

def myGroup = new DefaultPGroup(10)  //the pool will contain 10 threads

… (n+1 threads in the default pool after startup)
Actors.defaultActorPGroup.resize 1  //use one-thread pool
… (1 thread in the pool)
Actors.defaultActorPGroup.resetDefaultSize()
… (n+1 threads in the pool)
Actors.defaultActorPGroup.shutdown()

def daemonGroup = new DefaultPGroup()
def actor1 = daemonGroup.actor {
…
}
def nonDaemonGroup = new NonDaemonPGroup()
def actor2 = nonDaemonGroup.actor {
…
}
class MyActor {
    def MyActor() {
        this.parallelGroup = nonDaemonGroup
    }
    void act() {...}
}

def coreActors = new NonDaemonPGroup(5)  //5 non-daemon threads pool
def helperActors = new DefaultPGroup(1)  //1 daemon thread pool
def priceCalculator = coreActors.actor {
…
}
def paymentProcessor = coreActors.actor {
…
}
def emailNotifier = helperActors.actor {
…
}
def cleanupActor = helperActors.actor {
…
}
//increase size of the core actor group
coreActors.resize 6
//shutdown the group's pool once you no longer need the group to release resources
helperActors.shutdown()

def nonDaemonGroup = new NonDaemonPGroup()
def myActor = nonDaemonGroup.actor {...}

def nonDaemonGroup = new NonDaemonPGroup()
class MyActor extends DefaultActor {
    def MyActor() {
        this.parallelGroup = nonDaemonGroup
    }
    void act() {...}
}
def myActor = new MyActor()

def decryptor = blockingActor {
    while (true) {
        receive {message ->
            if (message instanceof String) reply message.reverse()
            else stop()
        }
    }
}
def console = blockingActor {
    decryptor.send 'lellarap si yvoorG'
    println 'Decrypted message: ' + receive()
    decryptor.send false
}
[decryptor, console]*.join()

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DynamicDispatchActor
final class MyActor extends DynamicDispatchActor {
    void onMessage(String message) {
        println 'Received string'
    }
    void onMessage(Integer message) {
        println 'Received integer'
        reply 'Thanks!'
    }
    void onMessage(Object message) {
        println 'Received object'
        sender.send 'Thanks!'
    }
    void onMessage(List message) {
        println 'Received list'
        stop()
    }
}
final def myActor = new MyActor().start()
Actors.actor {
    myActor 1
    myActor ''
    myActor 1.0
    myActor(new ArrayList())
    myActor.join()
}.join()

final Actor myActor = new DynamicDispatchActor().become {
    when {String msg -> println 'A String'; reply 'Thanks'}
    when {Double msg -> println 'A Double'; reply 'Thanks'}
    when {msg -> println 'A something ...'; reply 'What was that?';stop()}
}
myActor.start()
Actors.actor {
    myActor 'Hello'
    myActor 1.0d
    myActor 10 as BigDecimal
    myActor.join()
}.join()

final class MyDDA extends DynamicDispatchActor {
    void onMessage(String message) {
        println 'Received string'
    }
    void onMessage(Integer message) {
        println 'Received integer'
    }
    void onMessage(Object message) {
        println 'Received object'
    }
    void onMessage(List message) {
        println 'Received list'
        stop()
    }
}
final def myActor = new MyDDA().become {
    when {BigDecimal num -> println 'Received BigDecimal'}
    when {Float num -> println 'Got a float'}
}.start()
Actors.actor {
    myActor 'Hello'
    myActor 1.0f
    myActor 10 as BigDecimal
    myActor.send([])
    myActor.join()
}.join()

def fairActor = Actors.fairMessageHandler {...}

final class MyActor extends StaticDispatchActor<String> {
    void onMessage(String message) {
        println 'Received string ' + message
        switch (message) {
            case 'hello':
                reply 'Hi!'
                break
            case 'stop':
                stop()
        }
    }
}

final actor = staticMessageHandler {String message ->
    println 'Received string ' + message
    switch (message) {
        case 'hello':
            reply 'Hi!'
            break
        case 'stop':
            stop()
    }
}
println 'Reply: ' + actor.sendAndWait('hello')
actor 'bye'
actor 'stop'
actor.join()

final def group = new DefaultPGroup()
final def doubler = group.reactor {
    2 * it
}
group.actor {
    println 'Double of 10 = ' + doubler.sendAndWait(10)
}
group.actor {
    println 'Double of 20 = ' + doubler.sendAndWait(20)
}
group.actor {
    println 'Double of 30 = ' + doubler.sendAndWait(30)
}
for(i in (1..10)) {
    println "Double of $i = ${doubler.sendAndWait(i)}"
}
doubler.stop()
doubler.join()

import groovyx.gpars.actor.Actor
import groovyx.gpars.actor.Actors
final def doubler = Actors.reactor {
    2 * it
}
Actor actor = Actors.actor {
    (1..10).each {doubler << it}
    int i = 0
    loop {
        i += 1
        if (i > 10) stop()
        else {
            react {message ->
                println "Double of $i = $message"
            }
        }
    }
}
actor.join()
doubler.stop()
doubler.join()

public class ReactiveActor extends DefaultActor {
    Closure body
    void act() {
        loop {
            react {message ->
                reply body(message)
            }
        }
    }
}

def fairActor = Actors.fairReactor {...}

class MyDemoActor extends DefaultActor {
    protected void act() {
        handleA()
    }
    private void handleA() {
        react {a ->
            handleB(a)
        }
    }
    private void handleB(int a) {
        react {b ->
            println a + b
            reply a + b
        }
    }
}
final def demoActor = new MyDemoActor()
demoActor.start()
Actors.actor {
    demoActor 10
    demoActor 20
    react {
        println "Result: $it"
    }
}.join()

Actor demoActor = Actors.actor {
    delegate.metaClass {
        handleA = {->
            react {a ->
                 handleB(a)
            }
        }
        handleB = {a ->
            react {b ->
                println a + b
                reply a + b
            }
        }
    }
    handleA()
}
Actors.actor {
    demoActor 10
    demoActor 20
    react {
        println "Result: $it"
    }
}.join()

Closure handleB = {a ->
    react {b ->
        println a + b
        reply a + b
    }
}
Closure handleA = {->
    react {a ->
        handleB(a)
    }
}
Actor demoActor = Actors.actor {
    handleA.delegate = delegate
    handleB.delegate = delegate
    handleA()
}
Actors.actor {
    demoActor 10
    demoActor 20
    react {
        println "Result: $it"
    }
}.join()

class MyLoopActor extends DefaultActor {
    protected void act() {
        outerLoop()
    }
    private void outerLoop() {
        react {a ->
            println 'Outer: ' + a
            if (a != 0) innerLoop()
            else println 'Done'
        }
    }
    private void innerLoop() {
        react {b ->
            println 'Inner ' + b
            if (b == 0) outerLoop()
            else innerLoop()
        }
    }
}
final def actor = new MyLoopActor().start()
actor 10
actor 20
actor 0
actor 0
actor.join()

Actor actor = Actors.actor {
  delegate.metaClass {
      outerLoop = {->
          react {a ->
              println 'Outer: ' + a
              if (a!=0) innerLoop()
              else println 'Done'
          }
      }
      innerLoop = {->
          react {b ->
              println 'Inner ' + b
              if (b==0) outerLoop()
              else innerLoop()
          }
      }
  }
  outerLoop()
}
actor 10
actor 20
actor 0
actor 0
actor.join()

Closure innerLoop
Closure outerLoop = {->
    react {a ->
        println 'Outer: ' + a
        if (a!=0) innerLoop()
        else println 'Done'
    }
}
innerLoop = {->
    react {b ->
        println 'Inner ' + b
        if (b==0) outerLoop()
        else innerLoop()
    }
}
Actor actor = Actors.actor {
    outerLoop.delegate = delegate
    innerLoop.delegate = delegate
    outerLoop()
}
actor 10
actor 20
actor 0
actor 0
actor.join()

class MyLoopingActor extends DefaultActor {
  protected void act() {
      loop {
          outerLoop()
      }
  }
  private void outerLoop() {
      react {a ->
          println 'Outer: ' + a
          if (a!=0) innerLoop()
          else println 'Done for now, but will loop again'
      }
  }
  private void innerLoop() {
      react {b ->
          println 'Inner ' + b
          if (b == 0) outerLoop()
          else innerLoop()
      }
  }
}
final def actor = new MyLoopingActor().start()
actor 10
actor 20
actor 0
actor 0
actor 10
actor.stop()
actor.join()

import groovyx.gpars.activeobject.ActiveObject
import groovyx.gpars.activeobject.ActiveMethod
@ActiveObject
class Decryptor {
    @ActiveMethod
    def decrypt(String encryptedText) {
        return encryptedText.reverse()
    }
    @ActiveMethod
    def decrypt(Integer encryptedNumber) {
        return -1*encryptedNumber + 142
    }
}
final Decryptor decryptor = new Decryptor()
def part1 = decryptor.decrypt(' noitcA ni yvoorG')
def part2 = decryptor.decrypt(140)
def part3 = decryptor.decrypt('noitide dn')
print part1.get()
print part2.get()
println part3.get()

import groovyx.gpars.activeobject.ActiveMethod
import groovyx.gpars.activeobject.ActiveObject
import groovyx.gpars.dataflow.DataflowVariable
@ActiveObject
class Decryptor {
    @ActiveMethod(blocking=true)
    String decrypt(String encryptedText) {
        encryptedText.reverse()
    }
    @ActiveMethod(blocking=true)
    Integer decrypt(Integer encryptedNumber) {
        -1*encryptedNumber + 142
    }
}
final Decryptor decryptor = new Decryptor()
print decryptor.decrypt(' noitcA ni yvoorG')
print decryptor.decrypt(140)
println decryptor.decrypt('noitide dn')

import groovyx.gpars.activeobject.ActiveObject
import groovyx.gpars.activeobject.ActiveMethod
import groovyx.gpars.dataflow.DataflowVariable
@ActiveObject
class A {
    @ActiveMethod
    def fooA(value) {
        …
    }
}
class B extends A {
}
@ActiveObject
class C extends B {
    @ActiveMethod
    def fooC(value1, value2) {
        …
    }
}

@ActiveObject("group1")
class MyActiveObject {
    …
}

final DefaultPGroup group = new DefaultPGroup(10)
ActiveObjectRegistry.instance.register("group1", group)

@ActiveObject(actorName = "alternativeActorName")
class MyActiveObject {
    …
}

import groovyx.gpars.actor.DynamicDispatchActor
/**
 * Demonstrates concurrent implementation of the Sieve of Eratosthenes using actors
 *
 * In principle, the algorithm consists of concurrently run chained filters,
 * each of which detects whether the current number can be divided by a single prime number.
 * (generate nums 1, 2, 3, 4, 5, ...) -> (filter by mod 2) -> (filter by mod 3) -> (filter by mod 5) -> (filter by mod 7) -> (filter by mod 11) -> (caution! Primes falling out here)
 * The chain is built (grows) on the fly, whenever a new prime is found.
 */
int requestedPrimeNumberBoundary = 1000
final def firstFilter = new FilterActor(2).start()
/**
 * Generating candidate numbers and sending them to the actor chain
 */
(2..requestedPrimeNumberBoundary).each {
    firstFilter it
}
firstFilter.sendAndWait 'Poison'
/**
 * Filter out numbers that can be divided by a single prime number
 */
final class FilterActor extends DynamicDispatchActor {
    private final int myPrime
    private def follower
    def FilterActor(final myPrime) { this.myPrime = myPrime; }
    /**
     * Try to divide the received number with the prime. If the number cannot be divided, send it along the chain.
     * If there's no-one to send it to, I'm the last in the chain, the number is a prime and so I will create and chain
     * a new actor responsible for filtering by this newly found prime number.
     */
    def onMessage(int value) {
        if (value % myPrime != 0) {
            if (follower) follower value
            else {
                println "Found $value"
                follower = new FilterActor(value).start()
            }
        }
    }
    /**
     * Stop the actor on poisson reception
     */
    def onMessage(def poisson) {
        if (follower) {
            def sender = sender
            follower.sendAndContinue(poisson, {this.stop(); sender?.send('Done')})  //Pass the poisson along and stop after a reply
        } else {  //I am the last in the chain
            stop()
            reply 'Done'
        }
    }
}

import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.actor.DefaultActor
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.actor.Actor
final def group = new DefaultPGroup()
final def barber = group.actor {
    final def random = new Random()
    loop {
        react {message ->
            switch (message) {
                case Enter:
                    message.customer.send new Start()
                    println "Barber: Processing customer ${message.customer.name}"
                    doTheWork(random)
                    message.customer.send new Done()
                    reply new Next()
                    break
                case Wait:
                    println "Barber: No customers. Going to have a sleep"
                    break
            }
        }
    }
}
private def doTheWork(Random random) {
    Thread.sleep(random.nextInt(10) * 1000)
}
final Actor waitingRoom
waitingRoom = group.actor {
    final int capacity = 5
    final List<Customer> waitingCustomers = []
    boolean barberAsleep = true
    loop {
        react {message ->
            switch (message) {
                case Enter:
                    if (waitingCustomers.size() == capacity) {
                        reply new Full()
                    } else {
                        waitingCustomers << message.customer
                        if (barberAsleep) {
                            assert waitingCustomers.size() == 1
                            barberAsleep = false
                            waitingRoom.send new Next()
                        }
                        else reply new Wait()
                    }
                    break
                case Next:
                    if (waitingCustomers.size()>0) {
                        def customer = waitingCustomers.remove(0)
                        barber.send new Enter(customer:customer)
                    } else {
                        barber.send new Wait()
                        barberAsleep = true
                    }
            }
        }
    }
}
class Customer extends DefaultActor {
    String name
    Actor localBarbers
    void act() {
        localBarbers << new Enter(customer:this)
        loop {
            react {message ->
                switch (message) {
                    case Full:
                        println "Customer: $name: The waiting room is full. I am leaving."
                        stop()
                        break
                    case Wait:
                        println "Customer: $name: I will wait."
                        break
                    case Start:
                        println "Customer: $name: I am now being served."
                        break
                    case Done:
                        println "Customer: $name: I have been served."
                        stop();
                        break
                }
            }
        }
    }
}
class Enter { Customer customer }
class Full {}
class Wait {}
class Next {}
class Start {}
class Done {}
def customers = []
customers << new Customer(name:'Joe', localBarbers:waitingRoom).start()
customers << new Customer(name:'Dave', localBarbers:waitingRoom).start()
customers << new Customer(name:'Alice', localBarbers:waitingRoom).start()
sleep 15000
customers << new Customer(name: 'James', localBarbers: waitingRoom).start()
sleep 5000
customers*.join()
barber.stop()
waitingRoom.stop()

import groovyx.gpars.actor.DefaultActor
import groovyx.gpars.actor.Actors
Actors.defaultActorPGroup.resize 5
final class Philosopher extends DefaultActor {
    private Random random = new Random()
    String name
    def forks = []
    void act() {
        assert 2 == forks.size()
        loop {
            think()
            forks*.send new Take()
            def messages = []
            react {a ->
                messages << [a, sender]
                react {b ->
                    messages << [b, sender]
                    if ([a, b].any {Rejected.isCase it}) {
                        println "$name: tOops, can't get my forks! Giving up."
                        final def accepted = messages.find {Accepted.isCase it[0]}
                        if (accepted!=null) accepted[1].send new Finished()
                    } else {
                        eat()
                        reply new Finished()
                    }
                }
            }
        }
    }
    void think() {
        println "$name: tI'm thinking"
        Thread.sleep random.nextInt(5000)
        println "$name: tI'm done thinking"
    }
    void eat() {
        println "$name: tI'm EATING"
        Thread.sleep random.nextInt(2000)
        println "$name: tI'm done EATING"
    }
}
final class Fork extends DefaultActor {
    String name
    boolean available = true
    void act() {
        loop {
            react {message ->
                switch (message) {
                    case Take:
                        if (available) {
                            available = false
                            reply new Accepted()
                        } else reply new Rejected()
                        break
                    case Finished:
                        assert !available
                        available = true
                        break
                    default: throw new IllegalStateException("Cannot process the message: $message")
                }
            }
        }
    }
}
final class Take {}
final class Accepted {}
final class Rejected {}
final class Finished {}
def forks = [
        new Fork(name:'Fork 1'),
        new Fork(name:'Fork 2'),
        new Fork(name:'Fork 3'),
        new Fork(name:'Fork 4'),
        new Fork(name:'Fork 5')
]
def philosophers = [
        new Philosopher(name:'Joe', forks:[forks[0], forks[1]]),
        new Philosopher(name:'Dave', forks:[forks[1], forks[2]]),
        new Philosopher(name:'Alice', forks:[forks[2], forks[3]]),
        new Philosopher(name:'James', forks:[forks[3], forks[4]]),
        new Philosopher(name:'Phil', forks:[forks[4], forks[0]]),
]
forks*.start()
philosophers*.start()
sleep 10000
forks*.stop()
philosophers*.stop()

//Messages
private final class FileToSort { String fileName }
private final class SortResult { String fileName; List<String> words }
//Worker actor
class WordSortActor extends DefaultActor {
    private List<String> sortedWords(String fileName) {
        parseFile(fileName).sort {it.toLowerCase()}
    }
    private List<String> parseFile(String fileName) {
        List<String> words = []
        new File(fileName).splitEachLine(' ') {words.addAll(it)}
        return words
    }
    void act() {
        loop {
            react {message ->
                switch (message) {
                    case FileToSort:
                        println "Sorting file=${message.fileName} on thread ${Thread.currentThread().name}"
                        reply new SortResult(fileName: message.fileName, words: sortedWords(message.fileName))
                }
            }
        }
    }
}
//Master actor
final class SortMaster extends DefaultActor {
    String docRoot = '/'
    int numActors = 1
    List<List<String>> sorted = []
    private CountDownLatch startupLatch = new CountDownLatch(1)
    private CountDownLatch doneLatch
    private void beginSorting() {
        int cnt = sendTasksToWorkers()
        doneLatch = new CountDownLatch(cnt)
    }
    private List createWorkers() {
        return (1..numActors).collect {new WordSortActor().start()}
    }
    private int sendTasksToWorkers() {
        List<Actor> workers = createWorkers()
        int cnt = 0
        new File(docRoot).eachFile {
            workers[cnt % numActors] << new FileToSort(fileName: it)
            cnt += 1
        }
        return cnt
    }
    public void waitUntilDone() {
        startupLatch.await()
        doneLatch.await()
    }
    void act() {
        beginSorting()
        startupLatch.countDown()
        loop {
            react {
                switch (it) {
                    case SortResult:
                        sorted << it.words
                        doneLatch.countDown()
                        println "Received results for file=${it.fileName}"
                }
            }
        }
    }
}
//start the actors to sort words
def master = new SortMaster(docRoot: 'c:/tmp/Logs/', numActors: 5).start()
master.waitUntilDone()
println 'Done'
File file = new File("c:/tmp/Logs/sorted_words.txt")
file.withPrintWriter { printer ->
    master.sorted.each { printer.println it }
}

import groovyx.gpars.actor.Actor
import groovyx.gpars.actor.DefaultActor
/**
 * Demonstrates work balancing among adaptable set of workers.
 * The load balancer receives tasks and queues them in a temporary task queue.
 * When a worker finishes his assignment, it asks the load balancer for a new task.
 * If the load balancer doesn't have any tasks available in the task queue, the worker is stopped.
 * If the number of tasks in the task queue exceeds certain limit, a new worker is created
 * to increase size of the worker pool.
 */
final class LoadBalancer extends DefaultActor {
    int workers = 0
    List taskQueue = []
    private static final QUEUE_SIZE_TRIGGER = 10
    void act() {
        loop {
            react { message ->
                switch (message) {
                    case NeedMoreWork:
                        if (taskQueue.size() == 0) {
                            println 'No more tasks in the task queue. Terminating the worker.'
                            reply DemoWorker.EXIT
                            workers -= 1
                        } else reply taskQueue.remove(0)
                        break
                    case WorkToDo:
                        taskQueue << message
                        if ((workers == 0) || (taskQueue.size() >= QUEUE_SIZE_TRIGGER)) {
                            println 'Need more workers. Starting one.'
                            workers += 1
                            new DemoWorker(this).start()
                        }
                }
                println "Active workers=${workers}tTasks in queue=${taskQueue.size()}"
            }
        }
    }
}
final class DemoWorker extends DefaultActor {
    final static Object EXIT = new Object()
    private static final Random random = new Random()
    Actor balancer
    def DemoWorker(balancer) {
        this.balancer = balancer
    }
    void act() {
        loop {
            this.balancer << new NeedMoreWork()
            react {
                switch (it) {
                    case WorkToDo:
                        processMessage(it)
                        break
                    case EXIT: terminate()
                }
            }
        }
    }
    private void processMessage(message) {
        synchronized (random) {
            Thread.sleep random.nextInt(5000)
        }
    }
}
final class WorkToDo {}
final class NeedMoreWork {}
final Actor balancer = new LoadBalancer().start()
//produce tasks
for (i in 1..20) {
    Thread.sleep 100
    balancer << new WorkToDo()
}
//produce tasks in a parallel thread
Thread.start {
    for (i in 1..10) {
        Thread.sleep 1000
        balancer << new WorkToDo()
    }
}
Thread.sleep 35000  //let the queues get empty
balancer << new WorkToDo()
balancer << new WorkToDo()
Thread.sleep 10000
balancer.stop()
balancer.join()

agent = new Agent(0)  //created a new Agent wrapping an integer with initial value 0
agent.send {increment()}  //asynchronous send operation, sending the increment() function
…
//after some delay to process the message the internal Agent's state has been updated
…
assert agent.val== 1

import groovyx.gpars.agent.Agent
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
/**
 * Create a new Agent wrapping a list of strings
 */
def jugMembers = new Agent<List<String>>(['Me'])  //add Me
jugMembers.send {it.add 'James'}  //add James
final Thread t1 = Thread.start {
    jugMembers.send {it.add 'Joe'}  //add Joe
}
final Thread t2 = Thread.start {
    jugMembers << {it.add 'Dave'}  //add Dave
    jugMembers {it.add 'Alice'}    //add Alice (using the implicit call() method)
}
[t1, t2]*.join()
println jugMembers.val
jugMembers.valAsync {println "Current members: $it"}
jugMembers.await()

import groovyx.gpars.agent.Agent
/**
 * Conference stores number of registrations and allows parties to register and unregister.
 * It inherits from the Agent class and adds the register() and unregister() private methods,
 * which callers may use it the commands they submit to the Conference.
 */
class Conference extends Agent<Long> {
    def Conference() { super(0) }
    private def register(long num) { data += num }
    private def unregister(long num) { data -= num }
}
final Agent conference = new Conference()  //new Conference created
/**
 * Three external parties will try to register/unregister concurrently
 */
final Thread t1 = Thread.start {
    conference << {register(10L)}               //send a command to register 10 attendees
}
final Thread t2 = Thread.start {
    conference << {register(5L)}                //send a command to register 5 attendees
}
final Thread t3 = Thread.start {
    conference << {unregister(3L)}              //send a command to unregister 3 attendees
}
[t1, t2, t3]*.join()
assert 12L == conference.val

def jugMembers = Agent.agent ['Me']  //add Me

final Agent counter = new Agent()
counter.addListener {oldValue, newValue -> println "Changing value from $oldValue to $newValue"}
counter.addListener {agent, oldValue, newValue -> println "Agent $agent changing value from $oldValue to $newValue"}
counter.addValidator {oldValue, newValue -> if (oldValue > newValue) throw new IllegalArgumentException('Things can only go up in Groovy')}
counter.addValidator {agent, oldValue, newValue -> if (oldValue == newValue) throw new IllegalArgumentException('Things never stay the same for $agent')}
counter 10
counter 11
counter {updateValue 12}
counter 10  //Will be rejected
counter {updateValue it - 1}  //Will be rejected
counter {updateValue it}  //Will be rejected
counter {updateValue 11}  //Will be rejected
counter 12  //Will be rejected
counter 20
counter.await()

assert counter.hasErrors()
assert counter.errors.size() == 5

//Create an agent storing names, rejecting 'Joe'
final Closure rejectJoeValidator = {oldValue, newValue -> if ('Joe' in newValue) throw new IllegalArgumentException('Joe is not allowed to enter our list.')}
Agent agent = new Agent([])
agent.addValidator rejectJoeValidator
agent {it << 'Dave'}                    //Accepted
agent {it << 'Joe'}                     //Erroneously accepted, since by-passes the validation mechanism
println agent.val
//Solution 1 - never alter the supplied state object
agent = new Agent([])
agent.addValidator rejectJoeValidator
agent {updateValue(['Dave', * it])}      //Accepted
agent {updateValue(['Joe', * it])}       //Rejected
println agent.val
//Solution 2 - use custom copy strategy on the agent
agent = new Agent([], {it.clone()})
agent.addValidator rejectJoeValidator
agent {updateValue it << 'Dave'}        //Accepted
agent {updateValue it << 'Joe'}         //Rejected, since 'it' is now just a copy of the internal agent's state
println agent.val

final def group = new NonDaemonPGroup(5)  //create a group around a thread pool
def jugMembers = group.agent(['Me'])  //add Me

def jugMembers = new Agent<List<String>>(['Me'])  //add Me
final ExecutorService pool = Executors.newFixedThreadPool(10)
jugMembers.attachToThreadPool(new DefaultPool(pool))

import groovyx.gpars.agent.Agent
class ShoppingCart {
    private def cartState = new Agent([:])
//----------------- public methods below here ----------------------------------
    public void addItem(String product, int quantity) {
        cartState << {it[product] = quantity}  //the << operator sends
                                               //a message to the Agent
    }    public void removeItem(String product) {
        cartState << {it.remove(product)}
    }    public Object listContent() {
        return cartState.val
    }    public void clearItems() {
        cartState << performClear
    }
    public void increaseQuantity(String product, int quantityChange) {
        cartState << this.&changeQuantity.curry(product, quantityChange)
    }
//----------------- private methods below here ---------------------------------
    private void changeQuantity(String product, int quantityChange, Map items) {
        items[product] = (items[product] ?: 0) + quantityChange
    }    private Closure performClear = { it.clear() }
}
//----------------- script code below here -------------------------------------
final ShoppingCart cart = new ShoppingCart()
cart.addItem 'Pilsner', 10
cart.addItem 'Budweisser', 5
cart.addItem 'Staropramen', 20
cart.removeItem 'Budweisser'
cart.addItem 'Budweisser', 15
println "Contents ${cart.listContent()}"
cart.increaseQuantity 'Budweisser', 3
println "Contents ${cart.listContent()}"
cart.clearItems()
println "Contents ${cart.listContent()}"

public void addItem(String product, int quantity) {
    cartState[product]=quantity
}

public void addItem(String product, int quantity) {
    cartState << {it[product] = quantity}
}

public void clearItems() {
    cartState << performClear
}
private Closure performClear = { it.clear() }

import groovyx.gpars.agent.Agent
/**
 * A non-thread-safe service that slowly prints documents on at a time
 */
class PrinterService {
    String document
    String quality
    public void printDocument() {
        println "Printing $document in $quality quality"
        Thread.sleep 5000
        println "Done printing $document"
    }
}
def printer = new Agent<PrinterService>(new PrinterService())
final Thread thread1 = Thread.start {
    for (num in (1..3)) {
        final String text = "document $num"
        printer << {printerService ->
            printerService.document = text
            printerService.quality = 'High'
            printerService.printDocument()
        }
        Thread.sleep 200
    }
    println 'Thread 1 is ready to do something else. All print tasks have been submitted'
}
final Thread thread2 = Thread.start {
    for (num in (1..4)) {
        final String text = "picture $num"
        printer << {printerService ->
            printerService.document = text
            printerService.quality = 'Medium'
            printerService.printDocument()
        }
        Thread.sleep 500
    }
    println 'Thread 2 is ready to do something else. All print tasks have been submitted'
}
[thread1, thread2]*.join()
printer.await()

def jugMembers = new Agent<List>()
    assert jugMembers.errors.empty
    jugMembers.send {throw new IllegalStateException('test1')}
    jugMembers.send {throw new IllegalArgumentException('test2')}
    jugMembers.await()
    List errors = jugMembers.errors
    assert 2 == errors.size()
    assert errors[0] instanceof IllegalStateException
    assert 'test1' == errors[0].message
    assert errors[1] instanceof IllegalArgumentException
    assert 'test2' == errors[1].message
    assert jugMembers.errors.empty

def jugMembers = new Agent<List>(['Me'])  //add Me
    jugMembers.makeFair()

import static groovyx.gpars.dataflow.Dataflow.task
final def x = new DataflowVariable()
final def y = new DataflowVariable()
final def z = new DataflowVariable()
task {
    z << x.val + y.val
}
task {
    x << 10
}
task {
    y << 5
}
println "Result: ${z.val}"

import static groovyx.gpars.dataflow.Dataflow.task
final def df = new Dataflows()
task {
    df.z = df.x + df.y
}
task {
    df.x = 10
}
task {
    df.y = 5
}
println "Result: ${df.z}"

import static groovyx.gpars.dataflow.Dataflow.task
def words = ['Groovy', 'fantastic', 'concurrency', 'fun', 'enjoy', 'safe', 'GPars', 'data', 'flow']
final def buffer = new DataflowQueue()
task {
    for (word in words) {
        buffer << word.toUpperCase()  //add to the buffer
    }
}
task {
    while(true) println buffer.val  //read from the buffer in a loop
}

DataflowWriteChannel broadcastStream = new DataflowBroadcast()
DataflowReadChannel stream1 = broadcastStream.createReadChannel()
DataflowReadChannel stream2 = broadcastStream.createReadChannel()
broadcastStream << 'Message1'
broadcastStream << 'Message2'
broadcastStream << 'Message3'
assert stream1.val == stream2.val
assert stream1.val == stream2.val
assert stream1.val == stream2.val

import groovyx.gpars.dataflow.stream.DataflowStream
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.scheduler.ResizeablePool
/**
 * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks
 *
 * In principle, the algorithm consists of a concurrently run chained filters,
 * each of which detects whether the current number can be divided by a single prime number.
 * (generate nums 1, 2, 3, 4, 5, ...) -> (filter by mod 2) -> (filter by mod 3) -> (filter by mod 5) -> (filter by mod 7) -> (filter by mod 11) -> (caution! Primes falling out here)
 * The chain is built (grows) on the fly, whenever a new prime is found
 */
/**
 * We need a resizeable thread pool, since tasks consume threads while waiting blocked for values at DataflowQueue.val
 */
group = new DefaultPGroup(new ResizeablePool(true))
final int requestedPrimeNumberCount = 100
/**
 * Generating candidate numbers
 */
final DataflowStream candidates = new DataflowStream()
group.task {
    candidates.generate(2, {it + 1}, {it < 1000})
}
/**
 * Chain a new filter for a particular prime number to the end of the Sieve
 * @param inChannel The current end channel to consume
 * @param prime The prime number to divide future prime candidates with
 * @return A new channel ending the whole chain
 */
def filter(DataflowStream inChannel, int prime) {
    inChannel.filter { number ->
        group.task {
            number % prime != 0
        }
    }
}
/**
 * Consume Sieve output and add additional filters for all found primes
 */
def currentOutput = candidates
requestedPrimeNumberCount.times {
    int prime = currentOutput.first
    println "Found: $prime"
    currentOutput = filter(currentOutput, prime)
}

import groovyx.gpars.dataflow.DataflowQueue
import groovyx.gpars.dataflow.stream.DataflowStream
import groovyx.gpars.dataflow.stream.DataflowStreamReadAdapter
import groovyx.gpars.dataflow.stream.DataflowStreamWriteAdapter
import static groovyx.gpars.dataflow.Dataflow.selector
import static groovyx.gpars.dataflow.Dataflow.operator
/**
 * Demonstrates the use of DataflowStreamAdapters to allow dataflow operators to use DataflowStreams
 */
final DataflowStream a = new DataflowStream()
final DataflowStream b = new DataflowStream()
def aw = new DataflowStreamWriteAdapter(a)
def bw = new DataflowStreamWriteAdapter(b)
def ar = new DataflowStreamReadAdapter(a)
def br = new DataflowStreamReadAdapter(b)
def result = new DataflowQueue()
def op1 = operator(ar, bw) {
    bindOutput it
}
def op2 = selector([br], [result]) {
    result << it
}
aw << 1
aw << 2
aw << 3
assert([1, 2, 3] == [result.val, result.val, result.val])
op1.stop()
op2.stop()
op1.join()
op2.join()

import groovyx.gpars.dataflow.Select
import groovyx.gpars.dataflow.stream.DataflowStream
import groovyx.gpars.dataflow.stream.DataflowStreamReadAdapter
import groovyx.gpars.dataflow.stream.DataflowStreamWriteAdapter
import static groovyx.gpars.dataflow.Dataflow.select
import static groovyx.gpars.dataflow.Dataflow.task
/**
 * Demonstrates the use of DataflowStreamAdapters to allow dataflow select to select on DataflowStreams
 */
final DataflowStream a = new DataflowStream()
final DataflowStream b = new DataflowStream()
def aw = new DataflowStreamWriteAdapter(a)
def bw = new DataflowStreamWriteAdapter(b)
def ar = new DataflowStreamReadAdapter(a)
def br = new DataflowStreamReadAdapter(b)
final Select<?> select = select(ar, br)
task {
    aw << 1
    aw << 2
    aw << 3
}
assert 1 == select().value
assert 2 == select().value
assert 3 == select().value
task {
    bw << 4
    aw << 5
    bw << 6
}
def result = (1..3).collect{select()}.sort{it.value}
assert result*.value == [4, 5, 6]
assert result*.index == [1, 0, 1]

def a = new DataflowVariable()
a >> {println "The variable has just been bound to $it"}
a.whenBound {println "Just to confirm that the variable has been really set to $it"}
...

def queue = new DataflowQueue()
queue.wheneverBound {println "A value $it arrived to the queue"}

Promise bookingPromise = task {
    final data = collectData()
    return broker.makeBooking(data)
}
…
bookingPromise.whenBound {booking -> printAgenda booking}
bookingPromise.whenBound {booking -> sendMeAnEmailTo booking}
bookingPromise.whenBound {booking -> updateTheCalendar booking}

final group = new NonDaemonPGroup()
    //Calling asynchronous services and receiving back promises for the reservations
    Promise flightReservation = flightBookingService('PRG <-> BRU')
    Promise hotelReservation = hotelBookingService('BRU:Feb 24 2009 - Feb 29 2009')
    Promise taxiReservation = taxiBookingService('BRU:Feb 24 2009 10:31')
    //when all reservations have been made we need to build an agenda for our trip
    Promise agenda = group.whenAllBound(flightReservation, hotelReservation, taxiReservation) {flight, hotel, taxi ->
        "Agenda: $flight | $hotel | $taxi"
    }
    //since this is a demo, we will only print the agenda and block till it is ready
    println agenda.val

Promise module1 = task {
    compile(module1Sources)
}
Promise module2 = task {
    compile(module2Sources)
}
//We don't know the number of modules that will be jarred together, so use a List
final jarCompiledModules = {List modules -> ...}
whenAllBound([module1, module2], jarCompiledModules)

final DataflowVariable variable = new DataflowVariable()
final DataflowVariable result = new DataflowVariable()
variable.then {it * 2} then {it + 1} then {result << it}
variable << 4
assert 9 == result.val

final DataflowVariable variable = new DataflowVariable()
final DataflowVariable result = new DataflowVariable()
final doubler = {it * 2}
final adder = {it + 1}
variable.then doubler then adder then {result << it}
Thread.start {variable << 4}
assert 9 == result.val

@ActiveObject
class ActiveDemoCalculator {
    @ActiveMethod
    def doubler(int value) {
        value * 2
    }
    @ActiveMethod
    def adder(int value) {
        value + 1
    }
}
final DataflowVariable result = new DataflowVariable()
final calculator = new ActiveDemoCalculator();
calculator.doubler(4).then {calculator.adder it}.then {result << it}
assert 9 == result.val

final DataflowVariable variable = new DataflowVariable()
final DataflowVariable result = new DataflowVariable()
final doubler = {it * 2}
final adder = {it + 1}
variable >> doubler >> adder >> {result << it}
Thread.start {variable << 4}
assert 9 == result.val

Promise<Integer> initial = new DataflowVariable<Integer>()
    Promise<String> result = initial.then {it * 2} then {100 / it}                  //Will throw exception for 0
            .then {println "Logging the value $it as it passes by"; return it}      //Since no error handler is defined, exceptions will be ignored
                                                                                    //and silently re-thrown to the next handler in the chain
            .then({"The result for $num is $it"}, {"Error detected for $num: $it"}) //Here the exception is caught
    initial << 0
    println result.get()

promise.then({it+1})                                                         //Implicitly re-throws potential exceptions bound to promise
promise.then({it+1}, {e -> throw e})                                         //Explicitly re-throws potential exceptions bound to promise
promise.then({it+1}, {e -> throw new RuntimeException('Error occurred', e})  //Explicitly re-throws a new exception wrapping a potential exception bound to promise

task {
    'gpars.codehaus.org'.toURL().text  //should throw MalformedURLException
}
.then {page -> page.toUpperCase()}
.then {page -> page.contains('GROOVY')}
.then({mentionsGroovy -> println "Groovy found: $mentionsGroovy"}, {error -> println "Error: $error"}).join()

url.then(download)
    .then(calculateHash, {MalformedURLException e -> return 0})
    .then(formatResult)
    .then(printResult, printError)
    .then(sendNotificationEmail);

`
Promise<Object> result = url.then(download).then(calculateHash).then(formatResult).then(printResult);
try {
    result.get()
} catch (Exception e) {
    //handle exceptions here
}

withPool {
    Closure download = {String url ->
        sleep 3000  //Simulate a web read
        'web content'
    }.asyncFun()
    Closure loadFile = {String fileName ->
        'file content'  //simulate a local file read
    }.asyncFun()
    Closure hash = {s -> s.hashCode()}
    Closure compare = {int first, int second ->
        first == second
    }
    Closure errorHandler = {println "Error detected: $it"}
    def all = whenAllBound([
                  download('http://www.gpars.org') >> hash,
                  loadFile('/coolStuff/gpars/website/index.html') >> hash
              ], compare).then({println it}, errorHandler)
    all.join()  //optionally block until the calculation is all done

promise.thenForkAndJoin(task1, task2, task3).then{...}

task {
    2
}.thenForkAndJoin({ it ** 2 }, { it**3 }, { it**4 }, { it**5 }).then({ println it}).join()

Closure<String> download = {url ->
    println "Downloading"
    url.toURL().text
}
def pageContent = new LazyDataflowVariable(download.curry("http://gpars.codehaus.org"))

def moduleA = new LazyDataflowVariable({->
    println "Loading moduleA into memory"
    sleep 3000
    println "Loaded moduleA into memory"
    return "moduleA"
})
def moduleB = new LazyDataflowVariable({->
    moduleA.then {
        println "->Loading moduleB into memory, since moduleA is ready"
        sleep 3000
        println "  Loaded moduleB into memory"
        return "moduleB"
    }
})
def moduleC = new LazyDataflowVariable({->
    moduleA.then {
        println "->Loading moduleC into memory, since moduleA is ready"
        sleep 3000
        println "  Loaded moduleC into memory"
        return "moduleC"
    }
})
def moduleD = new LazyDataflowVariable({->
    whenAllBound(moduleB, moduleC) { b, c ->
        println "-->Loading moduleD into memory, since moduleB and moduleC are ready"
        sleep 3000
        println "   Loaded moduleD into memory"
        return "moduleD"
    }
})
println "Nothing loaded so far"
println "==================================================================="
println "Load module: " + moduleD.get()
println "==================================================================="
println "All requested modules loaded"

import groovyx.gpars.dataflow.Dataflow
def pageContent = Dataflow.lazyTask {
        println "Downloading"
        "http://gpars.codehaus.org".toURL().text
    }
println "No-one has asked for the value just yet. Bound = ${pageContent.bound}"
sleep 1000
println "Now going to ask for a value"
println pageContent.get().size()
println "Repetitive requests will receive the already calculated value. No additional downloading."
println pageContent.get().size()

def initialDistance = new DataflowVariable()
def acceleration = new DataflowVariable()
def time = new DataflowVariable()
task {
    initialDistance << 100
    acceleration << 2
    time << 10
}
def result = initialDistance + acceleration*0.5*time**2
println 'Total distance ' + result.val

def name = new DataflowVariable()
task {
    name << '  adam   '
}
println name.toUpperCase().trim().val

def title = new DataflowVariable()
def searchPhrase = new DataflowVariable()
task {
    title << ' Groovy in Action 2nd edition   '
}
task {
    searchPhrase << '2nd'
}
println title.trim().contains(searchPhrase).val

def book = new DataflowVariable()
def searchPhrase = new DataflowVariable()
task {
    book << [
             title:'Groovy in Action 2nd edition   ',
             author:'Dierk Koenig',
             publisher:'Manning']
}
task {
    searchPhrase << '2nd'
}
book.title.trim().contains(searchPhrase).whenBound {println it}  //Asynchronous waiting
println book.title.trim().contains(searchPhrase).val  //Synchronous waiting

final DataflowVariable variable = new DataflowVariable()
        variable.getBindErrorManager().addBindErrorListener(new BindErrorListener() {
            @Override
            void onBindError(final Object oldValue, final Object failedValue, final boolean uniqueBind) {
                println "Bind failed!"
            }
            @Override
            void onBindError(final Object oldValue, final Throwable failedError) {
                println "Binding an error failed!"
            }
            @Override
            public void onBindError(final Throwable oldError, final Object failedValue, final boolean uniqueBind) {
                println "Bind failed!"
            }
            @Override
            public void onBindError(final Throwable oldError, final Throwable failedError) {
                println "Binding an error failed!"
            }
        })

import static groovyx.gpars.GParsPool.withPool
import groovyx.gpars.dataflow.DataflowVariable
import static groovyx.gpars.dataflow.Dataflow.task

/**
 * A simple mashup sample, downloads content of three websites
 * and checks how many of them refer to Groovy.
 */
def dzone = new DataflowVariable()
def jroller = new DataflowVariable()
def theserverside = new DataflowVariable()
task {
    println 'Started downloading from DZone'
    dzone << 'http://www.dzone.com'.toURL().text
    println 'Done downloading from DZone'
}
task {
    println 'Started downloading from JRoller'
    jroller << 'http://www.jroller.com'.toURL().text
    println 'Done downloading from JRoller'
}
task {
    println 'Started downloading from TheServerSide'
    theserverside << 'http://www.theserverside.com'.toURL().text
    println 'Done downloading from TheServerSide'
}
task {
    withPool {
        println "Number of Groovy sites today: " +
                ([dzone, jroller, theserverside].findAllParallel {
                    it.val.toUpperCase().contains 'GROOVY'
                }).size()
    }
}.join()

import groovyx.gpars.group.DefaultPGroup
def group = new DefaultPGroup()
group.with {
    task {
        …
    }
    task {
        …
    }
}

Dataflow.usingGroup(group) {
    task {
        'http://gpars.codehaus.org'.toURL().text  //should throw MalformedURLException
    }
    .then {page -> page.toUpperCase()}
    .then {page -> page.contains('GROOVY')}
    .then({mentionsGroovy -> println "Groovy found: $mentionsGroovy"}, {error -> println "Error: $error"}).join()
}

Dataflow.usingGroup(group) {
    anotherGroup.task {
        'http://gpars.codehaus.org'.toURL().text  //should throw MalformedURLException
    }
    .then(anotherGroup) {page -> page.toUpperCase()}
    .then(anotherGroup) {page -> page.contains('GROOVY')}.then(anotherGroup) {println Dataflow.retrieveCurrentDFPGroup();it}
    .then(anotherGroup, {mentionsGroovy -> println "Groovy found: $mentionsGroovy"}, {error -> println "Error: $error"}).join()
}

package groovyx.gpars.samples.dataflow
import static groovyx.gpars.GParsExecutorsPool.withPool
import groovyx.gpars.dataflow.DataflowVariable
import static groovyx.gpars.dataflow.Dataflow.task

/**
 * A simple mashup sample, downloads content of three websites and checks how many of them refer to Groovy.
 */
final List urls = ['http://www.dzone.com', 'http://www.jroller.com', 'http://www.theserverside.com']
task {
    def pages = urls.collect { downloadPage(it) }
    withPool {
        println "Number of Groovy sites today: " +
                (pages.findAllParallel {
                    it.val.toUpperCase().contains 'GROOVY'
                }).size()
    }
}.join()
def downloadPage(def url) {
    def page = new DataflowVariable()
    task {
        println "Started downloading from $url"
        page << url.toURL().text
        println "Done downloading from $url"
    }
    return page
}

import groovyx.gpars.dataflow.DataflowVariable
import static groovyx.gpars.dataflow.Dataflow.task
final def mass = new DataflowVariable()
final def radius = new DataflowVariable()
final def volume = new DataflowVariable()
final def density = new DataflowVariable()
final def acceleration = new DataflowVariable()
final def time = new DataflowVariable()
final def velocity = new DataflowVariable()
final def decelerationForce = new DataflowVariable()
final def deceleration = new DataflowVariable()
final def distance = new DataflowVariable()
def t = task {
    println """
Calculating distance required to stop a moving ball.
====================================================
The ball has a radius of ${radius.val} meters and is made of a material with ${density.val} kg/m3 density,
which means that the ball has a volume of ${volume.val} m3 and a mass of ${mass.val} kg.
The ball has been accelerating with ${acceleration.val} m/s2 from 0 for ${time.val} seconds and so reached a velocity of ${velocity.val} m/s.
Given our ability to push the ball backwards with a force of ${decelerationForce.val} N (Newton), we can cause a deceleration
of ${deceleration.val} m/s2 and so stop the ball at a distance of ${distance.val} m.
=======================================================================================================================
This example has been calculated asynchronously in multiple tasks using GPars Dataflow concurrency in Groovy.
Author: ${author.val}
"""
    System.exit 0
}
task {
    mass << volume.val * density.val
}
task {
    volume << Math.PI * (radius.val ** 3)
}
task {
    radius << 2.5
    density << 	998.2071  //water
    acceleration << 9.80665 //free fall
    decelerationForce << 900
}
task {
    println 'Enter your name:'
    def name = new InputStreamReader(System.in).readLine()
    author << (name?.trim()?.size()>0 ? name : 'anonymous')
}
task {
    time << 10
    velocity << acceleration.val * time.val
}
task {
    deceleration << decelerationForce.val / mass.val
}
task {
    distance << deceleration.val * ((velocity.val/deceleration.val) ** 2) * 0.5
}
t.join()

task {
    println a.val
    b << 'Hi there'
}
task {
    println b.val
    a << 'Hello man'
}

def df = new Dataflows()
df.x = 'value1'
assert df.x == 'value1'
Dataflow.task {df.y = 'value2}
assert df.y == 'value2'

new Dataflows().with {
    x = 'value1'
    assert x == 'value1'
    Dataflow.task {y = 'value2}
    assert y == 'value2'
}

final Promise t1 = task {
        return 10
    }
    final Promise t2 = task {
        return 20
    }
    def results = [t1, t2]*.val
    println 'Both sub-tasks finished and returned values: ' + results

def task = task {
    println 'The task is running and calculating the return value'
    30
}
task >> {value -> println "The task finished and returned $value"}

task {
     final Promise t1 = task {
         println 'First sub-task running.'
     }
     final Promise t2 = task {
         println 'Second sub-task running'
     }
     [t1, t2]*.join()
     println 'Both sub-tasks finished'
 }.join()

import groovyx.gpars.dataflow.DataflowQueue
import groovyx.gpars.dataflow.DataflowVariable
import static groovyx.gpars.dataflow.Dataflow.select
import static groovyx.gpars.dataflow.Dataflow.task
/**
 * Shows a basic use of Select, which monitors a set of input channels for values and makes these values
 * available on its output irrespective of their original input channel.
 * Note that dataflow variables and queues can be combined for Select.
 *
 * You might also consider checking out the prioritySelect method, which prioritizes values by the index of their input channel
 */
def a = new DataflowVariable()
def b = new DataflowVariable()
def c = new DataflowQueue()
task {
    sleep 3000
    a << 10
}
task {
    sleep 1000
    b << 20
}
task {
    sleep 5000
    c << 30
}
def select = select([a, b, c])
println "The fastest result is ${select().value}"

def sel = select(a, b, c, d)
def result = sel.select()                                       //Random selection
def result = sel()                                              //Random selection (a short-hand variant)
def result = sel.select([true, true, false, true])              //Random selection with guards specified
def result = sel([true, true, false, true])                     //Random selection with guards specified (a short-hand variant)
def result = sel.prioritySelect()                               //Priority selection
def result = sel.prioritySelect([true, true, false, true])      //Priority selection with guards specifies

def sel = select(a, b, c, d)
Promise result = sel.selectToPromise()                                       //Random selection
Promise result = sel.selectToPromise([true, true, false, true])              //Random selection with guards specified
Promise result = sel.prioritySelectToPromise()                               //Priority selection
Promise result = sel.prioritySelectToPromise([true, true, false, true])      //Priority selection with guards specifies

def handler = actor {...}
def sel = select(a, b, c, d)
sel.select(handler)                                         //Random selection
sel(handler)                                                //Random selection (a short-hand variant)
sel.select(handler, [true, true, false, true])              //Random selection with guards specified
sel(handler, [true, true, false, true])                     //Random selection with guards specified (a short-hand variant)
sel.prioritySelect(handler)                                 //Priority selection
sel.prioritySelect(handler, [true, true, false, true])      //Priority selection with guards specifies

def sel = select(leaders, seniors, experts, juniors)
def teamLead = sel([true, true, false, false]).value        //Only 'leaders' and 'seniors' qualify for becoming a teamLead here

import groovyx.gpars.dataflow.DataflowQueue
import static groovyx.gpars.dataflow.Dataflow.select
import static groovyx.gpars.dataflow.Dataflow.task
/**
 * Demonstrates the ability to enable/disable channels during a value selection on a select by providing boolean guards.
 */
final DataflowQueue operations = new DataflowQueue()
final DataflowQueue numbers = new DataflowQueue()
def t = task {
    final def select = select(operations, numbers)
    3.times {
        def instruction = select([true, false]).value
        def num1 = select([false, true]).value
        def num2 = select([false, true]).value
        final def formula = "$num1 $instruction $num2"
        println "$formula = ${new GroovyShell().evaluate(formula)}"
    }
}
task {
    operations << '+'
    operations << '+'
    operations << '*'
}
task {
    numbers << 10
    numbers << 20
    numbers << 30
    numbers << 40
    numbers << 50
    numbers << 60
}
t.join()

/**
 * Shows a basic use of Priority Select, which monitors a set of input channels for values and makes these values
 * available on its output irrespective of their original input channel.
 * Note that dataflow variables, queues and broadcasts can be combined for Select.
 * Unlike plain select method call, the prioritySelect call gives precedence to input channels with lower index.
 * Available messages from high priority channels will be served before messages from lower-priority channels.
 * Messages received through a single input channel will have their mutual order preserved.
 *
 */
def critical = new DataflowVariable()
def ordinary = new DataflowQueue()
def whoCares = new DataflowQueue()
task {
    ordinary << 'All working fine'
    whoCares << 'I feel a bit tired'
    ordinary << 'We are on target'
}
task {
    ordinary << 'I have just started my work. Busy. Will come back later...'
    sleep 5000
    ordinary << 'I am done for now'
}
task {
    whoCares << 'Huh, what is that noise'
    ordinary << 'Here I am to do some clean-up work'
    whoCares << 'I wonder whether unplugging this cable will eliminate that nasty sound.'
    critical << 'The server room goes on UPS!'
    whoCares << 'The sound has disappeared'
}
def select = select([critical, ordinary, whoCares])
println 'Starting to monitor our IT department'
sleep 3000
10.times {println "Received: ${select.prioritySelect().value}"}

import groovyx.gpars.dataflow.Promise
import groovyx.gpars.dataflow.Select
import groovyx.gpars.group.DefaultPGroup
/**
 * Demonstrates the use of dataflow tasks and selects to pick the fastest result of concurrently run calculations.
 */
final group = new DefaultPGroup()
group.with {
    Promise p1 = task {
        sleep(1000)
        10 * 10 + 1
    }
    Promise p2 = task {
        sleep(1000)
        5 * 20 + 2
    }
    Promise p3 = task {
        sleep(1000)
        1 * 100 + 3
    }
    final alt = new Select(group, p1, p2, p3)
    def result = alt.select()
    println "Result: " + result
}

import groovyx.gpars.dataflow.Promise
import groovyx.gpars.dataflow.Select
import groovyx.gpars.group.DefaultPGroup
/**
 * Demonstrates the use of dataflow tasks and selects to pick the fastest result of concurrently run calculations.
 */
final group = new DefaultPGroup()
group.with {
    Promise p1 = task {
        sleep(1000)
        10 * 10 + 1
    }
    Promise p2 = task {
        sleep(1000)
        5 * 20 + 2
    }
    Promise p3 = task {
        sleep(1000)
        1 * 100 + 3
    }
    final timeoutChannel = Select.createTimeout(500)
    final alt = new Select(group, p1, p2, p3, timeoutChannel)
    def result = alt.select()
    println "Result: " + result
}

import groovyx.gpars.dataflow.Promise
import groovyx.gpars.dataflow.Select
import groovyx.gpars.group.DefaultPGroup
import java.util.concurrent.atomic.AtomicBoolean
/**
 * Demonstrates the use of dataflow tasks and selects to pick the fastest result of concurrently run calculations.
 * It shows a waz to cancel the slower tasks once a result is known
 */
final group = new DefaultPGroup()
final done = new AtomicBoolean()
group.with {
    Promise p1 = task {
        sleep(1000)
        if (done.get()) return
        10 * 10 + 1
    }
    Promise p2 = task {
        sleep(1000)
        if (done.get()) return
        5 * 20 + 2
    }
    Promise p3 = task {
        sleep(1000)
        if (done.get()) return
        1 * 100 + 3
    }
    final alt = new Select(group, p1, p2, p3, Select.createTimeout(500))
    def result = alt.select()
    done.set(true)
    println "Result: " + result
}

operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
    …
    bindOutput 0, x + y + z
}

/**
 * CACHE
 *
 * Caches sites' contents. Accepts requests for url content, outputs the content. Outputs requests for download
 * if the site is not in cache yet.
 */
operator(inputs: [urlRequests], outputs: [downloadRequests, sites]) {request ->
    if (!request.content) {
        println "[Cache] Retrieving ${request.site}"
        def content = cache[request.site]
        if (content) {
            println "[Cache] Found in cache"
            bindOutput 1, [site: request.site, word:request.word, content: content]
        } else {
            def downloads = pendingDownloads[request.site]
            if (downloads != null) {
                println "[Cache] Awaiting download"
                downloads << request
            } else {
                pendingDownloads[request.site] = []
                println "[Cache] Asking for download"
                bindOutput 0, request
            }
        }
    } else {
        println "[Cache] Caching ${request.site}"
        cache[request.site] = request.content
        bindOutput 1, request
        def downloads = pendingDownloads[request.site]
        if (downloads != null) {
            for (downloadRequest in downloads) {
                println "[Cache] Waking up"
                bindOutput 1, [site: downloadRequest.site, word:downloadRequest.word, content: request.content]
            }
            pendingDownloads.remove(request.site)
        }
    }
}

operator(inputs:[a, b], outputs:[c, d]) {...}
splitter(c, [e, f])
selector(inputs:[e, d]: outputs:[]) {...}

def op1 = operator(inputs:[a, b], outputs:[c, d]) {...}
def sp1 = splitter(op1.outputs[0], [e, f])                            //takes the first output of op1
selector(inputs:[sp1.outputs[0], op1.outputs[1]]: outputs:[]) {...}   //takes the first output of sp1 and the second output of op1

import groovyx.gpars.group.DefaultPGroup
def group = new DefaultPGroup()
group.with {
    operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
        …
        bindOutput 0, x + y + z
    }
}

Dataflow.usingGroup(group) {
    operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
        …
        bindOutput 0, x + y + z
    }
}

Dataflow.usingGroup(group) {
    anotherGroup.operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
        …
        bindOutput 0, x + y + z
    }
}

import groovyx.gpars.dataflow.Dataflow
import groovyx.gpars.dataflow.DataflowQueue
import static groovyx.gpars.dataflow.ProcessingNode.node
/**
 * Shows how to build operators using the ProcessingNode class
 */
final DataflowQueue aValues = new DataflowQueue()
final DataflowQueue bValues = new DataflowQueue()
final DataflowQueue results = new DataflowQueue()
//Create a config and gradually set the required properties - channels, code, etc.
def adderConfig = node {valueA, valueB ->
    bindOutput valueA + valueB
}
adderConfig.inputs << aValues
adderConfig.inputs << bValues
adderConfig.outputs << results
//Build the operator
final adder = adderConfig.operator(Dataflow.DATA_FLOW_GROUP)
//Now the operator is running and processing the data
aValues << 10
aValues << 20
bValues << 1
bValues << 2
assert [11, 22] == (1..2).collect {
    results.val
}

int counter = 0
operator(inputs: [a], outputs: [b]) {value ->
    counter += 1
}

operator(inputs: [a], outputs: [b], stateObject: [counter: 0]) {value ->
    stateObject.counter += 1
}

def op = operator(inputs: [a, b, c], outputs: [d, e], maxForks: 2) {x, y, z ->
    bindOutput 0, x + y + z
    bindOutput 1, x * y * z
}

operator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg ->
    bindOutput 0, msg
    bindOutput 1, msg
}
inputChannel << 1
inputChannel << 2
inputChannel << 3
inputChannel << 4
inputChannel << 5

a -> 1, 3, 2, 4, 5
b -> 2, 1, 3, 5, 4

def lock = new Object()
operator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg ->
    doStuffThatIsThreadSafe()
    synchronized(lock) {
        doSomethingThatMustNotBeAccessedByMultipleThreadsAtTheSameTime()
        bindOutput 0, msg
        bindOutput 1, 2*msg
    }
}

operator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg ->
    doStuffThatIsThreadSafe()
        bindAllOutputValuesAtomically msg, 2*msg
    }
}

public interface DataflowEventListener {
    /**
     * Invoked immediately after the operator starts by a pooled thread before the first message is obtained
     *
     * @param processor The reporting dataflow operator/selector
     */
    void afterStart(DataflowProcessor processor);
    /**
     * Invoked immediately after the operator terminates
     *
     * @param processor The reporting dataflow operator/selector
     */
    void afterStop(DataflowProcessor processor);
    /**
     * Invoked if an exception occurs.
     * If any of the listeners returns true, the operator will terminate.
     * Exceptions outside of the operator's body or listeners' messageSentOut() handlers will terminate the operator irrespective of the listeners' votes.
     *
     * @param processor The reporting dataflow operator/selector
     * @param e         The thrown exception
     * @return True, if the operator should terminate in response to the exception, false otherwise.
     */
    boolean onException(DataflowProcessor processor, Throwable e);
    /**
     * Invoked when a message becomes available in an input channel.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The input channel holding the message
     * @param index     The index of the input channel within the operator
     * @param message   The incoming message
     * @return The original message or a message that should be used instead
     */
    Object messageArrived(DataflowProcessor processor, DataflowReadChannel<Object> channel, int index, Object message);
    /**
     * Invoked when a control message (instances of ControlMessage) becomes available in an input channel.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The input channel holding the message
     * @param index     The index of the input channel within the operator
     * @param message   The incoming message
     * @return The original message or a message that should be used instead
     */
    Object controlMessageArrived(DataflowProcessor processor, DataflowReadChannel<Object> channel, int index, Object message);
    /**
     * Invoked when a message is being bound to an output channel.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The output channel to send the message to
     * @param index     The index of the output channel within the operator
     * @param message   The message to send
     * @return The original message or a message that should be used instead
     */
    Object messageSentOut(DataflowProcessor processor, DataflowWriteChannel<Object> channel, int index, Object message);
    /**
     * Invoked when all messages required to trigger the operator become available in the input channels.
     *
     * @param processor The reporting dataflow operator/selector
     * @param messages  The incoming messages
     * @return The original list of messages or a modified/new list of messages that should be used instead
     */
    List<Object> beforeRun(DataflowProcessor processor, List<Object> messages);
    /**
     * Invoked when the operator completes a single run
     *
     * @param processor The reporting dataflow operator/selector
     * @param messages  The incoming messages that have been processed
     */
    void afterRun(DataflowProcessor processor, List<Object> messages);
    /**
     * Invoked when the fireCustomEvent() method is triggered manually on a dataflow operator/selector
     *
     * @param processor The reporting dataflow operator/selector
     * @param data      The custom piece of data provided as part of the event
     * @return A value to return from the fireCustomEvent() method to the caller (event initiator)
     */
    Object customEvent(DataflowProcessor processor, Object data);
}

final listener = new DataflowEventAdapter() {
    @Override
    Object customEvent(DataflowProcessor processor, Object data) {
        println "Log: Getting quite high on the scale $data"
        return 100  //The value to use instead
    }
}
op = group.operator(inputs: [a, b], outputs: [c], listeners: [listener]) {x, y ->
    final sum = x + y
    if (sum > 100) bindOutput(fireCustomEvent(sum))  //Reporting that the sum is too high, binding the lowered value that comes back
    else bindOutput sum
}

selector (inputs : [a, b, c], outputs : [d, e]) {value ->
    ....
}

selector (inputs : [a, b, c], outputs : [d, e]) {value, index ->
    ....
}

prioritySelector(inputs : [a, b, c], outputs : [d, e]) {value, index ->
    …
}

def join = selector (inputs : [programmers, analysis, managers], outputs : [employees, colleagues])

selector (inputs : [a, b, c], outputs : [d, e], maxForks : 5) {value ->
    ....
}

import groovyx.gpars.dataflow.DataflowQueue
import static groovyx.gpars.dataflow.Dataflow.selector
import static groovyx.gpars.dataflow.Dataflow.task
/**
 * Demonstrates the ability to enable/disable channels during a value selection on a select by providing boolean guards.
 */
final DataflowQueue operations = new DataflowQueue()
final DataflowQueue numbers = new DataflowQueue()
def instruction
def nums = []
selector(inputs: [operations, numbers], outputs: [], guards: [true, false]) {value, index ->   //initial guards is set here
    if (index == 0) {
        instruction = value
        setGuard(0, false)  //setGuard() used here
        setGuard(1, true)
    }
    else nums << value
    if (nums.size() == 2) {
        setGuards([true, false])                                    //setGuards() used here
        final def formula = "${nums[0]} $instruction ${nums[1]}"
        println "$formula = ${new GroovyShell().evaluate(formula)}"
        nums.clear()
    }
}
task {
    operations << '+'
    operations << '+'
    operations << '*'
}
task {
    numbers << 10
    numbers << 20
    numbers << 30
    numbers << 40
    numbers << 50
    numbers << 60
}

allMyProcessors*.terminate()

def op1 = operator(inputs: [a, b, c], outputs: [d, e]) {x, y, z -> }
def op2 = selector(inputs: [d], outputs: [f, out]) { }
def op3 = prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }
[op1, op2, op3]*.terminate()  //Terminate all operators by calling the terminate() method on them
op1.join()
op2.join()
op3.join()

def op1 = operator(inputs: [a, b, c], outputs: [d, e]) {x, y, z -> }
def op2 = selector(inputs: [d], outputs: [f, out]) { }
def op3 = prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }
a << PoisonPill.instance  //Send the poisson
op1.join()
op2.join()
op3.join()

def op1 = selector(inputs: [a, b, c], outputs: [d, e]) {value, index -> }
def op2 = selector(inputs: [d], outputs: [f, out]) { }
def op3 = prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }
a << PoisonPill.immediateInstance
[op1, op2, op3]*.join()

operator(inputs: [a, b, c], outputs: [d, e]) {x, y, z -> }
selector(inputs: [d], outputs: [f, out]) { }
prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }
//Send the poisson indicating the number of operators than need to be terminated before we can continue
final pill = new CountingPoisonPill(3)
a << pill
//Wait for all operators to terminate
pill.join()
//At least 3 operators should be terminated by now

//Send the poisson indicating the number of operators than need to be terminated before we can continue
final pill = new CountingPoisonPill(3)
pill.termination.whenBound {println "Reporting asynchronously that the network has been stopped"}
a << pill
if (pill.termination.bound) println "Wow, that was quick. We are done already!"
else println "Things are being slow today. The network is still running."
//Wait for all operators to terminate
assert pill.termination.get()
//At least 3 operators should be terminated by now

import groovyx.gpars.dataflow.DataflowQueue
import groovyx.gpars.group.NonDaemonPGroup

def group = new NonDaemonPGroup()
final DataflowQueue q = new DataflowQueue()
// final destination
def customs = group.operator(inputs: [q], outputs: []) { value ->
    println "Customs received $value"
}
// big producer
def green = group.task {
    (1..100).each {
        q << 'green channel ' + it
        sleep 10
    }
}
// little producer
def red = group.task {
    (1..10).each {
        q << 'red channel ' + it
        sleep 15
    }
}
def terminator = group.operator(inputs: [green, red], outputs: []) { t1, t2 ->
    q << PoisonPill.instance
}
customs.join()
group.shutdown()

operator(networkLeavingChannel, otherNetworkEnteringChannel) {value ->
    if (!(value instanceOf PoisonPill)) bindOutput it
}

networkLeavingChannel.filter { !(it instanceOf PoisonPill) } into otherNetworkEnteringChannel

import groovyx.gpars.dataflow.DataflowBroadcast
import groovyx.gpars.dataflow.DataflowQueue
import groovyx.gpars.dataflow.operator.component.GracefulShutdownListener
import groovyx.gpars.dataflow.operator.component.GracefulShutdownMonitor
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.group.PGroup
PGroup group = new DefaultPGroup(10)
final a = new DataflowQueue()
final b = new DataflowQueue()
final c = new DataflowQueue()
final d = new DataflowQueue<Object>()
final e = new DataflowBroadcast<Object>()
final f = new DataflowQueue<Object>()
final result = new DataflowQueue<Object>()
final monitor = new GracefulShutdownMonitor(100);
def op1 = group.operator(inputs: [a, b], outputs: [c], listeners: [new GracefulShutdownListener(monitor)]) {x, y ->
    sleep 5
    bindOutput x + y
}
def op2 = group.operator(inputs: [c], outputs: [d, e], listeners: [new GracefulShutdownListener(monitor)]) {x ->
    sleep 10
    bindAllOutputs 2*x
}
def op3 = group.operator(inputs: [d], outputs: [f], listeners: [new GracefulShutdownListener(monitor)]) {x ->
    sleep 5
    bindOutput x + 40
}
def op4 = group.operator(inputs: [e.createReadChannel(), f], outputs: [result], listeners: [new GracefulShutdownListener(monitor)]) {x, y ->
    sleep 5
    bindOutput x + y
}
100.times{a << 10}
100.times{b << 20}
final shutdownPromise = monitor.shutdownNetwork()
100.times{assert 160 == result.val}
shutdownPromise.get()
[op1, op2, op3, op4]*.join()
group.shutdown()

def toUpperCase = {s -> s.toUpperCase()}
final encrypt = new DataflowQueue()
final DataflowReadChannel encrypted = encrypt | toUpperCase | {it.reverse()} | {'###encrypted###' + it + '###'}
encrypt << "I need to keep this message secret!"
encrypt << "GPars can build linear operator pipelines really easily"
println encrypted.val
println encrypted.val

def toUpperCase = {s -> s.toUpperCase()}
final encrypt = new DataflowQueue()
final DataflowReadChannel encrypted = encrypt.chainWith toUpperCase chainWith {it.reverse()} chainWith {'###encrypted###' + it + '###'}
encrypt << "I need to keep this message secret!"
encrypt << "GPars can build linear operator pipelines really easily"
println encrypted.val
println encrypted.val

def toUpperCase = {s -> s.toUpperCase()}
def save = {text ->
    //Just pretending to be saving the text to disk, database or whatever
    println 'Saving ' + text
}
final toEncrypt = new DataflowQueue()
final DataflowReadChannel encrypted = toEncrypt.chainWith toUpperCase chainWith {it.reverse()} chainWith {'###encrypted###' + it + '###'}
final DataflowQueue fork1 = new DataflowQueue()
final DataflowQueue fork2 = new DataflowQueue()
splitter(encrypted, [fork1, fork2])  //Split the data flow
fork1.chainWith save  //Hook in the save operation
//Hook in a sneaky decryption pipeline
final DataflowReadChannel decrypted = fork2.chainWith {it[15..-4]} chainWith {it.reverse()} chainWith {it.toLowerCase()}
      .chainWith {'Groovy leaks! Check out a decrypted secret message: ' + it}
toEncrypt << "I need to keep this message secret!"
toEncrypt << "GPars can build operator pipelines really easy"
println decrypted.val
println decrypted.val

final encrypt = new DataflowQueue()
final DataflowWriteChannel messagesToSave = new DataflowQueue()
encrypt.chainWith toUpperCase chainWith {it.reverse()} into messagesToSave
task {
    encrypt << "I need to keep this message secret!"
    encrypt << "GPars can build operator pipelines really easy"
}
task {
    2.times {
        println "Saving " + messagesToSave.val
    }
}

final encrypt = new DataflowQueue()
final DataflowWriteChannel messagesToSave = new DataflowQueue()
final DataflowWriteChannel messagesToLog = new DataflowQueue()
encrypt.chainWith toUpperCase chainWith {it.reverse()}.split(messagesToSave, messagesToLog)

queue.chainWith {it * 2}.tap(logChannel).chainWith{it + 1}.tap(logChannel).into(PrintChannel)

maleChannel.merge(femaleChannel) {m, f -> m.marry(f)}.into(mortgageCandidatesChannel)

queue1.separate([queue2, queue3, queue4]) {a -> [a-1, a, a+1]}

queue1.binaryChoice(queue2, queue3) {a -> a > 0}
queue1.choice([queue2, queue3, queue4]) {a -> a % 3}

final DataflowQueue queue1 = new DataflowQueue()
        final DataflowQueue queue2 = new DataflowQueue()
        final odd = {num -> num % 2 != 0 }
        queue1.filter(odd) into queue2
        (1..5).each {queue1 << it}
        assert 1 == queue2.val
        assert 3 == queue2.val
        assert 5 == queue2.val

final DataflowQueue queue1 = new DataflowQueue()
        final DataflowQueue queue2 = new DataflowQueue()
        final odd = {num ->
            if (num == 5) return null  //null values are normally passed on
            if (num % 2 != 0) return num
            else return NullObject.nullObject  //this value gets blocked
        }
        queue1.chainWith odd into queue2
        (1..5).each {queue1 << it}
        assert 1 == queue2.val
        assert 3 == queue2.val
        assert null == queue2.val

channel | {it * 2}
channel.chainWith(closure)
channel.chainWith(pool) {it * 2}
channel.chainWith(group) {it * 2}
channel.into(otherChannel)
channel.into(pool, otherChannel)
channel.into(group, otherChannel)
channel.split(otherChannel1, otherChannel2)
channel.split(otherChannels)
channel.split(pool, otherChannel1, otherChannel2)
channel.split(pool, otherChannels)
channel.split(group, otherChannel1, otherChannel2)
channel.split(group, otherChannels)
channel.tap(otherChannel)
channel.tap(pool, otherChannel)
channel.tap(group, otherChannel)
channel.merge(otherChannel)
channel.merge(otherChannels)
channel.merge(pool, otherChannel)
channel.merge(pool, otherChannels)
channel.merge(group, otherChannel)
channel.merge(group, otherChannels)
channel.filter( otherChannel)
channel.filter(pool, otherChannel)
channel.filter(group, otherChannel)
channel.binaryChoice( trueBranch, falseBranch)
channel.binaryChoice(pool, trueBranch, falseBranch)
channel.binaryChoice(group, trueBranch, falseBranch)
channel.choice( branches)
channel.choice(pool, branches)
channel.choice(group, branches)
channel.separate( outputs)
channel.separate(pool, outputs)
channel.separate(group, outputs)

Dataflow.usingGroup(group) {
    channel.choice(branches)
}
//Is identical to
channel.choice(group, branches)

import groovyx.gpars.dataflow.DataflowQueue
import groovyx.gpars.dataflow.operator.Pipeline
import groovyx.gpars.scheduler.DefaultPool
import groovyx.gpars.scheduler.Pool
final DataflowQueue queue = new DataflowQueue()
final DataflowQueue result1 = new DataflowQueue()
final DataflowQueue result2 = new DataflowQueue()
final Pool pool = new DefaultPool(false, 2)
final negate = {-it}
final Pipeline pipeline = new Pipeline(pool, queue)
pipeline | {it * 2} | {it + 1} | negate
pipeline.split(result1, result2)
queue << 1
queue << 2
queue << 3
assert -3 == result1.val
assert -5 == result1.val
assert -7 == result1.val
assert -3 == result2.val
assert -5 == result2.val
assert -7 == result2.val
pool.shutdown()

new Pipeline(group, queue1).merge([maxForks: 4, listeners: [listener]], queue2) {a, b -> a + b}.into queue3

final DataflowVariable a = new DataflowVariable()
final Actor doubler = Actors.actor {
    react {message->
        a << 2 * message
    }
}
final Actor fakingDoubler = actor {
    react {
        doubler.send it  //send a number to the doubler
        println "Result ${a.val}"  //wait for the result to be bound to 'a'
    }
}
fakingDoubler << 10

import groovyx.gpars.dataflow.DataflowVariable
final DataflowVariable a = new DataflowVariable<String>()
final DataflowVariable b = new DataflowVariable<String>()
Thread.start {
    println "Received: $a.val"
    Thread.sleep 2000
    b << 'Thank you'
}
Thread.start {
    Thread.sleep 2000
    a << 'An important message from the second thread'
    println "Reply: $b.val"
}

import groovyx.gpars.dataflow.SyncDataflowQueue
import groovyx.gpars.group.NonDaemonPGroup
/**
 * Shows how synchronous dataflow queues can be used to throttle fast producer when serving data to a slow consumer.
 * Unlike when using asynchronous channels, synchronous channels block both the writer and the readers until all parties are ready to exchange messages.
 */
def group = new NonDaemonPGroup()
final SyncDataflowQueue channel = new SyncDataflowQueue()
def producer = group.task {
    (1..30).each {
        channel << it
        println "Just sent $it"
    }
    channel << -1
}
def consumer = group.task {
    while (true) {
        sleep 500  //simulating a slow consumer
        final Object msg = channel.val
        if (msg == -1) return
        println "Received $msg"
    }
}
consumer.join()
group.shutdown()

import groovyx.gpars.dataflow.SyncDataflowBroadcast
import groovyx.gpars.group.NonDaemonPGroup
/**
 * Shows how synchronous dataflow broadcasts can be used to throttle fast producer when serving data to slow consumers.
 * Unlike when using asynchronous channels, synchronous channels block both the writer and the readers until all parties are ready to exchange messages.
 */
def group = new NonDaemonPGroup()
final SyncDataflowBroadcast channel = new SyncDataflowBroadcast()
def subscription1 = channel.createReadChannel()
def fastConsumer = group.task {
    while (true) {
        sleep 10  //simulating a fast consumer
        final Object msg = subscription1.val
        if (msg == -1) return
        println "Fast consumer received $msg"
    }
}
def subscription2 = channel.createReadChannel()
def slowConsumer = group.task {
    while (true) {
        sleep 500  //simulating a slow consumer
        final Object msg = subscription2.val
        if (msg == -1) return
        println "Slow consumer received $msg"
    }
}
def producer = group.task {
    (1..30).each {
        println "Sending $it"
        channel << it
        println "Sent $it"
    }
    channel << -1
}
[fastConsumer, slowConsumer]*.join()
group.shutdown()

import groovyx.gpars.dataflow.SyncDataflowVariable
import groovyx.gpars.group.NonDaemonPGroup
final NonDaemonPGroup group = new NonDaemonPGroup()
final SyncDataflowVariable value = new SyncDataflowVariable(2)  //two readers required to exchange the message
def writer = group.task {
    println "Writer about to write a value"
    value << 'Hello'
    println "Writer has written the value"
}
def reader = group.task {
    println "Reader about to read a value"
    println "Reader has read the value: ${value.val}"
}
def slowReader = group.task {
    sleep 5000
    println "Slow reader about to read a value"
    println "Slow reader has read the value: ${value.val}"
}
[reader, slowReader]*.join()
group.shutdown()

import static groovyx.gpars.dataflow.ProcessingNode.node
import groovyx.gpars.dataflow.KanbanFlow
def producer = node { down -> down 1 }
def consumer = node { up   -> println up.take() }
new KanbanFlow().with {
    link producer to consumer
    start()
    // run for a while
    stop()
}

node { down -> down.send 1 }
node { down -> down << 1 }
node { down -> down 1 }

node { down -> down.release() }
node { down -> ~down }

def firstFlow = new KanbanFlow()
def producer  = node(counter)
def consumer  = node(repeater)
firstFlow.link(producer).to(consumer)
def secondFlow = new KanbanFlow()
def producer2  = node(repeater)
def consumer2  = node(reporter)
secondFlow.link(producer2).to(consumer2)
flow = firstFlow + secondFlow
flow.start()

flow.start(0) // start without trays
flow.start(1) // start with one tray per link in the flow
flow.start()  // start with the optimal number of trays

import groovyx.gpars.dataflow.DataflowQueue
import static groovyx.gpars.dataflow.Dataflow.task
/**
 * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks
 */
final int requestedPrimeNumberCount = 1000
final DataflowQueue initialChannel = new DataflowQueue()
/**
 * Generating candidate numbers
 */
task {
    (2..10000).each {
        initialChannel << it
    }
}
/**
 * Chain a new filter for a particular prime number to the end of the Sieve
 * @param inChannel The current end channel to consume
 * @param prime The prime number to divide future prime candidates with
 * @return A new channel ending the whole chain
 */
def filter(inChannel, int prime) {
    def outChannel = new DataflowQueue()
    task {
        while (true) {
            def number = inChannel.val
            if (number % prime != 0) {
                outChannel << number
            }
        }
    }
    return outChannel
}
/**
 * Consume Sieve output and add additional filters for all found primes
 */
def currentOutput = initialChannel
requestedPrimeNumberCount.times {
    int prime = currentOutput.val
    println "Found: $prime"
    currentOutput = filter(currentOutput, prime)
}

import groovyx.gpars.dataflow.DataflowQueue
       import static groovyx.gpars.dataflow.Dataflow.operator
       import static groovyx.gpars.dataflow.Dataflow.task
       /**
        * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks and operators
        */
       final int requestedPrimeNumberCount = 100
       final DataflowQueue initialChannel = new DataflowQueue()
       /**
        * Generating candidate numbers
        */
       task {
           (2..1000).each {
               initialChannel << it
           }
       }
       /**
        * Chain a new filter for a particular prime number to the end of the Sieve
        * @param inChannel The current end channel to consume
        * @param prime The prime number to divide future prime candidates with
        * @return A new channel ending the whole chain
        */
       def filter(inChannel, int prime) {
           def outChannel = new DataflowQueue()
           operator([inputs: [inChannel], outputs: [outChannel]]) {
               if (it % prime != 0) {
                   bindOutput it
               }
           }
           return outChannel
       }
       /**
        * Consume Sieve output and add additional filters for all found primes
        */
       def currentOutput = initialChannel
       requestedPrimeNumberCount.times {
           int prime = currentOutput.val
           println "Found: $prime"
           currentOutput = filter(currentOutput, prime)
       }

import groovyx.gpars.stm.GParsStm
import org.multiverse.api.references.TxnInteger
import static org.multiverse.api.StmUtils.newTxnInteger
public class Account {
    private final TxnInteger amount = newTxnInteger(0);
    public void transfer(final int a) {
        GParsStm.atomic {
            amount.increment(a);
        }
    }
    public int getCurrentAmount() {
        GParsStm.atomicWithInt {
            amount.get();
        }
    }
}

import groovyx.gpars.stm.GParsStm
import org.multiverse.api.AtomicBlock
import org.multiverse.api.PropagationLevel
final TxnExecutor block = GParsStm.createTxnExecutor(maxRetries: 3000, familyName: 'Custom', PropagationLevel: PropagationLevel.Requires, interruptible: false)
assert GParsStm.atomicWithBoolean(block) {
    true
}

import groovyx.gpars.stm.GParsStm
import org.multiverse.api.PropagationLevel
import org.multiverse.api.TxnExecutor
import static org.multiverse.api.StmUtils.newTxnInteger
final TxnExecutor block = GParsStm.createTxnExecutor(maxRetries: 3000, familyName: 'Custom', PropagationLevel: PropagationLevel.Requires, interruptible: false)
def counter = newTxnInteger(0)
final int max = 100
Thread.start {
    while (counter.atomicGet() < max) {
        counter.atomicIncrementAndGet(1)
        sleep 10
    }
}
assert max + 1 == GParsStm.atomicWithInt(block) { tx ->
    if (counter.get() == max) return counter.get() + 1
    tx.retry()
}

def group1 = new DefaultPGroup()
def group2 = new NonDaemonPGroup()
group1.with {
    task {...}
    task {...}
    def op = operator(...) {...}
    def actor = actor{...}
    def anotherActor = group2.actor{...}  //will belong to group2
    def agent = safe(0)
}

GParsPool.withPool {
    people.findAllParallel{it.isMale()}.collectParallel{it.name}.any{it == 'Joe'}
    people.parallel.filter{it.isMale()}.map{it.name}.filter{it == 'Joe'}.size() > 0
    people.parallelArray.withFilter({it.isMale()} as Predicate).withMapping({it.name} as Mapper).any{it == 'Joe'} != null
}

GParsPool.withPool(50) {
    …
}

actor {
    loop {
        react {msg ->
            switch(msg) {
                case String:…
                case Integer:…
            }
        }
    }
}

messageHandler {
    when{String msg -> ...}
    when{Integer msg -> ...}
}

class MyHandler extends DynamicDispatchActor {
    public void handleMessage(String msg) {
        …
    }
    public void handleMessage(Integer msg) {
        …
    }
}

def attackerGroup = new DefaultPGroup(new ResizeableFJPool(10))
def defenderGroup = new DefaultPGroup(new DefaultPool(5))
def attacker = attackerGroup.actor {...}
def defender = defenderGroup.messageHandler {...}
...

public final class GParsConfig {
    private static volatile PoolFactory poolFactory;
    private static volatile TimerFactory timerFactory;
    public static void setPoolFactory(final PoolFactory pool)
    public static PoolFactory getPoolFactory()
    public static Pool retrieveDefaultPool()
    public static void setTimerFactory(final TimerFactory timerFactory)
    public static TimerFactory getTimerFactory()
    public static GeneralTimer retrieveDefaultTimer(final String name, final boolean daemon)
    public static void shutdown()
}