core.clj

Branches > client-handling-rewrite > 30ddf0a > 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]
  (let [{:keys [stream val]} msg]
    (case stream
      :serverbound-packets
      (let [{:keys [client response]} (handle-packet client val)]
        (reduce (fn [client pk]
                  (handle-message client {:stream :clientbound-packets :val pk}))
                client response))

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

(defn feed
  "Sends all messages from the source to the sink with a transducer."
  [source xform sink]
  (let [stream (s/stream 1 xform)]
    (s/connect source stream)
    (s/connect stream sink {:upstream? false})))

(defn alt-onto
  "Takes a map of keywords to streams and returns a stream with vals {:stream keyword :val val-from-stream}."
  [chans]
  (let [out-stream (s/stream)]
    (doseq [[kw s] chans]
      (feed s
            (map (fn [x] {:stream kw :val x}))
            out-stream))
    out-stream))

(defn client-handler [stream _]
  (let [client-packet-stream (s/stream)
        client {::state/state ::state/handshaking
                ::stream stream
                ::packet-stream client-packet-stream}]
    (-> (s/reduce #'handle-message client (alt-onto {:serverbound-packets (s/transform packet-framer-xf stream)
                                                     :clientbound-packets client-packet-stream}))
        (d/catch (fn [e]
                   (println "Exception in client packets:\n" 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))