 ;; ============================================================================
 ;;
 
+(def cancel-ch (atom nil))
+
 (def service (reify system/Service
                (start! [this config]
-                 (reset! run true)
-                 (reset! video-source (if (:video-sim config)
-                                        (InetAddress/getByName nil)
-                                        comm/drone-ip))
-                 (task/submit :video video-streaming-task)
+                 (let [ch (async/chan)]
+                   (thread* (video-streaming-task ch))
+                   (reset! cancel-ch ch))
                  config)
                (stop! [this]
-                 (reset! run false)
-                 (task/cancel :video))))
+                 (when-let [ch @cancel-ch]
+                   (async/close! ch)
+                   (reset! cancel-ch nil)))))
 
 ; ffmpeg -f h264 -an -i capture.h264 stream.m4v

        (set! (.data_base packet) (ba->ia buffer ibuffer size))
        (set! (.data_offset packet) 0)
        (.avcodec_decode_video2 context frame got-picture? packet)
        (when (zero? (first got-picture?)) (throw (java.io.IOException. "Could not decode frame")))
        (let [picture         (.displayPicture (.priv_data context))
              width           (.imageWidth picture)
              height          (.imageHeight picture)
              picture-buffer  (int-array (* width height))
              image           (BufferedImage. width height BufferedImage/TYPE_INT_RGB)]
          (FrameUtils/YUV2RGB picture picture-buffer)
          (.setRGB image 0 0 width height picture-buffer 0 width)
          image)))))

(def video-source (atom nil))

(defn open-socket ^Socket []
  (println "Connecting to video stream:" (str @video-source))
  (doto (Socket.)
    (.setSoTimeout 2000)
    (.connect (InetSocketAddress. ^InetAddress @video-source 5555))))

(def run (atom false))

(defn send-init-video [^Socket socket]
  (doto (.getOutputStream socket)
    (.write (byte-array (map bin/ubyte [1 0 0 0])))
    (.flush)))

(defn video-streaming-task []
  (try
    (while @run
      (let [socket      (open-socket)
            reader      (make-reader (-> socket .getInputStream BufferedInputStream.))
            decoder     (make-decoder)]
        (try
          (send-init-video socket)
          (while @run
            (->> (reader) (decoder) (swap! world assoc :image)))
          (catch java.io.IOException e
            (println "I/O error:" e ": reconnecting...")
            (Thread/sleep 1000))
          (finally
            (try (.close socket) (catch Exception _))))))
    (catch Throwable e
      (println "exception while processing video stream" e)
      (.printStackTrace e))))

;;
;; ============================================================================

(ns sormilla.video
  (:import [com.twilight.h264.decoder AVFrame AVPacket H264Decoder MpegEncContext]
           [com.twilight.h264.player FrameUtils]
           [java.awt.image BufferedImage]
           [java.io InputStream BufferedInputStream]
           [java.net Socket InetAddress])
  (:require [clojure.java.io :as io]
            [clojure.core.async :as async :refer [>!! <!! alts!! go]]
            [metosin.system :as system]
            [sormilla.world :refer [world]]
            [sormilla.bin :as bin :refer [ba->ia]]
            [sormilla.io-utils :as io-utils]
            [sormilla.drone :as drone]
            [sormilla.drone-comm :as comm]))

(set! *warn-on-reflection* true)

(defn make-frame-reader [^InputStream in]
  (fn []
    (try
      (let [header       (io-utils/read-fully in (byte-array 12) 0 12)
            signature    (bin/get-int header 0)
            header-size  (bin/get-short header 6)
            payload-size (bin/get-int header 8)
            payload      (byte-array payload-size)]
        (when-not (= signature 0x45566150) (throw (java.io.IOException. (format "out of sync (0x%08X)" signature))))
        (io-utils/skip-fully in (- header-size 12))
        (io-utils/read-fully in payload 0 payload-size))
      (catch java.io.IOException _
        nil)
      (catch InterruptedException _
        nil))))

(defn make-frame-decoder []
  (let [codec           (H264Decoder.)
        context         (MpegEncContext/avcodec_alloc_context)
        packet          (AVPacket.)
