Async networking with cl-flow

client/src/connector.lisp

@@ -0,0 +1,97 @@
+(in-package :mortar-combat)
+
+
+(define-constant +supported-server-version+ 1)
+
+(declaim (special *message*))
+
+(defstruct (server-identity
+             (:constructor make-server-identity (id name)))
+  (id nil :read-only t)
+  (name nil :read-only t))
+
+
+(defun connection-stream-of (connector)
+  (usocket:socket-stream (connection-of connector)))
+
+
+(defclass connector (lockable disposable dispatcher)
+  ((enabled-p :initform t)
+   (connection :initarg :connection :reader connection-of)
+   (message-counter :initform 0)
+   (message-table :initform (make-hash-table :test 'eql))))
+
+
+(defmethod initialize-instance :after ((this connector) &key)
+  (with-slots (connection message-table enabled-p) this
+    (in-new-thread "connector-thread"
+      (loop while enabled-p
+         do (progn
+              (usocket:wait-for-input connection)
+              (let* ((message (conspack:decode-stream (connection-stream-of this)))
+                     (message-id (getf message :reply-for)))
+                (with-instance-lock-held (this)
+                  (if-let ((handler (gethash message-id message-table)))
+                    (progn
+                      (remhash message-id message-table)
+                      (funcall handler message))
+                    (log:error "Handler not found for message with id ~A" message-id)))))
+         finally (usocket:socket-close connection)))))
+
+
+(defun connect-to-server (host &optional (port 8778))
+  (make-instance 'connector
+                 :connection (usocket:socket-connect host port
+                                                     :element-type '(unsigned-byte 8)
+                                                     :timeout 30)))
+
+
+(defun disconnect-from-server (connector)
+  (with-slots (enabled-p) connector
+    (setf enabled-p nil)))
+
+
+(defun check-response (message expected-command)
+  (let ((command (getf message :command)))
+    (when (eq command :error)
+      (error "Server error of type ~A: ~A" (getf message :type) (getf message :text)))
+    (unless (eq command expected-command)
+      (error "Unexpected command received from server: wanted ~A, but ~A received"
+             expected-command command))))
+
+
+(defun send-command (connector &rest properties &key &allow-other-keys)
+  (let ((stream (connection-stream-of connector)))
+    (conspack:encode properties :stream stream)
+    (finish-output stream)))
+
+
+;; (,response (conspack:decode-stream (connection-stream-of ,connector))))
+(defmacro with-response (command-name (&rest properties) response &body body)
+    `(destructuring-bind (&key ,@properties &allow-other-keys) ,response
+       (check-response ,response ,command-name)
+       ,@body))
+
+
+(defmethod dispatch ((this connector) (task function) invariant &rest keys
+                     &key &allow-other-keys)
+  (with-slots (message-table message-counter) this
+    (with-instance-lock-held (this)
+      (let ((next-id (incf message-counter)))
+        (flet ((response-callback (message)
+                 (let ((*message* message))
+                   (funcall task))))
+          (setf (gethash next-id message-table) #'response-callback)
+          (apply #'send-command this :message-id next-id keys))))))
+
+
+(defun server-version (connector)
+  (-> (connector :command :version) ()
+    (with-response :version (version) *message*
+      version)))
+
+
+(defun identify (connector name)
+  (-> (connector :command :identify :name name) ()
+    (with-response :identified (id name) *message*
+      (make-server-identity id name))))

mortar-combat.asd

@@ -9,7 +9,7 @@
   :author "Pavel Korolev"
   :mailto "dev@borodust.org"
   :license "GPLv3"
-  :depends-on (log4cl uiop cl-muth bodge-blobs cl-bodge)
+  :depends-on (log4cl uiop cl-muth bodge-blobs cl-bodge usocket cl-conspack)
   :serial t
   :pathname "client/src/"
   :components ((:file "packages")
@@ -21,7 +21,8 @@
                (:file "dude")
                (:file "shaders/dude")
                (:file "shaders/passthru")
-               (:file "main")))
+               (:file "main")
+               (:file "connector")))
 
 
 (defsystem mortar-combat/distrib

proxy/commands.lisp

@@ -12,3 +12,12 @@
     (list :command :identified
           :name (name-of peer)
           :id (id-of peer))))
+
+
+(defmethod process-command ((command (eql :get-arena-list)) message))
+
+
+(defmethod process-command ((command (eql :create-arena)) message))
+
+
+(defmethod process-command ((command (eql :join-arena)) message))

proxy/proxy.lisp

@@ -17,10 +17,14 @@
           :text "Unknown command")))
 
 
+(defmethod process-command :around (command message)
+  (append (list :reply-for (getf message :message-id)) (call-next-method)))
+
+
 (defclass mortar-combat-proxy (enableable generic-system)
   ((proxy-socket :initform nil)
    (peer-registry :initform (make-instance 'peer-registry) :reader peer-registry-of)
-   (arenas :initform (make-hash-table :test #'equal))
+   (arenas :initform (make-hash-table :test #'equal) :reader arena-list-of)
    (routing-buffer :initform (make-array +routing-buffer-size+
                                          :element-type '(unsigned-byte 8)))
    (info-socket :initform nil)))