(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))))))