transport-zmq.lisp

   1 ;; Copyright 2010 Vitaly Mayatskikh <v.mayatskih@gmail.com>
   2 ;;
   3 ;; This file is a part of CL-Cluster
   4 ;;
   5 ;; CL-Cluster is free software: you can redistribute it and/or modify
   6 ;; it under the terms of the GNU General Public License as published by
   7 ;; the Free Software Foundation, either version 3 of the License, or
   8 ;; (at your option) any later version.
   9 ;;
  10 ;; CL-Cluster is distributed in the hope that it will be useful,
  11 ;; but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 ;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 ;; GNU General Public License for more details.
  14 ;;
  15 ;; You should have received a copy of the GNU General Public License
  16 ;; along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 (in-package :cl-cluster)
  19
  20 (defvar *context* (zmq:init 2))         ; FIXME: ask for actual CPUs nr
  21
  22 (defclass node-zmq (node)
  23   ((endpoint :initarg :endpoint :accessor node-zmq-endpoint)
  24    (socket  :initform nil :accessor node-zmq-socket)
  25    (type    :initform zmq:req :initarg :type :accessor node-zmq-type)
  26    (message :initform nil :accessor node-zmq-message)))
  27
  28 (defmethod print-object ((object node-zmq) stream)
  29   (format stream "#N<NODE:\"~a\" ENDPOINT:\"~a\">"
  30           (node-name object) (node-zmq-endpoint object)))
  31
  32 (defmethod node-alive-p ((object node-zmq))
  33   (not (null (node-zmq-socket object))))
  34
  35 (defmethod node-connect ((object node-zmq))
  36   (with-slots (socket endpoint type) object
  37     (setf socket (zmq:socket *context* type))
  38     (zmq:connect socket endpoint)))
  39
  40 (defmethod node-disconnect ((object node-zmq))
  41   (with-slots (socket) object
  42     (zmq:close socket)
  43     (setf socket nil)))
  44
  45 (defmethod node-send/unsafe ((object node-zmq) msg)
  46   (zmq:send (node-zmq-socket object) (make-instance 'zmq:msg :data msg)))
  47
  48 (defmethod node-recv/unsafe ((object node-zmq) &optional non-blocking)
  49   (let ((msg (make-instance 'zmq:msg)))
  50     (zmq:recv (node-zmq-socket object) msg)
  51     (read-from-string (zmq:msg-data-as-string msg))))
  52
  53 (defmethod node-flush/unsafe ((object node-zmq) &optional wait-input)
  54   t)
  55 ;  (zmq:flush (node-zmq-socket object)))