a09ae29

deps.edn modified (view file)

@@ -10,5 +10,4 @@   io.netty/netty-buffer {:mvn/version "4.1.111.Final"}   org.clojure/core.async {:mvn/version "1.9.829-alpha2"}   org.clojure/data.json {:mvn/version "2.5.1"}-  io.github.tonsky/clojure-plus {:mvn/version "1.7.2"}-  aleph/aleph {:mvn/version "0.9.3"}}}+  io.github.tonsky/clojure-plus {:mvn/version "1.7.2"}}}

src/net/bowmap/core.clj modified (view file)

@@ -3,8 +3,9 @@    [net.bowmap.minecraft-server :as mc]))  (defn start-server! [& _args]-  (mc/start-server! {:port 25565})-  (println "Server started!"))+  (let [shutdown (mc/start-server! {:port 25565})]+    (println "Server started!")+    @shutdown))  (comment   (binding [*out* ((requiring-resolve 'clojure.java.io/writer) System/out)]

src/net/bowmap/minecraft_server.clj modified (view file)

@@ -78,11 +78,11 @@ ; TODO make reloadable, if port is already running, just swap it out? ; TODO this is networking, and so should be in net.bowmap.server.network (defn start-server! [{:keys! [port]}]-  (let [{::server/keys! [server client-chan]} (server/start-tcp-server! {:port port})]+  (let [{::server/keys! [client-chan shutdown]} (server/start-tcp-server! {:port port})]+    ; Loop over the instance?     (io-thread-      ; Loop over the instance?       (loop []         (when-some [{::server/keys! [packets-in packets-out]} (<!! client-chan)]           (connect! (client) packets-in packets-out)-          (recur)))))-  nil)+          (recur))))+    shutdown))

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

@@ -1,31 +1,11 @@ (ns net.bowmap.network.protocol.packet   (:require-   [dev.grahp.spec :as S]-   [net.bowmap.network.protocol.readers :as read])+   [dev.grahp.spec :as S])   (:import-   [io.netty.buffer ByteBuf Unpooled]))+   [io.netty.buffer ByteBuf]))  ; The serialized form of a message, called a packet by everyone else :p ; See https://minecraft.wiki/w/Java_Edition_protocol/Packets  ; TODO rename to a frame, packet is still an overloaded term (S/def ::packet #(instance? ByteBuf %))--(defn packet-framer-xf-  "Transducer that frames bytes from incoming byte arrays and outputs ByteBufs of-   framed serialized serverbound packets."-  [rf]-  (let [buf (Unpooled/buffer)]-    (fn-      ([] (rf))-      ([res] (rf res))-      ([res input]-       (ByteBuf/.writeBytes buf input)-       (loop [res res]-         (if (read/can-read-packet? buf)-           (let [len (read/->varint buf)-                 res' (rf res (ByteBuf/.copy (ByteBuf/.readRetainedSlice buf len)))]-             (ByteBuf/.discardReadBytes buf)-             (recur res'))-           res))))))-

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

@@ -1,25 +1,61 @@ (ns net.bowmap.network.server   (:require-   [aleph.tcp :as tcp]-   [clojure.core.async :refer [chan put!]]-   [manifold.stream :as s]-   [net.bowmap.network.protocol.packet :as pk])+   [clojure.core.async :refer [<!! >!! chan close! io-thread]]+   [net.bowmap.network.stream :as stream])   (:import-   [io.netty.bootstrap ServerBootstrap]-   [io.netty.channel ChannelOption]))+   [java.io Closeable]+   [java.net ServerSocket Socket])) -(defn- handle-client [client-chan stream info]-  (let [packets-in (chan 1 pk/packet-framer-xf)+#_(defn- handle-client [client-chan stream info]+    (let [packets-in (chan 1 pk/packet-framer-xf)+          packets-out (chan)]+      (s/connect stream packets-in)+      (s/connect packets-out stream)+      (put! client-chan {::packets-in packets-in+                         ::packets-out packets-out})))++(defn new-client [socket]+  (let [packets-in (chan)         packets-out (chan)]-    (s/connect stream packets-in)-    (s/connect packets-out stream)-    (put! client-chan {::packets-in packets-in-                       ::packets-out packets-out})))+    (io-thread+      (loop []+        (try (let [frame (stream/read-frame!! (Socket/.getInputStream socket))]+               (>!! packets-in frame))+             (catch Exception _+               ; Disconnect packet?+               (Socket/.close socket)))+        (recur)))++    (io-thread+      (loop []+        (let [frame (<!! packets-out)]+          (stream/write-frame!! frame (Socket/.getOutputStream socket)))+        (recur)))+    {::packets-in packets-in+     ::packets-out packets-out}))++(defn start-server-process!+  "Starts a TCP server at the given port. Returns a channel of clients as they connect.+   When the server closes, client channel will be closed, and vice versa."+  [port]+  (let [server (ServerSocket. port)+        clients (chan 64)+        shutdown (promise)]+    (io-thread+      (try+        (loop []+          ; TODO TCP_NODELAY+          (let [socket (ServerSocket/.accept server)]+            (if (>!! clients (new-client socket))+              (recur)+              (do (Closeable/.close socket)+                  (deliver shutdown :shutdown)))))+        (catch Throwable t+          (close! clients)+          (deliver shutdown :shutdown)+          (throw t))))+    {::client-chan clients+     ::shutdown shutdown}))  (defn start-tcp-server! [& {:keys! [port]}]-  (let [client-chan (chan 64)]-    {::server (tcp/start-server (fn [stream info]-                                  (handle-client client-chan stream info))-                                {:port port-                                 :bootstrap-transform #(ServerBootstrap/.childOption % ChannelOption/TCP_NODELAY true)})-     ::client-chan client-chan}))+  (start-server-process! port))

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

@@ -0,0 +1,48 @@+(ns net.bowmap.network.stream+  (:require+   [dev.grahp.error :refer [raise]]+   [net.bowmap.network.protocol.types :as t])+  (:import+   [io.netty.buffer ByteBuf Unpooled]+   [java.io InputStream OutputStream]))++; Reading and writing frames to and from InputStreams and OutputStreams++(defn read-byte!!+  "Either returns an unsigned byte from 'in', or throws"+  [input-stream]+  (let [x (InputStream/.read input-stream)]+    (if (= -1 x)+      (raise "EOF while reading from InputStream")+      x)))++(defn read-varint!!+  "Either returns a varint from 'in', or throws"+  [input-stream]+  (loop [num-read 0+         result 0]+    (if (>= num-read t/max-varint-size)+      (raise "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!! [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 (Exception. "EOF while reading frame"))+            (recur (+ offset r))))))+    (Unpooled/wrappedBuffer arr)))++(defn write-frame!! [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)))