58deb69

deps.edn modified (view file)

@@ -4,5 +4,6 @@  :deps  {org.clojure/clojure {:mvn/version "1.13.0-alpha6"}   dev.weavejester/medley {:mvn/version "1.10.0"}+  org.clojure/core.async {:mvn/version "1.9.865"}   org.clojure/data.json {:mvn/version "2.5.2"}   io.netty/netty-buffer {:mvn/version "4.1.111.Final"}}}

src/net/bowmap/network/connection.clj added (view file)

@@ -0,0 +1,39 @@+(ns net.bowmap.network.connection+  "A connection to the Minecraft server."+  (:require+   [clojure.core.async :refer [<!! >!! chan close! io-thread]]+   [net.bowmap.network.streams :as streams])+  (:import+   [java.io IOException]+   [java.net Socket]))++(defn connection+  "Returns a connection from a given java.net.Socket.+   Connection has keys [::clientbound-frames ::serverbound-frames], both+   channels of clientbound and serverbound frames respectively."+  [socket]+  (let [clientbound-frames (chan 128)+        serverbound-frames (chan 128)+        input-stream (Socket/.getInputStream socket)+        output-stream (Socket/.getOutputStream socket)]+    (io-thread+      (try+        (loop []+          (when (>!! serverbound-frames (streams/read-frame!! input-stream))+            (recur)))+        (catch IOException _)+        (finally+          (close! serverbound-frames)+          (Socket/.close socket))))+    (io-thread+      (try+        (loop []+          (when-some [frame (<!! clientbound-frames)]+            (streams/write-frame!! frame output-stream)+            (recur)))+        (catch IOException _)+        (finally+          (close! clientbound-frames)+          (Socket/.close socket))))+    {::clientbound-frames clientbound-frames+     ::serverbound-frames serverbound-frames}))

src/net/bowmap/network/socket.clj modified (view file)

@@ -1,71 +1,24 @@ (ns net.bowmap.network.socket-  "Functions for reading frames to and from java.net.Sockets."   (:require-   [net.bowmap.network.frame.types :refer [max-varint-size]])+   [clojure.core.async :refer [>!! chan close! io-thread]]+   [net.bowmap.network.connection :as connection])   (:import-   [io.netty.buffer ByteBuf Unpooled]-   [java.io-    EOFException-    IOException-    InputStream-    OutputStream]-   [java.net Socket]))+   [java.net ServerSocket])) -(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 (EOFException. "EOF while reading from InputStream"))-      x)))+(def max-server-backlog 50) -(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 (EOFException. "EOF while reading frame"))-            (recur (+ offset r))))))-    (Unpooled/wrappedBuffer arr)))--(defn write-frame!!-  "Blocks and writes a frame to an OutputStream.-   Throws if the frame cannot be written."-  [frame output-stream]-  (let [len (ByteBuf/.readableBytes frame)-        arr (byte-array len)]-    (ByteBuf/.readBytes frame arr)-    (OutputStream/.write output-stream arr)-    (OutputStream/.flush output-stream)))--(defn receive-frame!!-  "Blocks and reads a frame from socket's InputStream.-   Throws if EOF is reached, or if the encoded frame is invalid."-  [socket]-  (read-frame!! (Socket/.getInputStream socket)))--(defn send-frame!!-  "Blocks and writes a frame to socket's OutputStream.-   Throws if the frame cannot be written."-  [frame socket]-  (write-frame!! frame (Socket/.getOutputStream socket)))+(defn server-socket+  "Starts a server socket at the supplied port and optional bind address.+   Returns a channel of connections that is coupled to the server's lifecycle."+  [& {:keys! [port] :keys [bind-addr]}]+  (let [server-socket (ServerSocket. port max-server-backlog bind-addr)+        connection-chan (chan max-server-backlog)]+    (io-thread+      (try+        (loop []+          (when (>!! connection-chan (connection/connection (ServerSocket/.accept server-socket)))+            (recur)))+        (finally+          (close! connection-chan)+          (ServerSocket/.close server-socket))))+    connection-chan))