1 ! Copyright (C) 2005 Chris Double. All Rights Reserved.
2 ! Copyright (C) 2018 Alexander Ilin.
3 ! See http://factorcode.org/license.txt for BSD license.
4 USING: accessors arrays assocs concurrency.messaging
5 continuations destructors fry init io io.encodings.binary
6 io.servers io.sockets io.streams.duplex kernel namespaces
7 sequences serialize threads ;
8 FROM: concurrency.messaging => send ;
9 IN: concurrency.distributed
13 : registered-remote-threads ( -- hash )
14 \ registered-remote-threads get-global ;
16 : thread-connections ( -- hash )
17 \ thread-connections get-global ;
21 : register-remote-thread ( thread name -- )
22 registered-remote-threads set-at ;
24 : unregister-remote-thread ( name -- )
25 registered-remote-threads delete-at ;
27 : get-remote-thread ( name -- thread )
28 dup registered-remote-threads at [ ] [ threads at ] ?if ;
32 : handle-node-client ( -- )
34 first2 get-remote-thread send handle-node-client
35 ] [ stop-this-server ] if* ;
37 : <node-server> ( addrspec -- threaded-server )
38 binary <threaded-server>
40 "concurrency.distributed" >>name
41 [ handle-node-client ] >>handler ;
43 : start-node ( addrspec -- )
44 <node-server> start-server local-node set-global ;
46 TUPLE: remote-thread node id ;
48 C: <remote-thread> remote-thread
50 TUPLE: connection remote stream local ;
52 C: <connection> connection
54 : connect ( remote-thread -- )
55 [ node>> dup binary <client> <connection> ]
56 [ thread-connections set-at ] bi ;
58 : disconnect ( remote-thread -- )
59 thread-connections delete-at*
60 [ stream>> dispose ] [ drop ] if ;
62 : with-connection ( remote-thread quot -- )
63 '[ connect @ ] over [ disconnect ] curry finally ; inline
65 : send-remote-message ( message node -- )
66 binary [ serialize ] with-client ;
68 : send-to-connection ( message connection -- )
69 stream>> [ serialize flush ] with-stream* ;
72 [ id>> 2array ] [ node>> ] [ thread-connections at ] tri
73 [ nip send-to-connection ] [ send-remote-message ] if* ;
76 id>> [ local-node get insecure>> ] dip <remote-thread> (serialize) ;
79 f local-node get insecure>> send-remote-message ;
82 H{ } clone \ registered-remote-threads set-global
83 H{ } clone \ thread-connections set-global
84 ] "remote-thread-registry" add-startup-hook