Commit d3022e4e authored by  Hayley Patton's avatar Hayley Patton 🐢

websockets and more ANSI CL compliance

parent dfb321fa
Pipeline #755 passed with stage
in 43 seconds
......@@ -2,7 +2,6 @@ image: daewok/lisp-devel
stages:
- test
- build
before_script:
- ln -s "${CI_PROJECT_DIR}" ~/quicklisp/local-projects/cl-decentralise
......
(defsystem decentralise-example
:depends-on (:decentralise)
:depends-on (:decentralise :alexandria)
:components ((:module example
:components (:file "example"))))
:components ((:file "example")))))
(defsystem decentralise-websocket
:depends-on (:hunchentoot :hunchensocket :bordeaux-threads
:jsown :decentralise :websocket-driver
:trivial-timeout)
:components ((:module websocket
:components ((:file "package")
(:file "host")
(:file "client")
(:file "network")))))
......@@ -14,15 +14,14 @@
:accessor meta-table)))
(defmethod listing-generator ((system my-system))
(let ((iterating t))
(with-hash-table-iterator (iterator (meta-table system))
(lambda ()
(when iterating
(multiple-value-bind (success name metadata)
(iterator)
(if success
(values name (car metadata))
(setf iterating nil))))))))
(let ((alist (alexandria:hash-table-alist (meta-table system))))
(lambda ()
(if (null alist)
nil
(destructuring-bind (name version &rest channels)
(pop alist)
(declare (ignore channels))
(values name version))))))
(defmethod put-item ((system my-system) name text version channels)
(setf (gethash name (data-table system)) text
......
......@@ -28,9 +28,6 @@
names)
(finish-output (connection-conn connection)))
(defun get-block (connection name)
(get-blocks connection (list name)))
(defmethod write-error ((connection connection) name &optional (reason ""))
(format (connection-conn connection)
"ERROR ~a ~a~%" name reason)
......
......@@ -3,6 +3,8 @@
(in-package :decentralise)
;;; All this is specific to the cl-decentralise protocol.
(defvar +eof-marker+ "--EOF--")
(defun split-by-spaces (line)
......@@ -54,4 +56,34 @@
(ecase (length lname) ; IPv4 or 6?
(4 (format nil "~{~d~^.~}" lname))
(6 (format nil "~{~d~^:~}" lname))))))
;;; This is somewhat internal, but could be applied to other protocols.
(defmethod bind ((system system) &key (host "0.0.0.0"))
(setf (slot-value system 'connection)
(socket-listen host
(slot-value system 'port)
:reuse-address t)))
(defmethod protocol-name ((connection connection)) "netfarmprot")
(defmethod connection-name ((connection connection))
(format nil "~a:~a:~d"
(protocol-name connection)
(connection-ip connection)
(connection-port connection)))
(setf (gethash "netfarmprot" *known-protocols*)
(lambda (ip port timeout)
(let ((socket (socket-connect ip port
:timeout timeout
:nodelay t)))
(make-instance 'connection
:ip ip
:port port
:connection (socket-stream socket)
:socket socket))))
(defmethod close-connection ((connection connection))
(close (connection-conn connection))
(when (slot-boundp connection 'raw-socket)
(socket-close (connection-socket connection))))
......@@ -3,31 +3,42 @@
(defpackage :decentralise
(:use :cl :usocket :split-sequence :bordeaux-threads :optima)
(:export :system
(:export #:system
;; event handlers
:with-event-handlers
#:with-event-handlers
;; sync
:bind
:accept
:connect-to-ip
#:bind
#:accept
#:connect-to-ip
;; network
:connection
:connection-ip
:connection-port
:connection-conn
#:connection
#:connection-ip
#:connection-port
#:connection-conn
#:*known-protocols*
#:make-connection-from-uri
;; client
:write-block
:get-block
:get-blocks
:write-error
:write-ok
:subscribe-channels
:announce
#:write-block
#:get-block
#:get-blocks
#:write-error
#:write-ok
#:subscribe-channels
#:announce
;; protocol
#:read-event
#:write-error
#:write-ok
#:write-block
#:subscribe-channels
#:announce
#:close-connection
;; accessors
:get-item
:put-item
:listing-generator
:stop-put-p
:system-sync-limit
#:get-item
#:put-item
#:listing-generator
#:stop-put-p
#:system-connection-list
#:system-sync-limit
;; readers
:system-port))
#:system-port))
......@@ -4,7 +4,7 @@
(in-package :decentralise)
(defclass system ()
((connection-list :initform nil)
((connection-list :initform nil :accessor system-connection-list)
(connection :initform nil)
(port :initarg :port
:initform 1892
......@@ -22,14 +22,16 @@ distributed program and its networking.
(defclass connection ()
((ip :initform "" :initarg :ip :reader connection-ip)
(port :initform 1892 :initarg :port :reader connection-port)
(conn :initform nil :initarg :connection :accessor connection-conn))
(conn :initarg :connection :accessor connection-conn)
(raw-socket :initarg :socket :accessor connection-socket))
(:documentation "A representation of a connection to a decentralise instance.
This representation assumes that the client has an IP address and port; but
these may usually remain at their defaults or possibly unbound with no detriment
to the core of cl-decentralise.
:IP sets the IP address of the client, and similarly
:PORT sets the port.
:CONN is the underlying stream the connection relies on."))
:CONN is the underlying stream the connection relies on.
:SOCKET is the (optional) socket CONN uses."))
(defstruct connection-meta
"The internal state of a connection, including the connection itself,
......@@ -62,14 +64,39 @@ and a flag set if the node should be published in the NODES block (ANNOUNCEP)."
(defgeneric handle-lines-from (connection system meta))
(defgeneric repl (connection system))
(defgeneric accept (system &key log))
(defgeneric connect-to (system host &key port timeout announcep))
(defgeneric connect-repl (system uri &key port timeout announcep))
(defgeneric connect-to (protocol-name host &key port timeout))
(defgeneric close-connection (connection))
;;; Network functions.
(defgeneric get-blocks (connection names))
(defun get-block (connection name)
;; A trivial cover over GET-BLOCKS.
(get-blocks connection (list name)))
(defgeneric write-block (connection name values &key version channels))
(defgeneric write-error (connection name &optional reason))
(defgeneric write-ok (connection name))
(defgeneric subscribe-channels (connection channels))
(defgeneric announce (connection &optional announcep))
(defgeneric read-event (connection))
;;; Node metadata.
(defgeneric protocol-name (connection))
(defgeneric connection-name (connection))
(defvar *known-protocols* (make-hash-table :test 'equal))
(defun make-connection-from-uri (uri &key (timeout 5))
(destructuring-bind (name ip port)
(split-sequence #\: uri)
(make-connection-from-info name ip (parse-integer port) timeout)))
(defun make-connection-from-info (name ip port timeout)
(multiple-value-bind (creator win)
(gethash name *known-protocols*)
(if (null win)
(error "no protocol with name ~a" name)
(funcall creator ip port timeout))))
......@@ -10,12 +10,6 @@
~t:Connection ~a
~t:Port ~a" connection-list connection port))))
(defmethod bind ((system system) &key (host "0.0.0.0"))
(setf (slot-value system 'connection)
(socket-listen host
(slot-value system 'port)
:reuse-address t)))
(defun get-every-list (fn)
(loop named list-collecting-loop
with table = (make-hash-table :test 'equal)
......@@ -35,7 +29,8 @@
unless (eql node system)
collect (connection-ip
(connection-meta-socket
node)))))
node)))
:version 0))
((string= name "list")
(write-block stream
"list"
......@@ -43,7 +38,8 @@
with table = (get-every-list (listing-generator system))
for name being the hash-keys of table
for version being the hash-values of table
collect (format nil "~a ~d" name version))))
collect (format nil "~a ~d" name version))
:version 0))
(t
(let (data version channels)
(handler-case
......@@ -91,18 +87,21 @@
(bt:with-lock-held (lock)
(get-blocks connection take))))
(defmethod handle-others-nodes (system text)
(defmethod handle-others-nodes ((system system) text)
(let* ((our-nodes
(remove-duplicates
(loop
for node in (slot-value system 'connection-list)
for socket = (slot-value node 'socket)
for ip = (slot-value node 'ip)
for conn = (connection-meta-socket node)
for ip = (connection-ip conn)
unless (member ip *ip-blacklist* :test #'string=)
collecting ip)
collecting (connection-name conn))
:test #'string=))
(new-nodes (set-difference text our-nodes
:test #'string=)))
(new-nodes (remove-duplicates
(set-difference (split-sequence #\Newline text)
our-nodes
:test #'string=)
:test #'string=)))
(loop
for node in new-nodes
while (< (length (slot-value system 'connection-list))
......@@ -110,9 +109,11 @@
do (bt:make-thread
(lambda ()
(ignore-errors
(connect-to system (print node)
:timeout 2
:announcep t)))
(format t "Trying to connect to ~a~%" node)
(connect-repl system node
:timeout 2
:announcep t
:subscriptions '("all"))))
:name "Discovered node"))))
(defmethod handle-new-block (name version text (system system) connection meta channels)
......@@ -188,10 +189,10 @@
(handle-lines-from connection system meta)))
(end-of-file ()
(format t "Byebye, ~a.~%" connection))
#-debug
#-debug
(condition (c)
(format t "Killing ~a because ~a~%" connection c)
(close (connection-conn connection))
(close-connection connection)
(setf (slot-value system 'connection-list)
(delete meta
(slot-value system 'connection-list)
......@@ -216,23 +217,17 @@
(get-peer-address connection)
(get-peer-port connection))))))
(defmethod connect-to ((system system) host &key
(port (system-port system))
(timeout 5)
(announcep t)
(subscriptions nil))
(let* ((socket (socket-connect host port :timeout timeout))
(raw-conn (socket-stream socket))
(conn (make-instance 'connection
:ip host
:port port
:connection raw-conn)))
(defmethod connect-repl ((system system) uri &key
(port (system-port system))
(timeout 5)
(announcep t)
(subscriptions nil))
(let* ((conn (make-connection-from-uri uri :timeout timeout)))
;; ask for a file listing and all nodes
(get-blocks conn '("list" "nodes"))
(when announcep
(announce conn t))
(subscribe-channels conn subscriptions)
(finish-output raw-conn)
(unwind-protect
(repl conn system)
(socket-close socket))))
(close-connection conn))))
......@@ -13,7 +13,7 @@
(defmacro test-output (system input &optional expected-output)
(alexandria:with-gensyms (input-stream output-stream two-way-stream
connection-meta connection)
`(ignore-errors
`(progn
(let* (;; Streams
(,input-stream (make-string-input-stream ,input))
(,output-stream (make-string-output-stream))
......
......@@ -70,6 +70,6 @@ He taught her how to split and define
--EOF--")
"GET/BLOCK is not identical")
(let ((listing (test-output *system* "GET list")))
(is (and (search "foo 1" listing)
(search "bar 2" listing))
(is (and (search "foo 1" listing :test #'string=)
(search "bar 2" listing :test #'string=))
"Could not find `foo 1` and `bar 2` in the listing")))
......@@ -7,15 +7,14 @@
:accessor meta-table)))
(defmethod listing-generator ((system my-system))
(let ((iterating t))
(with-hash-table-iterator (iterator (meta-table system))
(lambda ()
(when iterating
(multiple-value-bind (success name metadata)
(iterator)
(if success
(values name (car metadata))
(setf iterating nil))))))))
(let ((alist (alexandria:hash-table-alist (meta-table system))))
(lambda ()
(if (null alist)
nil
(destructuring-bind (name version &rest channels)
(pop alist)
(declare (ignore channels))
(values name version))))))
(defmethod put-item ((system my-system) name text version channels)
(setf (gethash name (data-table system)) text
......
(in-package :decentralise-websocket)
(defclass websocket-client-connection (decentralise:connection)
((decentralise::port :initarg :port :initform 80 :reader decentralise:connection-port)
(https :initarg :https :initform nil :reader decentralise-websocket-client-https-p)
(event-list :initform nil :accessor websocket-connection-event-list)))
(defmethod initialize-instance :after ((client websocket-client-connection) &key)
(let ((connection (wsd:make-client
(format nil "~:[ws~;wss~]://~a:~d/decentralise.ws"
(decentralise-websocket-client-https-p client)
(decentralise:connection-ip client)
(decentralise:connection-port client)))))
(setf (decentralise:connection-conn client) connection)
(wsd:on :message connection
(lambda (message)
(push (jsown:parse message)
(websocket-connection-event-list client))))
(wsd:start-connection connection)))
(defmethod decentralise:read-event ((connection websocket-client-connection))
(loop while (null (websocket-connection-event-list connection))
do (sleep 0.05))
(let* ((event (pop (websocket-connection-event-list connection)))
(event-type (val event "type")))
(macrolet ((prop (name) `(val event ,name)))
(cond
((string= event-type "get") `(:get ,(prop "names")))
((string= event-type "block") `(:block ,(prop "name")
,(prop "version")
,(prop "channels")
,(prop "data")))
((string= event-type "ok") `(:ok ,(prop "name")))
((string= event-type "error") `(:error ,(prop "name")
,(prop "reason")))
((string= event-type "subscribe") `(:subscribe ,(prop "channels")))
((string= event-type "announce") `(:announce ,(string= (prop "new_state") "yes")))
(t (error "invalid event-type: ~s" event-type))))))
(defmethod protocol-name ((connection websocket-client-connection)) "websocket")
(setf (gethash "websocket" decentralise:*known-protocols*)
(lambda (ip port timeout)
(trivial-timeout:with-timeout (timeout)
(make-instance 'websocket-client-connection :ip ip :port port))))
;;;; example.lisp: a short example for using cl-decentralised to create
;;; a very simple synchronised data store.
;;; This file is a part of cl-decentralise.
(defpackage :decentralise-websocket-example
(:use :cl :decentralise :decentralise-websocket))
(in-package :decentralise-websocket-example)
(defclass my-system (websocket-listener)
((data-table :initform (make-hash-table :test 'equal)
:accessor data-table)
(meta-table :initform (make-hash-table :test 'equal)
:accessor meta-table)))
(defmethod listing-generator ((system my-system))
(let ((iterating t))
(with-hash-table-iterator (iterator (meta-table system))
(lambda ()
(when iterating
(multiple-value-bind (success name metadata)
(iterator)
(if success
(values name (car metadata))
(setf iterating nil))))))))
(defmethod put-item ((system my-system) name text version channels)
(setf (gethash name (data-table system)) text
(gethash name (meta-table system)) (cons version channels)))
(defmethod get-item ((system my-system) name)
(multiple-value-bind (data win)
(gethash name (data-table system))
(if win
(destructuring-bind (version &rest channels)
(gethash name (meta-table system))
(values data version channels))
(error "not found"))))
(defvar *system* (make-instance 'my-system))
(defvar *server* (make-instance 'hunchensocket:websocket-acceptor :port 12345))
(push (lambda (request)
(when (string= (hunchentoot:script-name request) "/decentralise.ws") *system*))
hunchensocket:*websocket-dispatch-table*)
(hunchentoot:start *server*)
(in-package :decentralise-websocket)
(defclass websocket-listener (decentralise:system websocket-resource)
()
(:default-initargs :client-class 'websocket-connection))
(defclass websocket-connection (decentralise:connection websocket-client)
((event-list :initform nil :accessor websocket-connection-event-list)
(thread :initarg :thread :accessor websocket-connection-thread)))
;;; The hunchensocket part.
(defmethod client-connected ((listener websocket-listener) client)
(setf (websocket-connection-thread client)
(make-thread (lambda ()
(decentralise::repl client listener)))))
(defmethod client-disconnected ((listener websocket-listener) client)
(setf (decentralise:system-connection-list listener)
(delete client (decentralise:system-connection-list listener)
:key #'decentralise::connection-meta-socket))
(bt:destroy-thread (websocket-connection-thread client)))
(defmethod text-message-received (listener (client websocket-connection) text)
(push (jsown:parse text) (websocket-connection-event-list client)))
;;; The cl-decentralise part.
(defmethod decentralise:bind ((system websocket-listener) &key host)
(error "Websockets can't bind without a Hunchentoot listener, push me
into a dispatch table instead"))
(defmethod decentralise:read-event ((connection websocket-connection))
(loop while (null (websocket-connection-event-list connection))
do (sleep 0.05))
(let* ((event (pop (websocket-connection-event-list connection)))
(event-type (val event "type")))
(macrolet ((prop (name) `(val event ,name)))
(cond
((string= event-type "get") `(:get ,(prop "names")))
((string= event-type "block") `(:block ,(prop "name")
,(prop "version")
,(prop "channels")
,(prop "data")))
((string= event-type "ok") `(:ok ,(prop "name")))
((string= event-type "error") `(:error ,(prop "name")
,(prop "cause")))
((string= event-type "subscribe") `(:subscribe ,(prop "channels")))
((string= event-type "announce") `(:announce ,(string= (prop "new_state") "yes")))
(t (error "invalid event-type: ~s" event-type))))))
(defmethod protocol-name ((connection websocket-connection)) "websocket")
(in-package :decentralise-websocket)
;;; Host
(defmethod decentralise:write-error ((connection websocket-connection) name &optional reason)
(send-text-message connection
(to-json `(:obj ("type" . "error")
("name" . ,name)
("reason" . ,reason)))))
(defmethod decentralise:write-ok ((connection websocket-connection) name)
(send-text-message connection
(to-json `(:obj ("type" . "ok")
("name" . ,name)))))
(defmethod decentralise:write-block ((connection websocket-connection) name values
&key version channels)
(let ((values
(etypecase values
(string values)
(list (format nil "~{~a~^~%~}" values)))))
(send-text-message connection
(to-json `(:obj ("type" . "block")
("name" . ,name)
("data" . ,values)
("version" . ,version)
("channels" . ,channels))))))
(defmethod decentralise:subscribe-channels ((connection websocket-connection) channels)
(send-text-message connection
(to-json `(:obj ("type" . "subscribe")
("channels" . ,channels)))))
(defmethod decentralise:announce ((connection websocket-connection) &optional (announcep t))
(send-text-message connection
(to-json `(:obj ("type" . "announce")
("new_state" . ,(if announcep "yes" "no"))))))
(defmethod decentralise:get-blocks ((connection websocket-connection) names)
(send-text-message connection
(to-json `(:obj ("type" . "get")
("names" . ,names)))))
(defmethod decentralise:close-connection ((connection websocket-connection))
(hunchensocket:close-connection connection))
;;; Client
(defmethod decentralise:write-error ((connection websocket-client-connection) name &optional reason)
(send-text-message (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "error")
("name" . ,name)
("reason" . ,reason)))))
(defmethod decentralise:write-ok ((connection websocket-client-connection) name)
(send-text-message (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "ok")
("name" . ,name)))))
(defmethod decentralise:write-block ((connection websocket-client-connection) name values
&key version channels)
(let ((values
(etypecase values
(string values)
(list (format nil "~{~a~^~%~}" values)))))
(wsd:send (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "block")
("name" . ,name)
("data" . ,values)
("version" . ,version)
("channels" . ,channels))))))
(defmethod decentralise:subscribe-channels ((connection websocket-client-connection) channels)
(wsd:send (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "subscribe")
("channels" . ,channels)))))
(defmethod decentralise:announce ((connection websocket-client-connection) &optional (announcep t))
(wsd:send (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "announce")
("new_state" . ,(if announcep "yes" "no"))))))
(defmethod decentralise:get-blocks ((connection websocket-client-connection) names)
(wsd:send (decentralise:connection-conn connection)
(to-json `(:obj ("type" . "get")
("names" . ,names)))))
(defmethod decentralise:close-connection ((connection websocket-client-connection))
(wsd:close-connection (decentralise:connection-conn connection)))
(defpackage :decentralise-websocket
(:use :cl :hunchensocket :hunchentoot :jsown :bt)
(:export #:websocket-listener #:websocket-connection #:websocket-client-connection))
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment