connection.clj

Bowmap · bowmap.git · Tickets

Branches > main > af0a324 > files / src / net / bowmap / server / network / connection.clj

(ns net.bowmap.server.network.connection
  (:require
   [clojure.core.async :refer [io-thread]]
   [net.bowmap.server.network.streams :as streams])
  (:import
   [java.io IOException]
   [java.net Socket]
   [java.util.concurrent LinkedBlockingQueue]))

(defprotocol Connection
  (send-frame! [conn frame]
    "Sends 'frame' to the connection asynchronously. Returns immediately,
     or throws if the connection is closed.")
  (receive-frame!! [conn]
    "Blocks until a frame can be read from 'conn'.
     Throws after frames can no longer be read from the connection.")
  (close! [conn]
    "Closes the connection, causing sends and receives to fail."))

(defn connection
  "Creates a connection from a given java.net.Socket."
  [socket]
  (let [input-stream (Socket/.getInputStream socket)
        output-stream (Socket/.getOutputStream socket)
        send-queue (LinkedBlockingQueue.)]
    (Socket/.setTcpNoDelay socket true)
    (io-thread
      (try
        (loop []
          (let [frame (.take send-queue)]
            (streams/write-frame!! frame output-stream)
            (recur)))
        (catch IOException _)
        (finally
          (Socket/.close socket))))
    (reify Connection
      (close! [_]
        (Socket/.close socket))
      (send-frame! [_ frame]
        (if (Socket/.isClosed socket)
          (throw (ex-info "Tried to send a frame to a closed connection."
                          {:frame frame}))
          (.put send-queue frame)))
      (receive-frame!! [_]
        (if (Socket/.isClosed socket)
          (throw (Exception. "Connection closed."))
          (try (streams/read-frame!! input-stream)
               (catch Exception e
                 (Socket/.close socket)
                 (throw e))))))))