streams.clj

Bowmap · bowmap.git · Tickets

Branches > main > 81fab7e > files / src / net / bowmap / server / network / streams.clj

(ns net.bowmap.server.network.streams
  "Functions for reading and writing frames to Input and OutputStreams"
  (:require
   [net.bowmap.server.protocol.frame :as frame]
   [net.bowmap.server.protocol.types :refer [max-varint-size]])
  (:import
   [java.io IOException InputStream OutputStream]))

(defn read-byte!!
  "Blocks and reads an unsigned byte (int) from an InputStream.
   Throws if EOF is reached."
  [input-stream]
  (let [x (InputStream/.read input-stream)]
    (if (= x -1)
      (throw (IOException. "EOF while reading from InputStream"))
      x)))

(defn read-varint!!
  "Blocks and reads a varint from an InputStream.
   Throws if EOF is reached, or if the encoded varint is invalid."
  [input-stream]
  (loop [num-read 0
         result 0]
    (if (>= num-read max-varint-size)
      (throw (IOException. "VarInt too long"))
      (let [b (read-byte!! input-stream)
            result' (bit-or result (bit-shift-left (bit-and b 0x7F) (* 7 num-read)))]
        (if (zero? (bit-and b 0x80))
          result'
          (recur (inc num-read) result'))))))

(defn read-frame!!
  "Blocks and reads a frame from an InputStream.
   Throws if EOF is reached, or if the encoded frame is invalid."
  [input-stream]
  (let [len (read-varint!! input-stream)
        arr (byte-array len)]
    (loop [offset 0]
      (when (< offset len)
        (let [r (InputStream/.read input-stream arr offset (- len offset))]
          (if (= r -1)
            (throw (IOException. "EOF while reading frame"))
            (recur (+ offset r))))))
    (frame/frame arr)))

(defn write-varint!!
  "Blocks and writes a varint to an OutputStream."
  [output-stream n]
  (loop [n n]
    (let [b (bit-and n 0x7F)
          nxt (unsigned-bit-shift-right n 7)]
      (if (zero? nxt)
        (OutputStream/.write output-stream b)
        (do (OutputStream/.write output-stream (bit-or b 0x80))
            (recur nxt))))))

(defn write-frame!!
  "Blocks and writes a frame to an OutputStream.
   Throws if the frame cannot be written."
  [frame output-stream]
  (frame/write! frame
    (fn [array length]
        (write-varint!! output-stream length)
        (OutputStream/.write output-stream array 0 length)
        (OutputStream/.flush output-stream))))