source: branches/home/psmith/restructure/src/protocol/yarpc/yarpc-state-machine.lisp @ 74

#|
Copyright (c) 2007
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
   notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
   notice, this list of conditions and the following disclaimer in the
   documentation and/or other materials provided with the distribution.
3. The name of the author may not be used to endorse or promote products
   derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|#
(in-package :nio-yarpc)

(declaim (optimize (debug 3) (speed 3) (space 0)))

;; YetAnotherRPC state machine
;;
;; A server that processes remote procedure calls and returns results
;;
(defclass yarpc-state-machine (state-machine)
  (
   (result-queue :initform (nio-compat:concurrent-queue)
                 :accessor result-queue
                 :documentation "The queue used to return results from an external thread to the nio thread")))

(defparameter job-queue (nio-compat:concurrent-queue)
  "The queue used to hand off work from the NIO thread to an external thread for execution")

(defparameter yarpc-pf (yarpc-packet-factory))

(defmethod get-packet-factory((sm yarpc-state-machine))
  yarpc-pf)

(defmethod print-object ((sm yarpc-state-machine) stream)
  (format stream "#<YARPC-STATE-MACHINE ~A >" (call-next-method sm nil)))

(defconstant STATE-INITIALISED 0)
(defconstant STATE-SEND-RESPONSE 1)


(defparameter +process-jobs-inline+ t
  "Set this to make the NIO thread process the RPC calls - warning the procedure should not block!")



(defun run-job(&key (wait-on-job-pdw t))
  (format-log t "yarpc-state-machine:run-job - Server toplevel waiting for job~%")
  (destructuring-bind (job result-queue) (nio-compat:take nio-yarpc:job-queue :blocking-call wait-on-job-pdw)
    (format-log t "yarpc-state-machine:run-job - Server received job ~A~%" job)
    (nio-compat:add result-queue (nio-yarpc:execute-call job))))


(defmethod process-outgoing-packet((sm yarpc-state-machine))
  (format-log t "yarpc-state-machine:process-outgoing-packet - called, polling the results-queue ~%" )
  (let ((result (nio-compat:take (result-queue sm) :blocking-call nil)))
    (format-log t "yarpc-state-machine:process-outgoing-packet - got result ~A ~%" result)
     (when result
        (method-response-packet result))))

;Process a call method packet by placing it in the job-queue
(defmethod process-incoming-packet ((sm yarpc-state-machine) (call call-method-packet))
  (assert (eql (state sm) STATE-INITIALISED))
  (format-log t "yarpc-state-machine:process-incoming-packet - called :sm ~A :packet ~A~%" sm call)
  (nio-compat:add job-queue (list (call-string call) (result-queue sm)))
  (when +process-jobs-inline+ (run-job :wait-on-job-pdw nil)))



;;TODO move somewhere suitable

(defparameter *remote-fns* nil)

(defun register-remote-fn(name)
    (push name *remote-fns*))

(defmacro defremote (name args &rest body)
  `(progn 
     (defun ,name (,@args) ,@body)
     (register-remote-fn #',name)))

(defremote test-rpc-list()
  (list 3 "as" 's (code-char #x2211)))

(defremote test-rpc-string(a b c)
  (format nil "response - ~A ~A ~A ~A~%" a b c (code-char #x2211)))

(define-condition authorization-error (error) ())

(defun execute-call (call-string)
  (handler-case
      (let* ((rpc-call-list (read-from-string call-string ))
             (fn (member (symbol-function (first rpc-call-list)) *remote-fns* )))
        (format-log t "yarpc-state-machine:execute-call - fn ~A authorised? : ~A~%" (symbol-function (first rpc-call-list)) fn)
        (if fn
            (apply (first rpc-call-list) (rest rpc-call-list))
            (error 'authorization-error)))
    (reader-error (re) (format-log t "yarpc-state-machine:execute-call - reader error on call-string ~A ~%" re))))

;;end move TODO





;  (handler-case
;    (let ((result (execute-call (call-string call))))
;      (when result 
;       (let ((response-packet (progn 
;                                  (setf state STATE-SEND-RESPONSE)
;                                  (queue-outgoing-packet sm (method-response-packet result)))))
;          t)))
;    (reader-error (re) (format t "No such function ~A~%" (call-string call)))
;    (authorization-error (ae) (format t "Function not declared with defremote ~A~%" (call-string call))))