core.clj

Branches > main > 8605065 > files / src / net / bowmap / core.clj

(ns net.bowmap.core
  (:require
   [manifold.deferred :as d]
   [manifold.stream :as s]
   [net.bowmap.client.state :as state]
   [net.bowmap.packet :as pk]
   [net.bowmap.packet-handling :as handle]
   [net.bowmap.packet.client :as cp]
   [net.bowmap.packet.read :as read]
   [net.bowmap.packet.server :as sp]
   [net.bowmap.server :as server])
  (:import
   [io.netty.buffer ByteBuf Unpooled]))

(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/read-varint! buf)
                 res' (rf res (ByteBuf/.copy (ByteBuf/.readRetainedSlice buf len)))]
             (ByteBuf/.discardReadBytes buf)
             (recur res'))
           res))))))

(defn handle-packet
  "Handles one ByteBuf of a framed serialized serverbound packet, and returns a
   map of :client and :packets."
  [client packet-buf]
  (let [pk (sp/decode-packet packet-buf (::state/state client))]
    (when (= (::pk/id pk) ::sp/finish-config)
      ; Because the return map is arbitrary, this can definitely be idk literally anywhere else lmao
      ; These 3 functions definitely need to change although I don't know what that looks like.
      (s/connect (s/periodically 5000 0 (fn []
                                          {::pk/id ::cp/keep-alive
                                           :id (long (* (rand) Long/MAX_VALUE))}))
                 (::packet-stream client)
                 {:downstream? false}))
    (handle/response client pk)))

(defn handle-message
  "Handles either a serverbound or clientbound packet, returning the new client
   state, and sending either the clientbound packets it was given, or the
   packets that handling the serverbound packet it was given returned.
   Should only be used synchronously as serialization depends on client state!"
  [client msg]
  (case (:type msg)
    :server
    (let [{:keys [client response]} (handle-packet client (:bytes msg))]
      (reduce (fn [client pk]
                (handle-message client {:type :client :pk pk}))
              client response))

    :client
    (let [pk (cp/encode-packet (:pk msg) (::state/state client))]
      @(s/put! (::stream client) pk)
      client)))

(defn client-handler [stream _]
  (let [client-packet-stream (s/stream)
        client {::state/state ::state/handshaking
                ::stream stream
                ::packet-stream client-packet-stream}
        message-stream (s/stream)]

    (-> (s/transform
          (comp
            packet-framer-xf
            (map (fn [buf] {:type :server
                            :bytes buf})))
          stream)
        (s/connect message-stream {:upstream? false}))

    (-> (s/transform
          (map (fn [pk] {:type :client
                         :pk pk}))
          client-packet-stream)
        (s/connect message-stream {:upstream? false}))

    (d/catch (s/reduce #'handle-message client message-stream)
      (fn [e]
        (println "Exception in client packets:" e)))))

(defn start-server []
  (server/start-server! {:handler #'client-handler
                         :port 25565}))

(defn -main [& _args]
  (start-server)
  (println "Server Started"))

(comment
  (require 'dev.grahp.bytebuf)
  (def server (start-server))
  (.close server))