(in-package :fwoar.lw-curator)

(defun o->s (jo)
  (map 'string 'code-char
       (lw-ji:primitive-array-to-lisp-array jo)))

(defun al->lisp (al)
  (lw-ji:map-java-object-array 'identity
                               (fwoar.lw-curator.ji.al:.toarray al)
                               :collect t))

(defun client (host)
  (fwoar.lw-curator.ji.cff:.newclient host
                                      (fwoar.lw-curator.ji.bebr:.new 100 5000 10)))

(defun call-with-client (client lambda)
  (fwoar.lw-curator.ji.framework:.start client)
  (unwind-protect (funcall lambda client)
    (fwoar.lw-curator.ji.framework:.close client)))

(defmacro with-client ((sym host) &body body)
  `(let ((,sym (client ,host)))
     (call-with-client ,sym
                       (lambda (,sym)
                         ,@body))))

(defun get-consumers (client)
  (al->lisp
   (fwoar.lw-curator.ji.gcb:.forpath
    (fwoar.lw-curator.ji.framework:.getchildren client)
    "/consumers")))

(defun get-consumer-topics (client consumer)
  (al->lisp
   (fwoar.lw-curator.ji.gcb:.forpath
    (fwoar.lw-curator.ji.framework:.getchildren client)
    (format nil "/consumers/~a/offsets" consumer))))

(defun get-consumer-partitions (client consumer topic)
  (al->lisp
   (fwoar.lw-curator.ji.gcb:.forpath
    (fwoar.lw-curator.ji.framework:.getchildren client)
    (format nil "/consumers/~a/offsets/~a" consumer topic))))

(defun get-consumer-owner (client consumer topic partition)
  (o->s
   (fwoar.lw-curator.ji.gdb:.forpath
    (fwoar.lw-curator.ji.framework:.getdata client)
    (format nil "/consumers/~a/owners/~a/~a" consumer topic partition))))

(defun get-consumer-offset (client consumer topic partition)
  (nth-value 0
             (parse-integer
              (o->s
               (fwoar.lw-curator.ji.gdb:.forpath
                (fwoar.lw-curator.ji.framework:.getdata client)
                (format nil "/consumers/~a/offsets/~a/~a" consumer topic partition))))))