.gitignore

.gitignore

@@ -0,0 +1 @@
+compiled/

marketplace-support.rkt

@@ -0,0 +1,12 @@
+#lang racket/base
+;; Reexport racket-matrix module contents.
+
+(require marketplace/sugar-untyped)
+(require marketplace/drivers/tcp)
+(require marketplace/drivers/timer-untyped)
+(require marketplace/drivers/event-relay)
+
+(provide (all-from-out marketplace/sugar-untyped))
+(provide (all-from-out marketplace/drivers/tcp))
+(provide (all-from-out marketplace/drivers/timer-untyped))
+(provide (all-from-out marketplace/drivers/event-relay))

new-server.rkt

@@ -14,22 +14,20 @@
 (require "ssh-channel.rkt")
 (require "ssh-message-types.rkt")
 (require "ssh-exceptions.rkt")
-(require "os2-support.rkt")
+(require "marketplace-support.rkt")
 
 (define (main)
-  (ground-vm
-   (transition 'no-state
-     (spawn (timer-driver 'timer-driver))
-     ;; PAPER NOTE: remove #:debug-name for presentation economy
-     (spawn tcp-driver #:debug-name 'tcp-driver)
-     (spawn tcp-spy #:debug-name 'tcp-spy)
-     (spawn listener #:debug-name 'ssh-tcp-listener))))
+  (ground-vm (timer-driver)
+	     (tcp-driver)
+	     (tcp-spy)
+	     (spawn #:debug-name 'ssh-tcp-listener #:child listener)))
 
 (define listener
-  (transition 'no-state
-    (role (tcp-listener 2322)
-      #:topic t
-      #:on-presence (spawn (session-vm t) #:debug-name (debug-name 'ssh-session-vm t)))))
+  (transition/no-state
+    (endpoint #:subscriber (tcp-channel ? (tcp-listener 2322) ?)
+	      #:observer
+	      #:conversation r
+	      #:on-presence (session-vm r))))
 
 ;;---------------------------------------------------------------------------
 
@@ -45,59 +43,81 @@
 	   peer-identification-string)))
 
 (define (spy marker)
-  (role (or (topic-subscriber (wild) #:monitor? #t)
-	    (topic-publisher (wild) #:monitor? #t))
-    [message
-     (write `(,marker ,message))
-     (newline)
-     (flush-output)
-     (void)]))
+  (define (dump what message)
+    (write `(,marker ,what ,message))
+    (newline)
+    (flush-output)
+    (void))
+  (list
+   (endpoint #:subscriber (wild) #:everything
+	     #:role r
+	     #:on-presence (dump 'arrived r)
+	     #:on-absence (dump 'departed r)
+	     [message (dump 'message message)])
+   (endpoint #:publisher (wild) #:everything
+	     #:role r
+	     #:on-presence (dump 'arrived r)
+	     #:on-absence (dump 'departed r)
+	     [message (dump 'message message)])))
 
-(define-syntax-rule (wait-for topic-of-interest action ...)
-  (role/fresh role-name topic-of-interest
-    #:state state
-    #:on-presence (sequence-actions (transition state)
-		    (delete-role role-name)
-		    action ...)))
+(define-syntax-rule (wait-as my-orientation topic action ...)
+  (endpoint my-orientation topic #:observer
+	    #:let-name endpoint-name
+	    #:state state
+	    #:on-presence (begin (printf "WAIT ENDED: ~v\n" topic)
+				 (sequence-actions (transition state
+						     (delete-endpoint endpoint-name)
+						     action ...)))))
 
-(define (session-vm new-connection-topic)
-  (define-values (cin cout in-topic out-topic) (topic->tcp-connection new-connection-topic))
+(define (session-vm new-conversation)
+  (match-define (tcp-channel remote-addr local-addr _) new-conversation)
   (define local-identification #"SSH-2.0-RacketSSH_0.0")
 
   (define (issue-identification-string)
-    (at-meta-level (cout (bytes-append local-identification #"\r\n"))))
+    (at-meta-level
+     (send-message (tcp-channel local-addr remote-addr
+				(bytes-append local-identification #"\r\n")))))
 
   (define (read-handshake-and-become-reader)
     (transition 'handshake-is-stateless ;; but, crucially, the ssh-reader proper isn't!
       (at-meta-level
-       (role in-topic
-	 #:name 'socket-reader
-	 #:state state
+       (endpoint #:subscriber (tcp-channel remote-addr local-addr ?)
+		 #:name 'socket-reader
+		 #:state state
 	 [(tcp-channel _ _ (? eof-object?))
 	  (transition state (quit))]
 	 [(tcp-channel _ _ (? bytes? remote-identification))
-	  (check-remote-identification! remote-identification)
-	  (sequence-actions (transition state)
-	    ;; First, set the incoming mode to bytes.
-	    (at-meta-level (cin (tcp-mode 'bytes)))
-	    ;; Then initialise the reader, switching to packet-reading mode.
-	    (lambda (ignored-state) (ssh-reader new-connection-topic))
-	    ;; Finally, spawn the remaining processes and issue the initial credit to the reader.
-	    (spawn (ssh-writer new-connection-topic)
-		   #:exit-signal? #t
-		   #:debug-name 'ssh-writer)
-	    ;; Wait for the reader and writer get started, then tell
-	    ;; the reader we are ready for a single packet and spawn
-	    ;; the session manager.
-	    (wait-for (topic-subscriber (inbound-packet (wild) (wild) (wild) (wild)) #:monitor? #t)
-	      (wait-for (topic-publisher (outbound-packet (wild)) #:monitor? #t)
-		(send-message (inbound-credit 1))
-		(spawn (ssh-session local-identification
-				    remote-identification
-				    repl-boot
-				    'server)
-		       #:exit-signal? #t
-		       #:debug-name 'ssh-session))))]))))
+	  (begin
+	    (check-remote-identification! remote-identification)
+	    (sequence-actions (transition state)
+	      ;; First, set the incoming mode to bytes.
+	      (at-meta-level
+	       (send-feedback (tcp-channel remote-addr local-addr (tcp-mode 'bytes))))
+	      ;; Then initialise the reader, switching to packet-reading mode.
+	      (lambda (ignored-state) (ssh-reader new-conversation))
+	      ;; Finally, spawn the remaining processes and issue the initial credit to the reader.
+	      (spawn #:debug-name 'ssh-writer
+		     #:child (ssh-writer new-conversation)
+		     ;; TODO: canary: #:exit-signal? #t
+		     )
+	      ;; Wait for the reader and writer get started, then tell
+	      ;; the reader we are ready for a single packet and spawn
+	      ;; the session manager.
+	      (printf "BOO\n")
+	      (wait-as #:subscriber (inbound-packet (wild) (wild) (wild) (wild))
+		       (printf "YAY\n") (flush-output)
+		       (wait-as #:publisher (outbound-packet (wild))
+				(printf "ALSO YAY\n") (flush-output)
+				(send-message (inbound-credit 1))
+				(spawn #:debug-name 'ssh-session
+				       #:pid session-pid
+				       #:child (ssh-session session-pid
+							    local-identification
+							    remote-identification
+							    repl-boot
+							    'server)
+				       ;; TODO: canary: #:exit-signal? #t
+				       )))))]))))
 
   (define (exn->outbound-packet reason)
     (outbound-packet (ssh-msg-disconnect (exn:fail:contract:protocol-reason-code reason)
@@ -123,44 +143,43 @@
   (define (inert-exception-handler reason)
     inert-exception-handler)
 
-  (nested-vm #:debug-name (debug-name 'ssh-session-vm new-connection-topic)
-    (transition 'no-state
-      (spawn event-relay #:debug-name (debug-name 'session-event-relay))
-      (spawn (timer-relay 'ssh-timer-relay) #:debug-name 'ssh-timer-relay)
-      (spy 'SSH)
+  (nested-vm #:debug-name (list 'ssh-session-vm new-conversation)
+    (event-relay 'ssh-event-relay)
+    (timer-relay 'ssh-timer-relay)
+    (spy 'SSH)
 
-      (issue-identification-string)
+    (issue-identification-string)
 
-      ;; Expect identification string, then update (!) our inbound
-      ;; subscription handler to switch to packet mode.
-      (at-meta-level (cin (tcp-mode 'lines)))
-      (at-meta-level (cin (tcp-credit 1)))
+    ;; Expect identification string, then update (!) our inbound
+    ;; subscription handler to switch to packet mode.
+    (at-meta-level
+     (send-feedback (tcp-channel remote-addr local-addr (tcp-mode 'lines)))
+     (send-feedback (tcp-channel remote-addr local-addr (tcp-credit 1))))
 
-      (spawn (read-handshake-and-become-reader)
-	     #:exit-signal? #t
-	     #:debug-name 'ssh-reader)
+    (spawn #:debug-name 'ssh-reader
+	   #:child (read-handshake-and-become-reader)
+	   ;; TODO: canary: #:exit-signal? #t
+	   )
 
-      (spawn (transition active-exception-handler
-	       (role (topic-subscriber (exit-signal (wild) (wild)))
-		 #:state current-handler
-		 #:reason reason
-		 #:on-absence (current-handler reason)))))))
+    ;; TODO: canary:
+    ;; (spawn #:child
+    ;; 	   (transition active-exception-handler
+    ;; 	     (role (topic-subscriber (exit-signal (wild) (wild)))
+    ;; 		   #:state current-handler
+    ;; 		   #:reason reason
+    ;; 		   #:on-absence (current-handler reason))))
+    ))
 
 ;;---------------------------------------------------------------------------
 
 (define (repl-boot user-name)
-  (transition 'no-repl-state
-    (spawn event-relay #:debug-name (debug-name 'repl-event-relay))
-    (spy 'APP)
-    (at-meta-level
-     (role (topic-subscriber (channel-message (channel-stream-name #t (wild)) (wild)))
-       #:state state
-       #:topic t
-       #:on-presence (match t
-		       [(topic _ (channel-message (channel-stream-name _ cname) _) _)
-			(transition state (spawn (repl-instance user-name cname)
-						 #:debug-name cname))])))))
-
+  (list
+   (event-relay 'app-event-relay)
+   (spy 'APP)
+   (at-meta-level
+    (endpoint #:subscriber (channel-message (channel-stream-name #t (wild)) (wild))
+	      #:conversation (channel-message (channel-stream-name _ cname) _)
+	      #:on-presence (spawn #:debug-name cname #:child (repl-instance user-name cname))))))
 
 ;; (repl-instance InputPort OutputPort InputPort OutputPort)
 (struct repl-instance-state (c2s-in ;; used by thread to read input from relay
@@ -193,9 +212,9 @@
        (define repl-thread (thread (lambda () (repl-shell user-name c2s-in s2c-out))))
        (transition state
 	 (ch-do send-feedback inbound-stream (channel-stream-ok))
-	 (role (topic-subscriber (cons (thread-dead-evt repl-thread) (wild)))
+	 (endpoint #:subscriber (cons (thread-dead-evt repl-thread) (wild))
 	   [_ (quit #:reason "REPL thread exited")])
-	 (role (topic-subscriber (cons (peek-bytes-avail!-evt dummy-buffer 0 #f s2c-in) (wild)))
+	 (endpoint #:subscriber (cons (peek-bytes-avail!-evt dummy-buffer 0 #f s2c-in) (wild))
 	   ;; We're using peek-bytes-avail!-evt rather than
 	   ;; read-bytes-avail!-evt because of potential overwriting
 	   ;; of the buffer. The overwriting can happen when there's
@@ -204,22 +223,24 @@
 	   ;; will overwrite its buffer next time it's synced on.
 	   #:state state
 	   [(cons _ (? eof-object?))
-	    (match-define (repl-instance-state c2s-in c2s-out s2c-in s2c-out) state)
-	    (close-input-port c2s-in)
-	    (close-output-port c2s-out)
-	    (close-input-port s2c-in)
-	    (close-output-port s2c-out)
-	    (transition state (quit))]
+	    (let ()
+	      (match-define (repl-instance-state c2s-in c2s-out s2c-in s2c-out) state)
+	      (close-input-port c2s-in)
+	      (close-output-port c2s-out)
+	      (close-input-port s2c-in)
+	      (close-output-port s2c-out)
+	      (transition state (quit)))]
 	   [(cons _ (? number? count))
 	    (transition state
 	      (ch-do send-message outbound-stream (channel-stream-data
 						   (read-bytes count s2c-in))))]))]
       [(or (channel-stream-data #"\4") ;; C-d a.k.a EOT
 	   (channel-stream-eof))
-       (close-output-port (repl-instance-state-c2s-out state))
-       ;; ^ this signals the repl thread to exit.
-       ;; Now, wait for it to do so.
-       (transition state)]
+       (let ()
+	 (close-output-port (repl-instance-state-c2s-out state))
+	 ;; ^ this signals the repl thread to exit.
+	 ;; Now, wait for it to do so.
+	 (transition state))]
       [(channel-stream-data bs)
        (write-bytes bs (repl-instance-state-c2s-out state))
        (flush-output (repl-instance-state-c2s-out state))
@@ -234,7 +255,7 @@
      (define-values (s2c-in s2c-out) (make-pipe))
      (transition (repl-instance-state c2s-in c2s-out s2c-in s2c-out)
        (at-meta-level
-	(role (topic-subscriber (channel-message inbound-stream (wild)))
+	(endpoint #:subscriber (channel-message inbound-stream (wild))
 	  #:state state
 	  #:on-presence (transition state
 			  (ch-do send-feedback inbound-stream (channel-stream-config
@@ -244,12 +265,13 @@
 	  [(channel-message _ body)
 	   (handle-channel-message state body)]))
        (at-meta-level
-	(role (topic-publisher (channel-message outbound-stream (wild)))
+	(endpoint #:publisher (channel-message outbound-stream (wild))
 	  [m
-	   (write `(channel outbound ,cname ,m)) (newline)
-	   (void)])))]
+	   (begin
+	     (write `(channel outbound ,cname ,m)) (newline)
+	     (void))])))]
     [type
-     (transition 'no-instance-state
+     (transition/no-state
        (at-meta-level (send-message
 		       (channel-message outbound-stream
 					(channel-stream-open-failure

os2-support.rkt

@@ -1,12 +0,0 @@
-#lang racket/base
-;; Reexport racket-matrix module contents.
-
-(require "../racket-matrix/os2.rkt")
-(require "../racket-matrix/os2-event-relay.rkt")
-(require "../racket-matrix/os2-timer.rkt")
-(require "../racket-matrix/fake-tcp.rkt")
-
-(provide (all-from-out "../racket-matrix/os2.rkt"))
-(provide (all-from-out "../racket-matrix/os2-event-relay.rkt"))
-(provide (all-from-out "../racket-matrix/os2-timer.rkt"))
-(provide (all-from-out "../racket-matrix/fake-tcp.rkt"))

ssh-channel.rkt

@@ -3,12 +3,6 @@
 (require racket/set)
 (require racket/match)
 
-(require "ssh-numbers.rkt")
-(require "ssh-message-types.rkt")
-(require "ssh-exceptions.rkt")
-
-(require "os2-support.rkt")
-
 (provide (struct-out ssh-channel)
 
 	 (struct-out channel-name)

ssh-session.rkt

@@ -16,7 +16,7 @@
 (require "ssh-transport.rkt")
 (require "ssh-channel.rkt")
 
-(require "os2-support.rkt")
+(require "marketplace-support.rkt")
 
 (provide rekey-interval
 	 rekey-volume
@@ -495,10 +495,9 @@
 			 conn))))
       (lambda (conn)
 	(transition conn
-	  (spawn (nested-vm #:debug-name 'ssh-application-vm
-		   ((connection-application-boot conn) user-name))
-		 #:exit-signal? #t
-		 #:debug-name 'ssh-application-vm))))]
+	  ;; TODO: canary for NESTED VM!: #:exit-signal? #t
+	  (nested-vm #:debug-name 'ssh-application-vm
+		     ((connection-application-boot conn) user-name)))))]
    [else
     (transition conn
       (send-message (outbound-packet (ssh-msg-userauth-failure '(none) #f))))]))
@@ -597,68 +596,70 @@
 	  [(remote)
 	   (case old-close-state
 	     [(neither local)
-	      (list (delete-role (list cname 'outbound))
-		    (delete-role (list cname 'inbound)))]
+	      (list (delete-endpoint (list cname 'outbound))
+		    (delete-endpoint (list cname 'inbound)))]
 	     [else (list)])])))]
    [else (transition conn)]))
 
-(define (channel-roles cname initial-message-producer)
+(define (channel-endpoints cname initial-message-producer)
   (define inbound-stream-name (channel-stream-name #t cname))
   (define outbound-stream-name (channel-stream-name #f cname))
   (define (! conn message)
     (transition conn (send-message (outbound-packet message))))
   (list
-   (role (topic-subscriber (channel-message outbound-stream-name (wild)))
-     #:name (list cname 'outbound)
-     #:state conn
-     #:on-presence (transition conn
-		     (initial-message-producer inbound-stream-name outbound-stream-name))
-     #:on-absence (maybe-close-channel cname conn 'local)
+   (endpoint #:subscriber (channel-message outbound-stream-name (wild))
+	     #:name (list cname 'outbound)
+	     #:state conn
+	     #:on-presence (transition conn
+			     (initial-message-producer inbound-stream-name outbound-stream-name))
+	     #:on-absence (maybe-close-channel cname conn 'local)
      [(channel-message _ body)
-      (define ch (findf (ssh-channel-name=? cname) (connection-channels conn)))
-      (define remote-ref (ssh-channel-remote-ref ch))
-      (match body
-	[(channel-stream-data data-bytes)
-	 ;; TODO: split data-bytes into packets if longer than max packet size
-	 (! conn (ssh-msg-channel-data remote-ref data-bytes))]
-	[(channel-stream-extended-data type data-bytes)
-	 (! conn (ssh-msg-channel-extended-data remote-ref type data-bytes))]
-	[(channel-stream-eof)
-	 (! conn (ssh-msg-channel-eof remote-ref))]
-	[(channel-stream-notify type data-bytes)
-	 (! conn (ssh-msg-channel-request remote-ref type #f data-bytes))]
-	[(channel-stream-request type data-bytes)
-	 (! conn (ssh-msg-channel-request remote-ref type #t data-bytes))]
-	[(channel-stream-open-failure reason description)
-	 (! (discard-channel cname conn)
-	    (ssh-msg-channel-open-failure remote-ref reason description #""))])])
-   (role (topic-publisher (channel-message inbound-stream-name (wild)))
-     #:name (list cname 'inbound)
-     #:state conn
+      (let ()
+	(define ch (findf (ssh-channel-name=? cname) (connection-channels conn)))
+	(define remote-ref (ssh-channel-remote-ref ch))
+	(match body
+	  [(channel-stream-data data-bytes)
+	   ;; TODO: split data-bytes into packets if longer than max packet size
+	   (! conn (ssh-msg-channel-data remote-ref data-bytes))]
+	  [(channel-stream-extended-data type data-bytes)
+	   (! conn (ssh-msg-channel-extended-data remote-ref type data-bytes))]
+	  [(channel-stream-eof)
+	   (! conn (ssh-msg-channel-eof remote-ref))]
+	  [(channel-stream-notify type data-bytes)
+	   (! conn (ssh-msg-channel-request remote-ref type #f data-bytes))]
+	  [(channel-stream-request type data-bytes)
+	   (! conn (ssh-msg-channel-request remote-ref type #t data-bytes))]
+	  [(channel-stream-open-failure reason description)
+	   (! (discard-channel cname conn)
+	      (ssh-msg-channel-open-failure remote-ref reason description #""))]))])
+   (endpoint #:publisher (channel-message inbound-stream-name (wild))
+	     #:name (list cname 'inbound)
+	     #:state conn
      [(channel-message _ body)
-      (define ch (findf (ssh-channel-name=? cname) (connection-channels conn)))
-      (define remote-ref (ssh-channel-remote-ref ch))
-      (match body
-	[(channel-stream-config maximum-packet-size extra-data)
-	 (if (channel-name-locally-originated? cname)
-	     ;; This must be intended to form the SSH_MSG_CHANNEL_OPEN.
-	     (! conn (ssh-msg-channel-open (channel-name-type cname)
-					   (ssh-channel-local-ref ch)
-					   0
-					   maximum-packet-size
-					   extra-data))
-	     ;; This must be intended to form the SSH_MSG_CHANNEL_OPEN_CONFIRMATION.
-	     (! conn (ssh-msg-channel-open-confirmation remote-ref
-							(ssh-channel-local-ref ch)
-							0
-							maximum-packet-size
-							extra-data)))]
-	[(channel-stream-credit count)
-	 (! conn (ssh-msg-channel-window-adjust remote-ref count))]
-	[(channel-stream-ok)
-	 (! conn (ssh-msg-channel-success remote-ref))]
-	[(channel-stream-fail)
-	 (! conn (ssh-msg-channel-failure remote-ref))])])))
+      (let ()
+	(define ch (findf (ssh-channel-name=? cname) (connection-channels conn)))
+	(define remote-ref (ssh-channel-remote-ref ch))
+	(match body
+	  [(channel-stream-config maximum-packet-size extra-data)
+	   (if (channel-name-locally-originated? cname)
+	       ;; This must be intended to form the SSH_MSG_CHANNEL_OPEN.
+	       (! conn (ssh-msg-channel-open (channel-name-type cname)
+					     (ssh-channel-local-ref ch)
+					     0
+					     maximum-packet-size
+					     extra-data))
+	       ;; This must be intended to form the SSH_MSG_CHANNEL_OPEN_CONFIRMATION.
+	       (! conn (ssh-msg-channel-open-confirmation remote-ref
+							  (ssh-channel-local-ref ch)
+							  0
+							  maximum-packet-size
+							  extra-data)))]
+	  [(channel-stream-credit count)
+	   (! conn (ssh-msg-channel-window-adjust remote-ref count))]
+	  [(channel-stream-ok)
+	   (! conn (ssh-msg-channel-success remote-ref))]
+	  [(channel-stream-fail)
+	   (! conn (ssh-msg-channel-failure remote-ref))]))])))
 
 (define (channel-notify conn ch inbound? body)
   (transition conn
@@ -670,11 +671,15 @@
 ;; Connection service
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
+(define (respond-to-opened-outbound-channel conn cname)
+  (if (and (ground? cname)
+	   (not (memf (ssh-channel-name=? cname) (connection-channels conn))))
+      (transition (update-channel cname values conn)
+	(channel-endpoints cname (lambda (inbound-stream outbound-stream)
+				   '())))
+      (transition conn)))
+
 (define (start-connection-service conn)
-  (define arbitrary-locally-originated-stream
-    (channel-stream-name (wild) (channel-name #t (wild) (wild))))
-  (define arbitrary-locally-originated-traffic
-    (channel-message arbitrary-locally-originated-stream (wild)))
   (sequence-actions
       (transition
 	  (set-handlers conn
@@ -694,20 +699,16 @@
     ;; application. We are responding to channels appearing from the
     ;; remote peer by virtue of our installation of the handler for
     ;; SSH_MSG_CHANNEL_OPEN above.
-    (role (set (topic-publisher arbitrary-locally-originated-traffic #:monitor? #t)
-	       (topic-subscriber arbitrary-locally-originated-traffic #:monitor? #t))
-      #:state conn
-      #:topic t
-      #:on-presence
-      (match t
-	[(or (topic 'publisher (channel-message (channel-stream-name #f cname) _) #f)
-	     (topic 'subscriber (channel-message (channel-stream-name #t cname) _) #f))
-	 (if (and (ground? cname)
-		  (not (memf (ssh-channel-name=? cname) (connection-channels conn))))
-	     (transition (update-channel cname values conn)
-	       (channel-roles cname (lambda (inbound-stream outbound-stream)
-				      '())))
-	     (transition conn))]))))
+    (endpoint #:publisher (channel-message (channel-stream-name ? (channel-name #t ? ?)) ?)
+	      #:observer
+	      #:state conn
+	      #:conversation (channel-message (channel-stream-name #t cname) _)
+	      #:on-presence (respond-to-opened-outbound-channel conn cname))
+    (endpoint #:subscriber (channel-message (channel-stream-name ? (channel-name #t ? ?)) ?)
+	      #:observer
+	      #:state conn
+	      #:conversation (channel-message (channel-stream-name #f cname) _)
+	      #:on-presence (respond-to-opened-outbound-channel conn cname))))
 
 (define (handle-msg-channel-open packet message conn)
   (match-define (ssh-msg-channel-open channel-type*
@@ -730,15 +731,15 @@
   (transition (update-channel cname
 			      (lambda (e) (struct-copy ssh-channel e [remote-ref remote-ref]))
 			      conn)
-    (channel-roles cname
-		   (lambda (inbound-stream outbound-stream)
-		     (list (send-feedback
-			    (channel-message outbound-stream
-					     (channel-stream-config maximum-packet-size
-								    extra-request-data)))
-			   (send-feedback
-			    (channel-message outbound-stream
-					     (channel-stream-credit initial-window-size))))))))
+    (channel-endpoints cname
+		       (lambda (inbound-stream outbound-stream)
+			 (list (send-feedback
+				(channel-message outbound-stream
+						 (channel-stream-config maximum-packet-size
+									extra-request-data)))
+			       (send-feedback
+				(channel-message outbound-stream
+						 (channel-stream-credit initial-window-size))))))))
 
 (define (handle-msg-channel-open-confirmation packet message conn)
   (match-define (ssh-msg-channel-open-confirmation local-ref
@@ -859,46 +860,45 @@
 	  SSH_MSG_DEBUG handle-msg-debug
 	  SSH_MSG_KEXINIT handle-msg-kexinit))
 
-(define (ssh-session local-identification-string
+(define (ssh-session self-pid
+		     local-identification-string
 		     peer-identification-string
 		     application-boot
 		     session-role)
-  (boot-specification
-   (lambda (self-pid)
-     (transition (connection #f
-			     base-packet-dispatcher
-			     0
-			     (rekey-in-seconds-or-bytes -1 -1 0)
-			     #f
-			     '()
-			     (case session-role ((client) #f) ((server) #t))
-			     local-identification-string
-			     peer-identification-string
-			     #f
-			     application-boot)
+  (transition (connection #f
+			  base-packet-dispatcher
+			  0
+			  (rekey-in-seconds-or-bytes -1 -1 0)
+			  #f
+			  '()
+			  (case session-role ((client) #f) ((server) #t))
+			  local-identification-string
+			  peer-identification-string
+			  #f
+			  application-boot)
 
-       (role (topic-subscriber (timer-expired 'rekey-timer (wild)))
-	 #:state conn
-	 [(timer-expired 'rekey-timer now)
-	  (sequence-actions (transition conn)
-	    maybe-rekey)])
+    (endpoint #:subscriber (timer-expired 'rekey-timer (wild))
+	      #:state conn
+	      [(timer-expired 'rekey-timer now)
+	       (sequence-actions (transition conn)
+				 maybe-rekey)])
 
-       (role (topic-subscriber (outbound-byte-credit (wild)))
-	 #:state conn
-	 [(outbound-byte-credit amount)
-	  (sequence-actions (transition conn)
-	    (bump-total amount)
-	    maybe-rekey)])
+    (endpoint #:subscriber (outbound-byte-credit (wild))
+	      #:state conn
+	      [(outbound-byte-credit amount)
+	       (sequence-actions (transition conn)
+				 (bump-total amount)
+				 maybe-rekey)])
 
-       (role (topic-subscriber (inbound-packet (wild) (wild) (wild) (wild)))
-	 #:state conn
-	 [(inbound-packet sequence-number payload message transfer-size)
-	  (sequence-actions (transition conn)
-	    (lambda (conn)
-	      (if (connection-discard-next-packet? conn)
-		  (transition (struct-copy connection conn [discard-next-packet? #f]))
-		  (dispatch-packet sequence-number payload message conn)))
-	    (bump-total transfer-size)
-	    (send-message (inbound-credit 1))
-	    maybe-rekey)])))
-   connection?))
+    (endpoint #:subscriber (inbound-packet (wild) (wild) (wild) (wild))
+	      #:state conn
+	      [(inbound-packet sequence-number payload message transfer-size)
+	       (sequence-actions (transition conn)
+				 (lambda (conn)
+				   (if (connection-discard-next-packet? conn)
+				       (transition
+					   (struct-copy connection conn [discard-next-packet? #f]))
+				       (dispatch-packet sequence-number payload message conn)))
+				 (bump-total transfer-size)
+				 (send-message (inbound-credit 1))
+				 maybe-rekey)])))

ssh-transport.rkt

@@ -14,7 +14,7 @@
 (require "ssh-message-types.rkt")
 (require "ssh-exceptions.rkt")
 
-(require "os2-support.rkt")
+(require "marketplace-support.rkt")
 
 (provide (struct-out inbound-packet)
 	 (struct-out inbound-credit)
@@ -279,114 +279,122 @@
 
 (struct ssh-reader-state (mode config sequence-number remaining-credit) #:prefab)
 
-(define (ssh-reader new-connection-topic)
-  (define-values (cin cout in-topic out-topic) (topic->tcp-connection new-connection-topic))
+(define (ssh-reader new-conversation)
+  (match-define (tcp-channel remote-addr local-addr _) new-conversation)
   (define packet-size-limit (default-packet-limit))
 
   (define (issue-credit state)
     (match-define (ssh-reader-state _ (crypto-configuration _ desc _ _) _ message-credit) state)
     (when (positive? message-credit)
-      (at-meta-level (cin (tcp-credit (supported-cipher-block-size desc))))))
+      (at-meta-level
+       (send-feedback (tcp-channel remote-addr local-addr
+				   (tcp-credit (supported-cipher-block-size desc)))))))
 
   (transition (ssh-reader-state 'packet-header initial-crypto-configuration 0 0)
     (at-meta-level
-     (role in-topic
-       #:name 'socket-reader
-       #:state (and state
-		    (ssh-reader-state mode
-				      (crypto-configuration cipher
-							    cipher-description
-							    hmac
-							    hmac-description)
-				      sequence-number
-				      remaining-credit))
+     (endpoint #:subscriber (tcp-channel remote-addr local-addr ?)
+	       #:name 'socket-reader
+	       #:state (and state
+			    (ssh-reader-state mode
+					      (crypto-configuration cipher
+								    cipher-description
+								    hmac
+								    hmac-description)
+					      sequence-number
+					      remaining-credit))
        [(tcp-channel _ _ (? eof-object?))
 	(transition state (quit))]
        [(tcp-channel _ _ (? bytes? encrypted-packet))
-	(define block-size (supported-cipher-block-size cipher-description))
-	(define first-block-size block-size)
-	(define subsequent-block-size (if cipher block-size 1))
-	(define decryptor (if cipher cipher values))
+	(let ()
+	  (define block-size (supported-cipher-block-size cipher-description))
+	  (define first-block-size block-size)
+	  (define subsequent-block-size (if cipher block-size 1))
+	  (define decryptor (if cipher cipher values))
 
-	(define (check-hmac packet-length payload-length packet)
-	  (define computed-hmac-bytes (apply-hmac hmac sequence-number packet))
-	  (define mac-byte-count (bytes-length computed-hmac-bytes))
-	  (if (positive? mac-byte-count)
-	      (transition (struct-copy ssh-reader-state state
-				       [mode `(packet-hmac ,computed-hmac-bytes
-							   ,mac-byte-count
-							   ,packet-length
-							   ,payload-length
-							   ,packet)])
-		(at-meta-level (cin (tcp-credit mac-byte-count))))
-	      (finish-packet 0 packet-length payload-length packet)))
+	  (define (check-hmac packet-length payload-length packet)
+	    (define computed-hmac-bytes (apply-hmac hmac sequence-number packet))
+	    (define mac-byte-count (bytes-length computed-hmac-bytes))
+	    (if (positive? mac-byte-count)
+		(transition (struct-copy ssh-reader-state state
+			      [mode `(packet-hmac ,computed-hmac-bytes
+						  ,mac-byte-count
+						  ,packet-length
+						  ,payload-length
+						  ,packet)])
+		  (at-meta-level
+		   (send-feedback (tcp-channel remote-addr local-addr
+					       (tcp-credit mac-byte-count)))))
+		(finish-packet 0 packet-length payload-length packet)))
 
-	(define (finish-packet mac-byte-count packet-length payload-length packet)
-	  (define bytes-read (+ packet-length mac-byte-count))
-	  (define payload (subbytes packet 5 (+ 5 payload-length)))
-	  (define new-credit (- remaining-credit 1))
-	  (define new-state (struct-copy ssh-reader-state state
-					 [mode 'packet-header]
-					 [sequence-number (+ sequence-number 1)]
-					 [remaining-credit new-credit]))
-	  (transition new-state
-	    (issue-credit new-state)
-	    (send-message
-	     (inbound-packet sequence-number payload (ssh-message-decode payload) bytes-read))))
+	  (define (finish-packet mac-byte-count packet-length payload-length packet)
+	    (define bytes-read (+ packet-length mac-byte-count))
+	    (define payload (subbytes packet 5 (+ 5 payload-length)))
+	    (define new-credit (- remaining-credit 1))
+	    (define new-state (struct-copy ssh-reader-state state
+				[mode 'packet-header]
+				[sequence-number (+ sequence-number 1)]
+				[remaining-credit new-credit]))
+	    (transition new-state
+	      (issue-credit new-state)
+	      (send-message
+	       (inbound-packet sequence-number payload (ssh-message-decode payload) bytes-read))))
 
-	(match mode
-	  ['packet-header
-	   (define decrypted-packet (decryptor encrypted-packet))
-	   (define first-block decrypted-packet)
-	   (define packet-length (integer-bytes->integer first-block #f #t 0 4))
-	   (check-packet-length! packet-length packet-size-limit subsequent-block-size)
-	   (define padding-length (bytes-ref first-block 4))
-	   (define payload-length (- packet-length padding-length 1))
-	   (define amount-of-packet-in-first-block
-	     (- (bytes-length first-block) 4)) ;; not incl length
-	   (define remaining-to-read (- packet-length amount-of-packet-in-first-block))
+	  (match mode
+	    ['packet-header
+	     (define decrypted-packet (decryptor encrypted-packet))
+	     (define first-block decrypted-packet)
+	     (define packet-length (integer-bytes->integer first-block #f #t 0 4))
+	     (check-packet-length! packet-length packet-size-limit subsequent-block-size)
+	     (define padding-length (bytes-ref first-block 4))
+	     (define payload-length (- packet-length padding-length 1))
+	     (define amount-of-packet-in-first-block
+	       (- (bytes-length first-block) 4)) ;; not incl length
+	     (define remaining-to-read (- packet-length amount-of-packet-in-first-block))
 
-	   (if (positive? remaining-to-read)
-	       (transition (struct-copy ssh-reader-state state
-					[mode `(packet-body ,packet-length
-							    ,payload-length
-							    ,first-block)])
-		 (at-meta-level (cin (tcp-credit remaining-to-read))))
-	       (check-hmac packet-length payload-length first-block))]
+	     (if (positive? remaining-to-read)
+		 (transition (struct-copy ssh-reader-state state
+			       [mode `(packet-body ,packet-length
+						   ,payload-length
+						   ,first-block)])
+		   (at-meta-level
+		    (send-feedback (tcp-channel remote-addr local-addr
+						(tcp-credit remaining-to-read)))))
+		 (check-hmac packet-length payload-length first-block))]
 
-	  [`(packet-body ,packet-length ,payload-length ,first-block)
-	   (define decrypted-packet (decryptor encrypted-packet))
-	   (check-hmac packet-length payload-length (bytes-append first-block decrypted-packet))]
+	    [`(packet-body ,packet-length ,payload-length ,first-block)
+	     (define decrypted-packet (decryptor encrypted-packet))
+	     (check-hmac packet-length payload-length (bytes-append first-block decrypted-packet))]
 
-	  [`(packet-hmac ,computed-hmac-bytes
-			 ,mac-byte-count
-			 ,packet-length
-			 ,payload-length
-			 ,main-packet)
-	   (define received-hmac-bytes encrypted-packet) ;; not really encrypted!
-	   (if (equal? computed-hmac-bytes received-hmac-bytes)
-	       (finish-packet mac-byte-count packet-length payload-length main-packet)
-	       (disconnect-with-error/local-info `((expected-hmac ,computed-hmac-bytes)
-						   (actual-hmac ,received-hmac-bytes))
-						 SSH_DISCONNECT_MAC_ERROR
-						 "Corrupt MAC"))])]))
-    (role (topic-subscriber (inbound-credit (wild)))
-      #:state state
+	    [`(packet-hmac ,computed-hmac-bytes
+			   ,mac-byte-count
+			   ,packet-length
+			   ,payload-length
+			   ,main-packet)
+	     (define received-hmac-bytes encrypted-packet) ;; not really encrypted!
+	     (if (equal? computed-hmac-bytes received-hmac-bytes)
+		 (finish-packet mac-byte-count packet-length payload-length main-packet)
+		 (disconnect-with-error/local-info `((expected-hmac ,computed-hmac-bytes)
+						     (actual-hmac ,received-hmac-bytes))
+						   SSH_DISCONNECT_MAC_ERROR
+						   "Corrupt MAC"))]))]))
+    (endpoint #:subscriber (inbound-credit (wild))
+	      #:state state
       [(inbound-credit amount)
-       (define new-state (struct-copy ssh-reader-state state
-				      [remaining-credit
-				       (+ amount (ssh-reader-state-remaining-credit state))]))
-       (transition new-state
-	 (issue-credit new-state))])
-    (role (topic-subscriber (new-keys (wild)
-				      (wild)
-				      (wild) (wild)
-				      (wild) (wild)
-				      (wild) (wild)))
-      #:state state
+       (let ()
+	 (define new-state (struct-copy ssh-reader-state state
+			     [remaining-credit
+			      (+ amount (ssh-reader-state-remaining-credit state))]))
+	 (transition new-state
+	   (issue-credit new-state)))])
+    (endpoint #:subscriber (new-keys (wild)
+				     (wild)
+				     (wild) (wild)
+				     (wild) (wild)
+				     (wild) (wild))
+	      #:state state
       [(? new-keys? nk)
        (transition (struct-copy ssh-reader-state state [config (apply-negotiated-options nk #f)]))])
-    (role (topic-publisher (inbound-packet (wild) (wild) (wild) (wild))))))
+    (endpoint #:publisher (inbound-packet (wild) (wild) (wild) (wild)))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Encrypted Packet Output
@@ -394,49 +402,53 @@
 
 (struct ssh-writer-state (config sequence-number) #:prefab)
 
-(define (ssh-writer new-connection-topic)
-  (define-values (cin cout in-topic out-topic) (topic->tcp-connection new-connection-topic))
+(define (ssh-writer new-conversation)
+  (match-define (tcp-channel remote-addr local-addr _) new-conversation)
   (transition (ssh-writer-state initial-crypto-configuration 0)
-    (role (set (topic-subscriber (outbound-packet (wild)))
-	       (topic-publisher (outbound-byte-credit (wild))))
-      #:state (and state
-		   (ssh-writer-state (crypto-configuration cipher
-							   cipher-description
-							   hmac
-							   hmac-description)
-				     sequence-number))
+    (endpoint #:publisher (outbound-byte-credit (wild)))
+    (endpoint #:subscriber (outbound-packet (wild))
+	      #:state (and state
+			   (ssh-writer-state (crypto-configuration cipher
+								   cipher-description
+								   hmac
+								   hmac-description)
+					     sequence-number))
       [(outbound-packet message)
-       (define pad-block-size (supported-cipher-block-size cipher-description))
-       (define encryptor (if cipher cipher values))
-       (define payload (ssh-message-encode message))
-       ;; There must be at least 4 bytes of padding, and padding needs to
-       ;; make the packet length a multiple of pad-block-size.
-       (define unpadded-length (+ 4 ;; length of length
-				  1 ;; length of length-of-padding indicator
-				  (bit-string-byte-count payload)))
-       (define min-padded-length (+ unpadded-length 4))
-       (define padded-length (round-up min-padded-length pad-block-size))
-       (define padding-length (- padded-length unpadded-length))
-       (define packet-length (- padded-length 4)) ;; the packet length does *not* include itself!
-       (define packet (bit-string->bytes
-		       (bit-string (packet-length :: integer bits 32)
-				   (padding-length :: integer bits 8)
-				   (payload :: binary)
-				   ((random-bytes padding-length) :: binary))))
-       (define encrypted-packet (encryptor packet))
-       (define computed-hmac-bytes (apply-hmac hmac sequence-number packet))
-       (define mac-byte-count (bytes-length computed-hmac-bytes))
-       (transition (struct-copy ssh-writer-state state [sequence-number (+ sequence-number 1)])
-	 (at-meta-level (cout encrypted-packet))
-	 (when (positive? mac-byte-count)
-	   (at-meta-level (cout computed-hmac-bytes)))
-	 (send-message (outbound-byte-credit (+ (bytes-length encrypted-packet) mac-byte-count))))])
-    (role (topic-subscriber (new-keys (wild)
-				      (wild)
-				      (wild) (wild)
-				      (wild) (wild)
-				      (wild) (wild)))
-      #:state state
+       (let ()
+	 (define pad-block-size (supported-cipher-block-size cipher-description))
+	 (define encryptor (if cipher cipher values))
+	 (define payload (ssh-message-encode message))
+	 ;; There must be at least 4 bytes of padding, and padding needs to
+	 ;; make the packet length a multiple of pad-block-size.
+	 (define unpadded-length (+ 4 ;; length of length
+				    1 ;; length of length-of-padding indicator
+				    (bit-string-byte-count payload)))
+	 (define min-padded-length (+ unpadded-length 4))
+	 (define padded-length (round-up min-padded-length pad-block-size))
+	 (define padding-length (- padded-length unpadded-length))
+	 (define packet-length (- padded-length 4)) ;; the packet length does *not* include itself!
+	 (define packet (bit-string->bytes
+			 (bit-string (packet-length :: integer bits 32)
+				     (padding-length :: integer bits 8)
+				     (payload :: binary)
+				     ((random-bytes padding-length) :: binary))))
+	 (define encrypted-packet (encryptor packet))
+	 (define computed-hmac-bytes (apply-hmac hmac sequence-number packet))
+	 (define mac-byte-count (bytes-length computed-hmac-bytes))
+	 (transition (struct-copy ssh-writer-state state [sequence-number (+ sequence-number 1)])
+	   (at-meta-level
+	    (send-message (tcp-channel local-addr remote-addr encrypted-packet)))
+	   (when (positive? mac-byte-count)
+	     (at-meta-level
+	      (send-message (tcp-channel local-addr remote-addr computed-hmac-bytes))))
+	   (send-message
+	    (outbound-byte-credit (+ (bytes-length encrypted-packet) mac-byte-count)))))])
+    (endpoint #:subscriber (new-keys (wild)
+				     (wild)
+				     (wild) (wild)
+				     (wild) (wild)
+				     (wild) (wild))
+	      #:state state
       [(? new-keys? nk)
        (transition
 	   (struct-copy ssh-writer-state state [config (apply-negotiated-options nk #t)]))])))