Async/Await (Incubating)

def deck = ['2♠', '3♥', 'K♦', 'A♣']
def card = async { deck.shuffled()[0] }
println "You drew: ${await card}"

def drawFromEmpty = async {
    throw new IllegalStateException('deck is empty')
}
try {
    await drawFromEmpty
} catch (IllegalStateException e) {
    // Original exception — no CompletionException wrapper
    assert e.message == 'deck is empty'
}

// await works with CompletableFuture from Java libraries
def future = CompletableFuture.supplyAsync { 'A♠' }
assert await(future) == 'A♠'

// Deal cards to three players concurrently
def deck = [*1..9,'J','Q','K','A'].collectMany { rank ->
    ['♠','♥','♦','♣'].collect { suit -> "$rank$suit" }
}.shuffled()
def index = new AtomicInteger()
def draw5 = { -> def i = index.getAndAdd(5); deck[i..i+4] }

def alice = async { draw5() }
def bob   = async { draw5() }
def carol = async { draw5() }

def (a, b, c) = await Awaitable.all(alice, bob, carol)
assert a.size() == 5 && b.size() == 5 && c.size() == 5

def a = async { 1 }
def b = async { 2 }
def c = async { 3 }

// Parenthesized multi-arg await (sugar for Awaitable.all):
def results = await(a, b, c)
assert results == [1, 2, 3]

// Race two servers — use whichever responds first
def primary  = async { Thread.sleep(200); 'primary-response' }
def fallback = async { 'fallback-response' }

def response = await Awaitable.any(primary, fallback)
assert response == 'fallback-response'

// Try multiple sources — use the first that succeeds
def failing    = async { throw new RuntimeException('server down') }
def succeeding = async { 'card-data-from-cache' }

def result = await Awaitable.first(failing, succeeding)
assert result == 'card-data-from-cache'

def save1 = async { 42 }
def save2 = async { throw new RuntimeException('db error') }

def results = await Awaitable.allSettled(save1, save2)
assert results[0].success && results[0].value == 42
assert !results[1].success && results[1].error.message == 'db error'

def dealCards = async {
    def suits = ['♠', '♥', '♦', '♣']
    def ranks = ['A', '2', '3', '4', '5', '6', '7', '8', '9', '10', 'J', 'Q', 'K']
    for (suit in suits) {
        for (rank in ranks) {
            yield return "$rank$suit"
        }
    }
}
def cards = dealCards.collect()
assert cards.size() == 52
assert cards.first() == 'A♠'
assert cards.last() == 'K♣'

def scores = async {
    for (s in [100, 250, 75]) { yield return s }
}
// Generators return Iterable — regular for and collect work
assert scores.collect { it * 2 } == [200, 500, 150]

def source = AsyncChannel.create(10)
def evens = source.filter { it % 2 == 0 }

async {
    (1..10).each { source.send(it) }
    source.close()
}

for await (val in evens) { println val }  // 2, 4, 6, 8, 10

def doubled = source.map { it * 2 }

def pipeline = source
    .filter { it > 50 }
    .map { it * 2 }

for await (val in pipeline) { process(val) }

def merged = channelA.merge(channelB)
for await (event in merged) { handle(event) }

def (highPriority, normal) = orders.split { it.priority > 5 }

def monitor = AsyncChannel.create(100)
def output = source.tap(monitor)
// output gets all values; monitor gets copies for logging

def broadcast = BroadcastChannel.create()
def publisher = broadcast.asPublisher()

// Compose with any Reactive Streams operator, or consume with `for await`:
for await (msg in publisher) {
    println "Got: $msg"
}

def topCards = async {
    for (card in ['A♠', 'K♥', 'Q♦']) {
        yield return card
    }
}
def hand = []
for await (card in topCards) {
    hand << card
}
assert hand == ['A♠', 'K♥', 'Q♦']

import groovy.concurrent.Agent

def agent = Agent.create(0)
try {
    async {
        3.times { agent.send { it + 1 } }
        agent.shutdown()
    }
    for await (v in agent.changes()) {
        println "value is now $v"
    }
} finally {
    agent.shutdown()
}

def slowPlayer = async { Thread.sleep(5000); 'finally played' }
try {
    await Awaitable.orTimeoutMillis(slowPlayer, 100)
} catch (TimeoutException e) {
    // Player took too long — turn forfeited
    assert true
}

def slowPlayer = async { Thread.sleep(5000); 'deliberate move' }
def move = await Awaitable.completeOnTimeoutMillis(slowPlayer, 'auto-pass', 100)
assert move == 'auto-pass'

long start = System.currentTimeMillis()
await Awaitable.delay(100)   // pause without blocking a thread
assert System.currentTimeMillis() - start >= 90

// Without groovy-rxjava — manual conversion
def result = Single.just('hello').toCompletionStage().toCompletableFuture().join()

// With groovy-rxjava — adapter handles the plumbing
def result = await Awaitable.from(Single.just('hello'))

import org.apache.groovy.runtime.async.AsyncSupport
import java.util.concurrent.Executors

AsyncSupport.setExecutor(Executors.newFixedThreadPool(4))
AsyncSupport.resetExecutor()  // restore default

// Task 1: Run a process and await its completion
def proc = async {
    def p = echoCmd.execute() // echo Hello from Groovy
    await p.onExit()
    p.text.trim()
}

// Task 2: Read a file asynchronously
def fileContent = async {
    await etcPassword.textAsync
}

// Task 3: Fetch a web page using Groovy's HttpBuilder
def webContent = async {
    def http = HttpBuilder.http { baseUri bankUrl }
    http.get('/withdraw/100').body
}

// All three tasks run concurrently — collect results
var (echo, file, html) = await proc, fileContent, webContent
assert echo == 'Hello from Groovy'
assert file =~ /sEcrEt/
assert html == '<html>SUCCESS</html>'

import java.net.http.HttpClient
import java.net.http.HttpRequest
import java.net.http.HttpResponse

def webContent = async {
    def client = HttpClient.newHttpClient()
    def req = HttpRequest.newBuilder()
        .uri(URI.create('https://api.example.com/data'))
        .build()
    def resp = await client.sendAsync(req, HttpResponse.BodyHandlers.ofString())
    resp.body()
}

// A method that returns an async task — a simple factory
def dealCard(List deck) {
    async { deck.shuffled()[0] }
}

def numPlayers = 3
// each player gets a card from their own deck
def cards = (1..numPlayers).collect { dealCard(['A♠', 'K♥', 'Q♦', 'J♣']) }

def hands = await Awaitable.all(*cards)
assert hands.size() == 3
assert hands.every { it in ['A♠', 'K♥', 'Q♦', 'J♣'] }

// UNSAFE — shared mutation without synchronization
var count = 0
def tasks = (1..100).collect { async { count++ } }
tasks.each { await it }
// count may not be 100!

// SAFE — no shared mutation
def tasks = (1..100).collect { n -> async { n } }
def results = await Awaitable.all(*tasks)
assert results.sum() == 5050