GPars Actors

(1)
import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

DefaultActor receiver = Actors.actor { (2)
    loop { (3)
        react { String msg -> (4)
            println "Hey thanks for telling me $msg"
        }
    }
}

(5)
receiver.send 'Message 1'
receiver << 'Message 2'
receiver 'Message 3'

import groovy.transform.Immutable

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

@Immutable (1)
class UserItem {
    String userName
    Long itemId
}

final DefaultActor loggingServiceActor = Actors.actor {
    loop {
        react { UserItem userItem -> (2)
            println "user ${userItem.userName} added item ${userItem.itemId} at ${new Date()}"
        }
    }
}

(3)
loggingServiceActor.send(new UserItem(userName:'Mario', itemId: 10L))
Thread.sleep(1000) // Adding a delay to see it working
loggingServiceActor << ([userName: 'Rober', itemId: 20L] as UserItem)

import groovy.transform.Immutable

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

@Immutable
class UserItem {
    String userName
    Long itemId
}

class LoggingActorService extends DefaultActor { (1)
    void act() { (2)
        loop {
            react { UserItem userItem ->
                println """
                    user ${userItem.userName} added item ${userItem.itemId}
                    at ${new Date()}
                """
            }
        }
    }
}

class UserService {
    def loggingService (3)

    void trackAddedItem(final UserItem userItem) {
        loggingService << userItem
    }
}

def loggingService = new LoggingActorService()
def userService = new UserService(loggingService: loggingService)

loggingService.start() (4)

userService.trackAddedItem(new UserItem(userName: 'John', itemId: 43L))
Thread.sleep(1000)
userService.trackAddedItem(new UserItem(userName: 'Ronnie', itemId: 23L))

import static groovyx.gpars.actor.Actors.actor

final def english2SpanishService = actor {
    loop {
        react { String word ->
            if (word == 'hello') {
                reply 'hola' (1)
            } else if (word == 'goodbye') {
                reply 'adios'
            } else {
                reply 'no idea :P'
            }
        }
    }
}

String helloTranslation = english2SpanishService.sendAndWait('hello') (2)
String goodbyeTranslation = english2SpanishService.sendAndWait('goodbye')
String seeyoulaterTranslation = english2SpanishService.sendAndWait('see you later')

assert helloTranslation == 'hola'
assert goodbyeTranslation == 'adios'
assert seeyoulaterTranslation == 'no idea :P'

import static groovyx.gpars.actor.Actors.actor
import groovyx.gpars.dataflow.Promise

def reallySlowService = actor {
    loop {
        react {
            Thread.sleep(2000)
            reply 43
        }
    }
}

def reallyFastService = actor {
    loop {
        react {
            reply 'No idea'
        }
    }
}

Promise answer1 =
    reallySlowService.
        sendAndPromise('Tell me the meaning of life the universe and everything')
String answer2 =
    reallyFastService.
        sendAndWait('Tellme how is gonna end 24 this season')

assert answer1.get() == 43
assert answer2 == 'No idea'

import static groovyx.gpars.actor.Actors.actor
import groovyx.gpars.actor.DefaultActor

final DefaultActor echoActor = actor {
    loop {
        react { String message ->
            reply "Nice to hear: $message"
        }
    }
}

String response = echoActor.sendAndWait('Hi there')

assert response == "Nice to hear: Hi there"

import static groovyx.gpars.actor.Actors.messageHandler
import groovyx.gpars.actor.DynamicDispatchActor

DynamicDispatchActor handler = messageHandler {
    when { String name ->
       println "Nice to meet you $name"
    }
    when { Integer age ->
        println "You look younger than $age"
    }
}

handler << "Mario"
handler << 37

import groovyx.gpars.actor.DynamicDispatchActor

class Handler extends DynamicDispatchActor {
    void onMessage(String name) {
        println "Nice to meet you $name"
    }

    void onMessage(Integer age) {
        println "You look younger than $age"
    }

}

final Handler handler = new Handler()
handler.start()

handler << "Mario"
handler << 37

import static java.security.MessageDigest.getInstance

String md5(String st) {
    return getInstance("MD5").digest(st.getBytes("UTF-8")).encodeHex().toString()
}

import static groovyx.gpars.actor.Actors.actor

final receiver = actor {
    loop {
        react { msg ->
            reply "Replying actor received: '$msg'"
        }
    }
}

final sender = actor {
    receiver << "Sent Ping"
    react { msg ->
        println msg
    }
}

[receiver, sender]*.join()
println "Never reached"

import static groovyx.gpars.actor.Actors.actor
import groovyx.gpars.actor.DefaultActor

class SendingActor extends DefaultActor {
    final DefaultActor destination
    SendingActor(DefaultActor destination) {
       this.destination = destination
    }
    void act() {
        println destination.sendAndWait("Sent Ping")
    }
}

class ReplyingActor extends DefaultActor {
    void act() {
        loop {
            react { String msg ->
                reply "ReplyingActor received: $msg'"
            }
        }
    }
}

final receiver = new ReplyingActor().start()
final sender = new SendingActor(receiver).start()

[receiver, sender]*.join()
println "Never reached"

import groovyx.gpars.actor.DefaultActor

class WithId extends DefaultActor {
    String id
    String toString() { return this.id }
}

class Bank extends WithId {
    void act() {
        loop {
            react { Integer amount ->
                reply "Amount $amount has been stored in -- ${sender} --"
            }
        }
    }
}
class BankBranch extends WithId {
    void act() {
        loop {
            react { String message ->
                println message
            }
        }
    }
}

final branch1 = new BankBranch(id: '0001').start()
final branch2 = new BankBranch(id: '0002').start()
final bank = new Bank(id: 'SuperRichBank').start()

bank.send(1000, branch1)
bank.send(2000, branch2)

println bank.sendAndWait(3000)

@Grab('org.imgscalr:imgscalr-lib:4.2')
import java.awt.image.BufferedImage
import javax.imageio.ImageIO

import org.imgscalr.Scalr

BufferedImage src = ImageIO.read(new File('//source'))
BufferedImage img = Scalr.resize(src, 300, 200)

ImageIO.write(img, 'jpg', new File('//destination'))

/* called when the actor's thread gets interrupted. Thread interruption will result
 * in the stopping the actor in any case  */
onInterrupt(InterruptedException ie)

/* called when no messages are sent to the actor within the timeout specified for
 * the currently blocking react method */
onTimeout()

/* called when an exception occurs in the actor's event handler.
 * Actor will stop after return from this method. */
onException(Throwable th)

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

DefaultActor buggyActor = Actors.actor {
    delegate.metaClass.onException = {
        println "Close database connection"
        println "Release other resources"
        println "Bye"
    }
    loop {
        react { String message ->
            throw new Exception("Shit happens :P")
        }
    }
}

buggyActor << "Something nice"

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

class BuggyActor extends DefaultActor {

    void onException(Throwable th) {
        println "Close database connection"
        println "Release other resources"
        println "Bye"
    }

    void act() {
        loop {
            react { String message ->
                throw new Exception("Shit happens :P")
            }
        }
    }
}

new BuggyActor().start() << "Something nice"

import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor

class SupervisedExceptionMessage { (1)
   String message
}

class Supervised extends DefaultActor {
    Supervisor supervisor  (2)
    void onException(Throwable th) {
        supervisor << new SupervisedExceptionMessage(message: th.message) (3)
    }
}

import groovyx.gpars.actor.DynamicDispatchActor

class Supervisor extends DynamicDispatchActor { (1)
    Supervised supervised (2)
    Class<? extends Supervised> supervisedClass (3)

    void link(Supervised supervised) { (4)
        this.supervised = supervised
        this.supervised.supervisor = this
        this.supervisedClass = supervised.getClass()
    }
}

class SimpleSupervisedActor extends Supervised {
    void act() {
        loop {
            react { String message ->
                if (message == 'gift') {
                    throw new Exception(message)
                }
                println "NORMAL: $message"
            }
        }
    }
}

def supervised = new SimpleSupervisedActor().start()
def supervisor = new Supervisor() {
    void onMessage(String query) {
        supervised << query
    }
    void onMessage(SupervisedException ex) {
        println "SUPERVISED ACTOR DIES WITH MESSAGE: ${ex.message}"
        supervised = supervisedClass.newInstance().start() (1)
    }
}.start()

supervisor.link(supervised)

supervisor << "something"
Thread.sleep(1000)
supervisor << "something"
Thread.sleep(1000)
supervisor << "gift"
Thread.sleep(1000)
supervisor << "Message processed by a new supervised instance 1"
Thread.sleep(1000)
supervisor << "Message processed by a new supervised instance 2"

import groovyx.gpars.dataflow.DataflowVariable
import groovyx.gpars.dataflow.remote.RemoteDataflows

    RemoteDataflows startServer() {
        def dataflows = RemoteDataflows.create() (1)
        dataflows.startServer('localhost',9000) (2)
        def variable = new DataflowVariable() (3)
        variable << 42
        dataflows.publish(variable,"meaning-of-life") (4)

        return dataflows
    }

    void 'reading a simple dataflow variable'() {
        setup: 'starting server exposing dataflow variable'
            def serverDataflows = startServer() (1)
        when: 'listening to that server'
            def clientDataflows = RemoteDataflows.create() (2)
            def remoteVariablePromise = (3)
                clientDataflows.getVariable(
                    "localhost",
                    9000,
                    "meaning-of-life"
                )
        and: 'asking for a given dataflow variable'
            def result = remoteVariablePromise.get() (4)
        then: 'you should get what you expected'
            result.value == 42
        cleanup: 'stopping server instance'
            serverDataflows.stopServer() (5)
    }

import static groovyx.gpars.actor.Actors.actor
import groovyx.gpars.actor.remote.RemoteActors

    void 'simple remote actors test'() {
        given: 'a remote actor server'
            (1)
            def remoteActorServer = RemoteActors.create()
            remoteActorServer.startServer('localhost',9000)
        and: 'a simple DefaultActor instance'
            (2)
            def simpleActor = actor {
                loop {
                    react { String msg ->
                        reply "Thanks for saying: $msg"
                    }
                }
            }
        when: 'publishing the actor remotely'
            (3)
            remoteActorServer.publish(simpleActor, 'greetingsActor')
        and: 'getting the actor reference'
            (4)
            def remoteActorsClient = RemoteActors.create()
            def actorPromise =
                remoteActorsClient
                    .get('localhost', 9000, 'greetingsActor')
            (5)
            def remoteActor = actorPromise.get()
        and: 'sending a message to actor'
            (6)
            def result = remoteActor.sendAndWait('hello')
        then: 'the reply should be the expected'
             result == 'Thanks for saying: hello'
        cleanup: 'shutting down the actor server'
            remoteActorServer.stopServer()
    }

import static groovyx.gpars.actor.Actors.actor

def printer = actor {
    loop {
        react { List<Integer> numbers ->
           println numbers.sum()
        }
    }
}

printer << (1..4) as List

import groovyx.gpars.actor.DefaultActor

class Printer extends DefaultActor {
    void act() {
        loop {
            react { numbers ->
                println numbers.sum()
            }
        }
    }
}

def printer = new Printer()
printer.start()
printer << (1..4)

import static groovyx.gpars.actor.Actors.actor
import groovyx.gpars.dataflow.Promise

def printer = actor {
    loop {
        react { List<Integer> numbers ->
           reply numbers.sum()
        }
    }
}

Promise first = printer.sendAndPromise((1..4))
Integer second = printer.sendAndWait((4..8))
Integer third = printer.sendAndWait((8..12))

assert (first.get() + second + third) == 90

import groovyx.gpars.actor.DefaultActor
import groovyx.gpars.dataflow.Promise

class Printer extends DefaultActor {
    void act() {
        loop {
            react { numbers ->
                reply numbers.sum()
            }
        }
    }
}

def printer = new Printer()
printer.start()

Promise first = printer.sendAndPromise((1..4))
Integer second = printer.sendAndWait((4..8))
Integer third = printer.sendAndWait((8..12))

assert (first.get() + second + third) == 90

import static groovyx.gpars.actor.Actors.messageHandler
import static java.security.MessageDigest.getInstance

import groovyx.gpars.actor.DynamicDispatchActor
import groovyx.gpars.dataflow.Promise

class Order { String content }
class NationalOrder extends Order {}
class InternationalOrder extends Order {}

final Closure<String> md5 = { String st ->
    return getInstance("MD5").digest(st.getBytes("UTF-8")).encodeHex().toString()
}

final DynamicDispatchActor orderProcessor = messageHandler {
    when { InternationalOrder international ->
        reply md5(international.content)
    }
    when { NationalOrder national ->
        println "I'm sure the order: ${national.content} will be delivered"
    }
}

Promise md5IdPromise =
    orderProcessor.sendAndPromise(new InternationalOrder(content: 'supersecret'))

orderProcessor << new NationalOrder(content: 'something really silly')
orderProcessor << new NationalOrder(content: 'something as silly as the previous one')

println "I need to hold the id: ${md5IdPromise.get()}"

import static groovyx.gpars.actor.Actors.messageHandler
import static java.security.MessageDigest.getInstance

import groovyx.gpars.actor.DynamicDispatchActor
import groovyx.gpars.dataflow.Promise

class Order { String content }
class NationalOrder extends Order {}
class InternationalOrder extends Order {}

class OrderProcessor extends DynamicDispatchActor {
    String md5(String st) {
        return getInstance("MD5").digest(st.getBytes("UTF-8")).encodeHex().toString()
    }
    void onMessage(InternationalOrder international) {
        reply md5(international.content)
    }
    void onMessage(NationalOrder national) {
        println "I'm sure the order: ${national.content} will be delivered"
    }
}

final OrderProcessor orderProcessor = new OrderProcessor().start()
final Promise md5IdPromise =
    orderProcessor.sendAndPromise(new InternationalOrder(content: 'supersecret'))

orderProcessor << new NationalOrder(content: 'something really silly')
orderProcessor << new NationalOrder(content: 'something as silly as the previous one')

println "I need to hold the id: ${md5IdPromise.get()}"

@Grab('org.imgscalr:imgscalr-lib:4.2')
import static javax.imageio.ImageIO.read
import static javax.imageio.ImageIO.write
import static org.imgscalr.Scalr.resize
import static groovyx.gpars.actor.Actors.actor
import static java.io.File.createTempFile

import groovyx.gpars.actor.DefaultActor

final DefaultActor imageActor = actor {
    loop {
        react { File image ->
            def src = read(image)
            def img = resize(src, 400, 300)

            write(img, 'jpg', createTempFile('actors_', image.name))

            println "IMAGE: ${image.name} processed"
        }
    }
}

final DefaultActor textActor = actor {
    loop {
        react { File textFile ->
            def dst = createTempFile('actors_', textFile.name)

            // Don't do this at home :P
            dst << "Author: MYBOSS"
            dst << textFile.text

            println "TEXT: ${textFile.name} processed"
        }
    }

}

final DefaultActor dispatchingActor = actor {
    loop {
        react { File file ->
            if (file.name.endsWith('jpg')) {
                imageActor << file
            } else if (file.name.endsWith('txt')) {
                textActor << file
            }
        }
    }
}

new File('/somedirwithfiles/').
    listFiles().each { dispatchingActor << it }

package exercises.forwarding.images

import groovyx.gpars.actor.DefaultActor

class FileProcessorActor extends DefaultActor {

    DefaultActor imageActor
    DefaultActor textActor

    void act() {
        loop {
            react { File file2Process ->
                println "Start to processing ${file2Process.name}"
                if (isImageFile(file2Process)) {
                    sendToImageActor(file2Process)
                }
                if (isTextFile(file2Process)) {
                    sendToTextActor(file2Process)
                }
            }
        }
    }

    Boolean isImageFile(File file2Process) {
        return doesItEndWith(file2Process, 'jpg')
    }

    Boolean isTextFile(File file2Process) {
        return doesItEndWith(file2Process, 'txt')
    }

    Boolean doesItEndWith(File file, String ending) {
        return file.name.endsWith(ending)
    }

    void sendToImageActor(File file) {
        imageActor << file
    }

    void sendToTextActor(File file) {
        textActor << file
    }

}

package exercises.forwarding.images

import static javax.imageio.ImageIO.read
import static javax.imageio.ImageIO.write
import static org.imgscalr.Scalr.resize
import static java.io.File.createTempFile

import exercises.test.NotificationActor
import groovyx.gpars.actor.DefaultActor

class ImageProcessorActor extends NotificationActor {

    ImageProcessorActor(DefaultActor monitor) {
        super(monitor)
    }

    void act() {
        loop {
            react { File image ->
                def src = read(image)
                def img = resize(src, 300, 200)
                def tmp = new File(System.getProperty('java.io.tmpdir'))

                write(img, 'jpg', new File(tmp, "processed_${image.name}"))

                notifyIfMonitor(image.name)
            }
        }
    }

}

package exercises.forwarding.images

import static java.io.File.createTempFile

import exercises.test.NotificationActor
import groovyx.gpars.actor.DefaultActor

class TextProcessorActor extends NotificationActor {

    TextProcessorActor(DefaultActor monitor) {
        super(monitor)
    }

    void act() {
        loop {
            react { File textFile ->
                def tmpfile = new File(System.getProperty('java.io.tmpdir'))
                def newFile = new File(tmpfile, "processed_${textFile.name}")

                newFile << 'Author: me'
                newFile << textFile.text

                notifyIfMonitor(textFile.name)
            }
        }
    }

}

package exercises.forwarding.images

import static groovyx.gpars.actor.Actors.actor

import exercises.test.NotifiedActor
import spock.lang.Specification

class ImagesProcessingSpecification extends Specification {

    final File RESOURCES_DIR = 'src/test/resources/exercises/forwarding/images/' as File
    final File TEMPORARY_DIR = new File(System.getProperty('java.io.tmpdir'))

    def 'Processing successfully images'() {
        setup: 'Creating needed directories references'
            def monitorActor = new NotifiedActor(3).start()
            def imageActor   = new ImageProcessorActor(monitorActor).start()
            def textActor    = new TextProcessorActor(monitorActor).start()
            def proxyActor   = new FileProcessorActor(imageActor: imageActor, textActor: textActor).start()
        when: 'Sending a couple of image files'
            proxyActor << new File(RESOURCES_DIR, 'image1.jpg')
            proxyActor << new File(RESOURCES_DIR, 'image2.jpg')
            proxyActor << new File(RESOURCES_DIR, 'text.txt')
        and: 'Expecting process to finish'
            monitorActor.join()
        and: 'Looking for processed files'
            def file1 = new File(TEMPORARY_DIR, 'processed_text.txt')
            def file2 = new File(TEMPORARY_DIR, 'processed_image1.jpg')
            def file3 = new File(TEMPORARY_DIR, 'processed_image2.jpg')
        then: 'We should make sure they exist'
            [file1, file2, file3].every { it.exists() }
        cleanup: 'Deleting processed files'
            [file1, file2, file3]*.delete()
    }
}