steve
/
factor
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

zeromq: bindings for ZeroMQ.

db4
John Benediktsson 2013-09-18 15:28:59 -07:00
parent ee785be30c
commit 0b5458da29
14 changed files with 604 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
extra/zeromq/authors.txt Normal file
View File

@ -0,0 +1,2 @@
Eungju PARK
John Benediktsson

25
extra/zeromq/examples/echo-client.factor Normal file
View File

@ -0,0 +1,25 @@
! Copyright (C) 2013 John Benediktsson.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar calendar.format destructors io
kernel present strings threads zeromq zeromq.ffi ;
IN: zeromq.examples.echo-client
: echo-client ( -- )
[
<zmq-context> &dispose
ZMQ_REQ <zmq-socket> &dispose
dup "tcp://127.0.0.1:5000" zmq-connect
[
now present
[ "Sending " write print flush ]
[ >byte-array dupd 0 zmq-send ] bi
dup 0 zmq-recv >string
"Received " write print flush
1 seconds sleep
t
] loop drop
] with-destructors ;
MAIN: echo-client

21
extra/zeromq/examples/echo-server.factor Normal file
View File

@ -0,0 +1,21 @@
! Copyright (C) 2013 John Benediktsson.
! See http://factorcode.org/license.txt for BSD license.
USING: destructors io kernel strings zeromq zeromq.ffi ;
IN: zeromq.examples.echoserver
: echo-server ( -- )
[
<zmq-context> &dispose
ZMQ_REP <zmq-socket> &dispose
dup "tcp://127.0.0.1:5000" zmq-bind
[
dup 0 zmq-recv
[ >string "Received " write print flush ]
[ dupd 0 zmq-send ] bi
t
] loop drop
] with-destructors ;
MAIN: echo-server

23
extra/zeromq/examples/hwclient.factor Normal file
View File

@ -0,0 +1,23 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays destructors formatting io kernel sequences
strings zeromq zeromq.ffi ;
IN: zeromq.examples.hwclient
: hwclient ( -- )
[
<zmq-context> &dispose
"Connecting to hello world server…" print
ZMQ_REQ <zmq-socket> &dispose
dup "tcp://localhost:5555" zmq-connect
10 iota [
[ "Hello" dup rot "Sending %s %d...\n" printf
dupd >byte-array 0 zmq-send ]
[ [ dup 0 zmq-recv >string ] dip
"Received %s %d\n" printf flush ]
bi
] each drop
] with-destructors ;
MAIN: hwclient

21
extra/zeromq/examples/hwserver.factor Normal file
View File

@ -0,0 +1,21 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar destructors io kernel strings
threads zeromq zeromq.ffi ;
IN: zeromq.examples.hwserver
: hwserver ( -- )
[
<zmq-context> &dispose
ZMQ_REP <zmq-socket> &dispose
dup "tcp://*:5555" zmq-bind
[ t ] [
dup
[ 0 zmq-recv >string "Received " write print flush ]
[ drop 1 seconds sleep ]
[ "World" >byte-array 0 zmq-send ]
tri
] while drop
] with-destructors ;
MAIN: hwserver

26
extra/zeromq/examples/tasksink.factor Normal file
View File

@ -0,0 +1,26 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar destructors formatting io kernel
math strings sequences zeromq zeromq.ffi ;
IN: zeromq.examples.tasksink
: tasksink ( -- )
[
<zmq-context> &dispose
ZMQ_PULL <zmq-socket> &dispose
dup "tcp://*:5558" zmq-bind
! Wait for start of batch
dup 0 zmq-recv drop
! Start our clock now
now
! Process 100 confirmations
100 iota [
pick 0 zmq-recv drop
10 rem zero? [ ":" ] [ "." ] if write flush
] each
! Calculate and report duration of batch
now swap time- duration>milliseconds "Total elapsed time: %d msec\n" printf
drop
] with-destructors ;
MAIN: tasksink

44
extra/zeromq/examples/taskvent.factor Normal file
View File

@ -0,0 +1,44 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar destructors formatting io kernel
math namespaces random threads zeromq zeromq.ffi ;
IN: zeromq.examples.taskvent
: taskvent ( -- )
[
<zmq-context> &dispose
[
! Socket to send messages on
ZMQ_PUSH <zmq-socket> &dispose
dup "tcp://*:5557" zmq-bind
] [
! Socket to send start of batch message on
ZMQ_PUSH <zmq-socket> &dispose
dup "tcp://localhost:5558" zmq-connect
] bi
"Press Enter when the workers are ready: " write flush
read1 drop
"Sending tasks to workers…\n" write flush
! The first message is "0" and signals start of batch
dup "0" >byte-array 0 zmq-send
! Send 100 tasks
0 100 [
! Random workload from 1 to 100msecs
100 random 1 +
dup [ + ] dip
[ pick ] dip "%d" sprintf >byte-array 0 zmq-send
] times
"Total expected cost: %d msec\n" printf
! Give 0MQ time to deliver
1 seconds sleep
drop
drop
] with-destructors ;
MAIN: taskvent

37
extra/zeromq/examples/taskwork.factor Normal file
View File

@ -0,0 +1,37 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar destructors formatting io kernel
math.parser strings threads zeromq zeromq.ffi ;
IN: zeromq.examples.taskwork
: taskwork ( -- )
[
<zmq-context> &dispose
[
! Socket to receive messages on
ZMQ_PULL <zmq-socket> &dispose
dup "tcp://localhost:5557" zmq-connect
] [
! Socket to send messages to
ZMQ_PUSH <zmq-socket> &dispose
dup "tcp://localhost:5558" zmq-connect
] bi
! Process tasks forever
[
over 0 zmq-recv >string
! Simple progress indicator for the viewer
dup "%s." printf flush
! Do the work
string>number milliseconds sleep
! Send results to sink
dup "" >byte-array 0 zmq-send
t
] loop
drop
drop
] with-destructors ;
MAIN: taskwork

26
extra/zeromq/examples/wuclient.factor Normal file
View File

@ -0,0 +1,26 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays command-line destructors formatting io kernel
math math.parser namespaces sequences splitting strings zeromq
zeromq.ffi ;
IN: zeromq.examples.wuclient
: wuclient ( -- )
[
<zmq-context> &dispose
"Collecting updates from weather server…" print
ZMQ_SUB <zmq-socket> &dispose
dup "tcp://localhost:5556" zmq-connect
command-line get [ "10001 " ] [ first ] if-empty
2dup >byte-array ZMQ_SUBSCRIBE swap zmq-setopt
0 100 dup [
[ pick 0 zmq-recv
>string " " split [ string>number ] map second +
] times
] dip
/ "Average temperature for zipcode '%s' was %dF\n" printf
drop
] with-destructors ;
MAIN: wuclient

25
extra/zeromq/examples/wuserver.factor Normal file
View File

@ -0,0 +1,25 @@
! Copyright (C) 2012 Eungju PARK.
! See http://factorcode.org/license.txt for BSD license.
USING: byte-arrays calendar destructors formatting kernel math
namespaces random zeromq zeromq.ffi ;
IN: zeromq.examples.wuserver
: wuserver ( -- )
[
<zmq-context> &dispose
ZMQ_PUB <zmq-socket> &dispose
dup "tcp://*:5556" zmq-bind
dup "ipc://weather.ipc" zmq-bind
random-generator get now timestamp>unix-time >fixnum seed-random [
[ t ] [
dup
100000 random
215 random 80 -
50 random 10 +
"%05d %d %d" sprintf
>byte-array 0 zmq-send
] while
] with-random drop
] with-destructors ;
MAIN: wuserver

236
extra/zeromq/ffi/ffi.factor Normal file
View File

@ -0,0 +1,236 @@
! Copyright (C) 2011-2013 Eungju PARK, John Benediktsson.
! See http://factorcode.org/license.txt for BSD license.
USING: alien alien.accessors alien.c-types alien.data
alien.libraries alien.syntax byte-arrays classes.struct
combinators kernel literals math system ;
IN: zeromq.ffi
<< "zmq" {
{ [ os windows? ] [ "libzmq.dll" cdecl add-library ] }
{ [ os macosx? ] [ "libzmq.dylib" cdecl add-library ] }
{ [ os unix? ] [ "libzmq.so" cdecl add-library ] }
} cond >>
LIBRARY: zmq
!
! 0MQ versioning support.
!
! Run-time API version detection
FUNCTION: void zmq_version ( int* major, int* minor, int* patch ) ;
!
! 0MQ errors.
!
! A number random enough not to collide with different errno ranges on
! different OSes. The assumption is that error_t is at least 32-bit type.
<< CONSTANT: ZMQ_HAUSNUMERO 156384712 >>
! Native 0MQ error codes.
CONSTANT: EFSM $[ ZMQ_HAUSNUMERO 51 + ]
CONSTANT: ENOCOMPATPROTO $[ ZMQ_HAUSNUMERO 52 + ]
CONSTANT: ETERM $[ ZMQ_HAUSNUMERO 53 + ]
CONSTANT: EMTHREAD $[ ZMQ_HAUSNUMERO 54 + ]
! This function retrieves the errno as it is known to 0MQ library. The goal
! of this function is to make the code 100% portable, including where 0MQ
! compiled with certain CRT library (on Windows) is linked to an
! application that uses different CRT library.
FUNCTION: int zmq_errno ( ) ;
! Resolves system errors and 0MQ errors to human-readable string.
FUNCTION: c-string zmq_strerror ( int errnum ) ;
!
! 0MQ infrastructure (a.k.a. context) initialisation & termination.
!
! New API
! Context options
CONSTANT: ZMQ_IO_THREADS 1
CONSTANT: ZMQ_MAX_SOCKETS 2
! Default for new contexts
CONSTANT: ZMQ_IO_THREADS_DFLT 1
CONSTANT: ZMQ_MAX_SOCKETS_DFLT 1024
FUNCTION: void* zmq_ctx_new ( ) ;
FUNCTION: int zmq_ctx_destroy ( void* context ) ;
FUNCTION: int zmq_ctx_set ( void* context, int option, int optval ) ;
FUNCTION: int zmq_ctx_get ( void* context, int option ) ;
! Old (legacy) API
FUNCTION: void* zmq_init ( int io_threads ) ;
FUNCTION: int zmq_term ( void* context ) ;
!
! 0MQ message definition.
!
STRUCT: zmq_msg_t
{ data uchar[32] } ;
FUNCTION: int zmq_msg_init ( zmq_msg_t* msg ) ;
FUNCTION: int zmq_msg_init_size ( zmq_msg_t* msg, size_t size ) ;
FUNCTION: int zmq_msg_init_data ( zmq_msg_t* msg, void* data, size_t size, void* ffn, void* hint ) ;
FUNCTION: int zmq_msg_send ( zmq_msg_t* msg, void* s, int flags ) ;
FUNCTION: int zmq_msg_recv ( zmq_msg_t* msg, void* s, int flags ) ;
FUNCTION: int zmq_msg_close ( zmq_msg_t* msg ) ;
FUNCTION: int zmq_msg_move ( zmq_msg_t* dest, zmq_msg_t* src ) ;
FUNCTION: int zmq_msg_copy ( zmq_msg_t* dest, zmq_msg_t* src ) ;
FUNCTION: void* zmq_msg_data ( zmq_msg_t* msg ) ;
FUNCTION: size_t zmq_msg_size ( zmq_msg_t* msg ) ;
FUNCTION: int zmq_msg_more ( zmq_msg_t* msg ) ;
FUNCTION: int zmq_msg_get ( zmq_msg_t* msg, int option ) ;
FUNCTION: int zmq_msg_set ( zmq_msg_t* msg, int option, int optval ) ;
!
! 0MQ socket definition.
!
! Socket types.
CONSTANT: ZMQ_PAIR 0
CONSTANT: ZMQ_PUB 1
CONSTANT: ZMQ_SUB 2
CONSTANT: ZMQ_REQ 3
CONSTANT: ZMQ_REP 4
CONSTANT: ZMQ_DEALER 5
CONSTANT: ZMQ_ROUTER 6
CONSTANT: ZMQ_PULL 7
CONSTANT: ZMQ_PUSH 8
CONSTANT: ZMQ_XPUB 9
CONSTANT: ZMQ_XSUB 10
! Deprecated aliases
ALIAS: ZMQ_XREQ ZMQ_DEALER
ALIAS: ZMQ_XREP ZMQ_ROUTER
! Socket options.
CONSTANT: ZMQ_AFFINITY 4
CONSTANT: ZMQ_IDENTITY 5
CONSTANT: ZMQ_SUBSCRIBE 6
CONSTANT: ZMQ_UNSUBSCRIBE 7
CONSTANT: ZMQ_RATE 8
CONSTANT: ZMQ_RECOVERY_IVL 9
CONSTANT: ZMQ_SNDBUF 11
CONSTANT: ZMQ_RCVBUF 12
CONSTANT: ZMQ_RCVMORE 13
CONSTANT: ZMQ_FD 14
CONSTANT: ZMQ_EVENTS 15
CONSTANT: ZMQ_TYPE 16
CONSTANT: ZMQ_LINGER 17
CONSTANT: ZMQ_RECONNECT_IVL 18
CONSTANT: ZMQ_BACKLOG 19
CONSTANT: ZMQ_RECONNECT_IVL_MAX 21
CONSTANT: ZMQ_MAXMSGSIZE 22
CONSTANT: ZMQ_SNDHWM 23
CONSTANT: ZMQ_RCVHWM 24
CONSTANT: ZMQ_MULTICAST_HOPS 25
CONSTANT: ZMQ_RCVTIMEO 27
CONSTANT: ZMQ_SNDTIMEO 28
CONSTANT: ZMQ_IPV4ONLY 31
CONSTANT: ZMQ_LAST_ENDPOINT 32
CONSTANT: ZMQ_ROUTER_MANDATORY 33
CONSTANT: ZMQ_TCP_KEEPALIVE 34
CONSTANT: ZMQ_TCP_KEEPALIVE_CNT 35
CONSTANT: ZMQ_TCP_KEEPALIVE_IDLE 36
CONSTANT: ZMQ_TCP_KEEPALIVE_INTVL 37
CONSTANT: ZMQ_TCP_ACCEPT_FILTER 38
CONSTANT: ZMQ_DELAY_ATTACH_ON_CONNECT 39
CONSTANT: ZMQ_XPUB_VERBOSE 40
! Message options
CONSTANT: ZMQ_MORE 1
! Send/recv options.
CONSTANT: ZMQ_DONTWAIT 1
CONSTANT: ZMQ_SNDMORE 2
! Deprecated aliases
ALIAS: ZMQ_NOBLOCK ZMQ_DONTWAIT
ALIAS: ZMQ_FAIL_UNROUTABLE ZMQ_ROUTER_MANDATORY
ALIAS: ZMQ_ROUTER_BEHAVIOR ZMQ_ROUTER_MANDATORY
!
! 0MQ socket events and monitoring
!
! Socket transport events (tcp and ipc only)
CONSTANT: ZMQ_EVENT_CONNECTED 1
CONSTANT: ZMQ_EVENT_CONNECT_DELAYED 2
CONSTANT: ZMQ_EVENT_CONNECT_RETRIED 4
CONSTANT: ZMQ_EVENT_LISTENING 8
CONSTANT: ZMQ_EVENT_BIND_FAILED 16
CONSTANT: ZMQ_EVENT_ACCEPTED 32
CONSTANT: ZMQ_EVENT_ACCEPT_FAILED 64
CONSTANT: ZMQ_EVENT_CLOSED 128
CONSTANT: ZMQ_EVENT_CLOSE_FAILED 256
CONSTANT: ZMQ_EVENT_DISCONNECTED 512
CONSTANT: ZMQ_EVENT_ALL flags{
ZMQ_EVENT_CONNECTED ZMQ_EVENT_CONNECT_DELAYED
ZMQ_EVENT_CONNECT_RETRIED ZMQ_EVENT_LISTENING
ZMQ_EVENT_BIND_FAILED ZMQ_EVENT_ACCEPTED
ZMQ_EVENT_ACCEPT_FAILED ZMQ_EVENT_CLOSED
ZMQ_EVENT_CLOSE_FAILED ZMQ_EVENT_DISCONNECTED
}
! Socket event data (union member per event)
STRUCT: zmq_event_t
{ event int }
{ addr c-string }
{ fd-or-err int } ;
FUNCTION: void* zmq_socket ( void* ctx, int type ) ;
FUNCTION: int zmq_close ( void* s ) ;
FUNCTION: int zmq_setsockopt ( void* s, int option, void* optval, size_t optvallen ) ;
FUNCTION: int zmq_getsockopt ( void* s, int option, void* optval, size_t* optvallen ) ;
FUNCTION: int zmq_bind ( void* s, c-string addr ) ;
FUNCTION: int zmq_connect ( void* s, c-string addr ) ;
FUNCTION: int zmq_unbind ( void* s, c-string addr ) ;
FUNCTION: int zmq_disconnect ( void* s, c-string addr ) ;
FUNCTION: int zmq_send ( void* s, void* buf, size_t len, int flags ) ;
FUNCTION: int zmq_recv ( void* s, void* buf, size_t len, int flags ) ;
FUNCTION: int zmq_socket_monitor ( void* s, c-string addr, int events ) ;
FUNCTION: int zmq_sendmsg ( void* s, zmq_msg_t* msg, int flags ) ;
FUNCTION: int zmq_recvmsg ( void* s, zmq_msg_t* msg, int flags ) ;
! Experimental
FUNCTION: int zmq_sendiov ( void* s, void* iov, size_t count, int flags ) ;
FUNCTION: int zmq_recviov ( void* s, void* iov, size_t* count, int flags ) ;
!
! I/O multiplexing.
!
CONSTANT: ZMQ_POLLIN 1
CONSTANT: ZMQ_POLLOUT 2
CONSTANT: ZMQ_POLLERR 4
! FIXME: { fd SOCKET } on Windows
STRUCT: zmq_pollitem_t
{ socket void* }
{ fd int }
{ events short }
{ revents short } ;
FUNCTION: int zmq_poll ( zmq_pollitem_t* items, int nitems, long timeout ) ;
! Built-in message proxy (3-way)
FUNCTION: int zmq_proxy ( void* frontend, void* backend, void* capture ) ;
! Deprecated aliases
CONSTANT: ZMQ_STREAMER 1
CONSTANT: ZMQ_FORWARDER 2
CONSTANT: ZMQ_QUEUE 3
! Deprecated method
FUNCTION: int zmq_device ( int type, void* frontend, void* backend ) ;

1
extra/zeromq/summary.txt Normal file
View File

@ -0,0 +1 @@
Bindings to 0MQ (zeromq).

11
extra/zeromq/zeromq-tests.factor Normal file
View File

@ -0,0 +1,11 @@
! Copyright (C) 2011-2013 Eungju PARK, John Benediktsson.
! See http://factorcode.org/license.txt for BSD license.
USING: destructors kernel tools.test ;
IN: zeromq
{ t } [
B{ 0 1 10 33 244 255 } dup byte-array>zmq-message
[ zmq-message>byte-array ] with-disposal =
] unit-test

106
extra/zeromq/zeromq.factor Normal file
View File

@ -0,0 +1,106 @@
! Copyright (C) 2011-2013 Eungju PARK, John Benediktsson.
! See http://factorcode.org/license.txt for BSD license.
USING: accessors alien.c-types alien.data arrays byte-arrays
classes.struct combinators continuations destructors fry io
kernel libc math namespaces sequences zeromq.ffi ;
IN: zeromq
ERROR: zmq-error n string ;
: throw-zmq-error ( -- )
zmq_errno dup zmq_strerror zmq-error ; inline
: check-zmq-error ( retval -- )
[ throw-zmq-error ] unless-zero ; inline
: zmq-version ( -- version )
{ int int int } [ zmq_version ] with-out-parameters 3array ;
GENERIC# zmq-setopt 2 ( obj name value -- )
GENERIC# zmq-getopt 1 ( obj name -- value )
TUPLE: zmq-message underlying ;
: <zmq-message> ( -- msg )
zmq_msg_t <struct>
[ zmq_msg_init check-zmq-error ]
[ zmq-message boa ] bi ;
M: zmq-message dispose
underlying>> zmq_msg_close check-zmq-error ;
: byte-array>zmq-message ( byte-array -- msg )
zmq_msg_t <struct>
[ over length zmq_msg_init_size check-zmq-error ]
[ zmq_msg_data swap dup length memcpy ]
[ zmq-message boa ] tri ;
: zmq-message>byte-array ( msg -- byte-array )
underlying>> [ zmq_msg_data ] [ zmq_msg_size ] bi
[ drop B{ } ] [ memory>byte-array ] if-zero ;
TUPLE: zmq-context underlying ;
! this uses the "New API" with version 3
! previous versions should use zmq_init and zmq_term
: <zmq-context> ( -- context )
zmq_ctx_new zmq-context boa ;
M: zmq-context dispose
underlying>> zmq_ctx_destroy check-zmq-error ;
M: zmq-context zmq-setopt
[ underlying>> ] 2dip zmq_ctx_set check-zmq-error ;
M: zmq-context zmq-getopt
[ underlying>> ] dip zmq_ctx_get ;
TUPLE: zmq-socket underlying ;
: <zmq-socket> ( context type -- socket )
[ underlying>> ] dip zmq_socket
dup [ throw-zmq-error ] unless
zmq-socket boa ;
M: zmq-socket dispose
underlying>> zmq_close check-zmq-error ;
M: zmq-socket zmq-setopt
[ underlying>> ] 2dip over {
{ ZMQ_SUBSCRIBE [ dup length ] }
{ ZMQ_UNSUBSCRIBE [ dup length ] }
{ ZMQ_RCVTIMEO [ 4 ] }
{ ZMQ_SNDTIMEO [ 4 ] }
} case zmq_setsockopt check-zmq-error ;
: zmq-bind ( socket addr -- )
[ underlying>> ] dip zmq_bind check-zmq-error ;
: zmq-unbind ( socket addr -- )
[ underlying>> ] dip zmq_unbind check-zmq-error ;
: zmq-connect ( socket addr -- )
[ underlying>> ] dip zmq_connect check-zmq-error ;
: zmq-disconnect ( socket addr -- )
[ underlying>> ] dip zmq_disconnect check-zmq-error ;
: zmq-sendmsg ( socket msg flags -- )
[ [ underlying>> ] bi@ ] dip zmq_sendmsg
0 < [ throw-zmq-error ] when ;
: zmq-recvmsg ( socket msg flags -- )
[ [ underlying>> ] bi@ ] dip zmq_recvmsg
0 < [ throw-zmq-error ] when ;
: zmq-send ( socket byte-array flags -- )
[ byte-array>zmq-message ] dip
'[ _ zmq-sendmsg ] with-disposal ;
: zmq-recv ( socket flags -- byte-array )
<zmq-message> [
[ swap zmq-recvmsg ] [ zmq-message>byte-array ] bi
] with-disposal ;