async channels with udp

still-horsin.clj

(ns udp-ping.core
  (:require [clojure.core.async :refer [>! <! chan go close! thread]])
  (:import (java.net DatagramPacket InetAddress MulticastSocket)))

(def port 10001)
(def group (InetAddress/getByName "224.1.0.0"))
(def buffer-size 1000)

(defprotocol Socket
  (send! [this msg])
  (receive! [this])
  (close [this]))

(defrecord Multicast [socket]
  Socket
  (send! [_ msg] (.send socket (DatagramPacket. (.getBytes msg) (.length msg) group port)))
  (receive! [_]
    (let [recv (DatagramPacket. (byte-array buffer-size) buffer-size)]
      (.receive socket recv)
      (String. (.getData recv))))
  (close [_] (.close socket)))

(defprotocol Client
  (out [this msg])
  (in [this]))

(defrecord AsyncClient [in out]
  Client
  (out [_ msg] (go (>! out msg)))
  (in [_] (go (<! in))))

(defn multicast-socket []
  (let [socket (MulticastSocket. port)]
    (.joinGroup socket group)
    (->Multicast socket)))

(defn start []
  (let [in (chan)
        out (chan)
        client (->AsyncClient in out)
        socket (multicast-socket)]
        (go
         (while true
             (send! socket (<! out))))
        client))

(defn broadcast [msg] (send! (multicast-socket) msg))
(defn read [] (receive! (multicast-socket)))