ten/ten.tcl

   1 package require Tcl 8.5
   2
   3 package require snit
   4 package require json::write
   5 package require json
   6
   7 namespace eval ::ten:: {
   8   set library [file dirname [info script]]
   9
  10   proc nofuture {args} {}
  11 }
  12
  13 snit::type ten::connector::perl {
  14
  15   method connect {} {
  16     set conn_fh [open {|object-remote-node} r+]
  17     set firstline [gets $conn_fh]
  18     switch $firstline {
  19       Shere {}
  20       default { error "Expected Shere, got $firstline" }
  21     }
  22
  23     set channel [ten::read_channel %AUTO% -fh $conn_fh]
  24
  25     set conn [
  26       ten::connection %AUTO% -send_to_fh $conn_fh -read_channel $channel
  27     ]
  28
  29     return $conn
  30   }
  31 }
  32
  33 snit::type ten::connection {
  34
  35   option -send_to_fh
  36   option -read_channel
  37
  38   constructor {args} {
  39     $self configurelist $args
  40     fconfigure $options(-send_to_fh) -buffering none
  41     $options(-read_channel) configurelist [list \
  42       -on_line_call [ mymethod Receive ] \
  43       -on_close_call [ mymethod ChannelClosed ] \
  44     ]
  45   }
  46
  47   method Receive {line} {
  48     set tcl_line [ json::json2dict $line ]
  49     set type [ lindex $tcl_line 0 ]
  50     set rest [ lrange $tcl_line 1 end ]
  51     $self "receive_$type" {*}$rest
  52   }
  53
  54   method receive_call_free {future_id id args} {
  55     $self receive_call $future_id $id "" {*}$args
  56     $self receive_free $id
  57   }
  58
  59   method receive_call {future_id id args} {
  60     if {$future_id == "NULL"} {
  61       set future ten::nofuture
  62     } else {
  63       set future $future_id
  64     }
  65     $self Invoke $future $id {*}$args
  66   }
  67
  68   method receive_free {id} {
  69     $id free
  70   }
  71
  72   method Invoke {future local ctx method args} {
  73     set result [$local $method {*}$args]
  74     $future done $result
  75   }
  76
  77   method ChannelClosed {} {
  78     error "Fail"
  79   }
  80
  81   method send {message_type args} {
  82     set future [ten::future %AUTO%]
  83     set call_args [list \
  84       [ json::write string $message_type ] \
  85       [ json::write string $future ] \
  86       {*}$args
  87     ]
  88     $self Send $call_args
  89     return $future
  90   }
  91
  92   method Send {to_send} {
  93     set send_this [ json::write array {*}$to_send ]
  94     puts $options(-send_to_fh) $send_this
  95   }
  96
  97   method remote_object {id} {
  98     return [ten::handle %AUTO% -connection $self -id $id]
  99   }
 100
 101 }
 102
 103 snit::type ten::future {
 104
 105   variable callbacks ""
 106   variable is_ready 0
 107   variable result ""
 108   variable failure ""
 109   variable retain_count 1
 110
 111   method done {args} {
 112     if [$self is_ready] {
 113       error "Future $self already completed"
 114     }
 115     set result $args
 116     $self MarkReady
 117   }
 118
 119   method fail {args} {
 120     if [$self is_ready] {
 121       error "Future $self already completed"
 122     }
 123     set failure $args
 124     $self MarkReady
 125   }
 126
 127   method on_ready {cb_code} {
 128     if [$self is_ready] {
 129       eval [concat $cb_code $self]
 130     } else {
 131       lappend callbacks [list ready $cb_code]
 132     }
 133   }
 134
 135   method on_done {cb_code} {
 136     if [llength $result] {
 137       eval [concat $cb_code [list $result]]
 138     } else {
 139       lappend callbacks [list done $cb_code]
 140     }
 141   }
 142
 143   method on_fail {cb_code} {
 144     if [llength $failure] {
 145       eval [concat $cb_code [list $failure]]
 146     } else {
 147       lappend callbacks [list fail $cb_code]
 148     }
 149   }
 150
 151   method MarkReady {} {
 152     set is_ready 1
 153     foreach cb $callbacks {
 154       set cb_type [lindex $cb 0]
 155       set cb_code [lindex $cb 1]
 156       switch $cb_type {
 157         ready { eval [concat $cb_code $self] }
 158         done {
 159           if ![$self is_failure] {
 160             eval [concat $cb_code [list $result]]
 161           }
 162         }
 163         failed {
 164           if [$self is_failure] {
 165             eval [concat $cb_code [list $failure]]
 166           }
 167         }
 168       }
 169     }
 170   }
 171
 172   method AssertReady {} {
 173     if {!$is_ready} {
 174       error "Future not ready"
 175     }
 176   }
 177
 178   method is_ready {} { return $is_ready }
 179
 180   method get {} {
 181     $self AssertReady
 182     if [$self is_failure] {
 183       error $failure
 184     }
 185     return $result
 186   }
 187
 188   method is_failure {} {
 189     $self AssertReady
 190     if [llength $failure] {
 191       return 1
 192     }
 193     return 0
 194   }
 195
 196   method failure {} {
 197     $self AssertReady
 198     return $failure
 199   }
 200
 201   method await_ready {} {
 202     vwait "${selfns}::is_ready"
 203   }
 204
 205   method await_get {} {
 206     $self retain
 207     $self await_ready
 208     if [$self is_failure] {
 209       set err [$self failure]
 210       $self free
 211       error $err
 212     }
 213     set res [$self get]
 214     $self free
 215     return $res
 216   }
 217
 218   method retain {} {
 219     incr retain_count
 220   }
 221
 222   method free {} {
 223     set retain_count [ expr $retain_count - 1 ]
 224     if {$retain_count == 0} {
 225       $self destroy
 226     }
 227   }
 228 }
 229
 230 snit::type ten::read_channel {
 231   option -fh
 232   option -on_close_call
 233   option -on_line_call
 234
 235   constructor {args} {
 236     $self configurelist $args
 237     fconfigure $options(-fh) -blocking 0
 238     fileevent $options(-fh) readable [mymethod ReceiveData]
 239   }
 240
 241   method ReceiveData {} {
 242     set chan $options(-fh)
 243     if [eof $chan] {
 244       if [llength $options(-on_close_call)] {
 245         eval $options(-on_close_call)
 246       }
 247     } else {
 248       if [llength $options(-on_line_call)] {
 249         while {[string length [set line [gets $chan]]] > 0} {
 250           eval [concat $options(-on_line_call) [ list $line ]]
 251         }
 252       }
 253     }
 254   }
 255 }
 256
 257 snit::type ten::handle {
 258   option -connection
 259   option -id
 260
 261   method call {name args} {
 262     return [[$self start $name {*}$args] await_get]
 263   }
 264
 265   method start {name args} {
 266     $options(-connection) send call [json::write string $options(-id)] 0 [json::write string $name] {*}$args
 267   }
 268 }
 269
 270 package provide ten 0.0.1