Rhombus events

select_demo2.rhm

#lang rhombus/static/and_meta

import:
  lib("racket/base.rkt") as rkt:
    rename:
      #{evt?} as is_evt
      #{always-evt} as always_evt
      #{alarm-evt} as alarm_evt
      #{guard-evt} as guard_evt
      #{handle-evt} as handle_evt

      #{current-milliseconds} as current_milliseconds

annot.macro 'Evt': 'satisfying(rkt.is_evt)'

import rhombus/thread open

namespace ~open:
  export: Channel
  class Channel(hand):
    opaque
    internal _Channel
    constructor ():
      super(rkt.#{make-channel}())
    property handle: hand

    method put(v):
      rkt.#{channel-put}(hand, v)

    method get(v):
      rkt.#{channel-get}(hand)

    export:
      from_handle

  fun from_handle(hand) :: Channel:
    ~who: who
    unless rkt.#{channel?}(hand)
    | error(~who: who, "not a channel handle",
            error.val(hand))
    _Channel(hand)

interface Evtable:
  method to_evt

  export to_evt
  annot.macro 'to_evt': 'converting(fun(ev): to_evt(ev))'

fun | to_evt(evt :: Evt): evt
    | to_evt(ch :: Channel): ch.handle
    | to_evt(th :: Thread): th.handle
    | to_evt(sem :: Semaphore): sem.handle
    | to_evt(ev :: Evtable):
        rkt.guard_evt(fun ():
                        recur rec (ev = ev):
                          match Evtable.to_evt(ev)
                          | evt :: Evt: evt
                          | ev :: Evtable: rec(ev))

operator (ev :: Evtable.to_evt) ~> (fn :: Function):
  rkt.handle_evt(ev, fn)

meta:
  syntax_class SelectClause:
  | '$clause ...':
      field as_evt: 'to_evt($clause ...)'

  syntax_class SelectOpt:
    fields: opt
    root_swap: opt orig
  | '~timeout: $n':
      field opt = {#'timeout: n}
  | '~break':
      field opt = {#'break: #true}

  syntax_class SelectOptions:
    fields: opt
    root_swap: opt orig
  | '$(o :: SelectOpt); ...':
      field opt = Map.append(o, ...)

expr.macro
| 'select:
     $(opts :: SelectOptions)
   | $(clause :: SelectClause)
   | ...':
    let opts :~ Map: opts
    def timeout = opts.get(#'timeout, '#false')
    cond
    | opts.get(#'break, #false):
        'rkt.#{sync/timeout/enable-break}($timeout, $clause.as_evt, ...)'
    | opts.get(#'timeout, #false):
        'rkt.#{sync/timeout}($timeout, $clause.as_evt, ...)'
    | ~else:
        'rkt.sync($clause.as_evt, ...)'
| 'select
   | $(clause :: SelectClause)
   | ...':
    'rkt.sync($clause.as_evt, ...)'

block:
  fun race(n = math.random(10)):
    rkt.sleep(n)

  fun go(a, b):
    def th_a: thread: race(a)
    def th_b: thread: race(b)

    select:
    | th_a ~> fun (_): #'a
    | th_b ~> fun (_): #'b
    | rkt.guard_evt(fun ():
                      let t = rkt.current_milliseconds()
                      rkt.alarm_evt(t + 4000))
        ~> fun(_): #'expired

  check go(1, 2) ~is #'a
  check go(3, 2) ~is #'b
  check go(6, 6) ~is #'expired

block:
  fun race(ch :: Channel, n):
    rkt.sleep(n)
    ch.put(n)

  fun make_racer(n = math.random(10)):
    let ch: Channel()
    let thd: thread: race(ch, n)
    ch

  fun go(a, b):
    def ch_a: make_racer(a)
    def ch_b: make_racer(b)

    fun timeout_evt(s):
      rkt.guard_evt(fun ():
                      let t = rkt.current_milliseconds()
                      rkt.handle_evt(rkt.alarm_evt(t + s * 1000),
                                     fun (_): #'expired))

    select:
    | ch_a ~> fun(s): values(#'a, s)
    | ch_b ~> fun(s): values(#'b, s)
    | timeout_evt(4) ~> fun(msg): values(msg, -1)

  check go(1, 6) ~is values(#'a, 1)
  check go(6, 1) ~is values(#'b, 1)
  check go(6, 6) ~is values(#'expired, -1)

block:
  def e = rkt.handle_evt(rkt.always_evt, fun (v): values(-1, 1))
  check (select | e ~> fun(&_): #'ok) ~is #'ok
  check (select | e ~> fun(v): #'oops) ~throws "arity mismatch"
  check (select | e ~> fun(a, b): a + b) ~is 0