jeapostrophe / zeromq Goto Github PK

#lang racket
(require net/zmq)
(define (main myself peers)
  (printf "Hello, I am ~a\n" myself)
  (define ctxt (context 1))
  ; State Back-End
  (define statebe (socket ctxt 'PUB))
  ; State Front-End
  (define statefe (socket ctxt 'SUB))
  (set-socket-option! statefe 'SUBSCRIBE #"")  
  (define bind-address (format "ipc://~a-state.ipc" myself))
  (socket-bind! statebe bind-address)
  (for ([p (in-list peers)])
    (socket-connect! statefe (format "ipc://~a-state.ipc" p)))  
  (define poller
    (vector (make-poll-item statefe 0 'POLLIN empty)))
  (let loop ()
    (printf "before polling\n")
    (poll! poller 1000)
    (printf "after polling\n")
    (define revents (poll-item-revents (vector-ref poller 0)))
    (if (equal? revents '(POLLIN))
        (printf "Received: ~a" (socket-recv! statefe))
        (socket-send! statebe
                      (string->bytes/utf-8
                       (format "~a ~a" bind-address (random 10)))))
    (loop))
  (context-close! ctxt))
(command-line #:program "peering1"
              #:args (myself . peers)
              (main myself peers))

(define (run-zmq-server port)
  (call-with-context (λ (ctx)
    (call-with-socket ctx 'PULL (λ (socket)
      (socket-bind! socket (format "tcp://127.0.0.1:~a" port))
      (display (format "Started mock CE Service on port ~a.\n" port))

      (let loop ()
        (let* ([dto-bytes (socket-recv! socket)] ; <--- The error is getting thrown here.
               [req (_bytes->dto dto-bytes)]
               [msg-type (hash-ref req 'messageType #f)])
          (cond
            [(equal? msg-type 'CLOSE)
              (display "Closing mock CE Service.\n")]
            [else
              (display (format "Received message:\n~a\n\n" msg-type))
              (loop)]))))))))