//设置package包名称以及导入依赖的类
package com.github.ybr.zk.client

import akka.NotUsed
import akka.stream.Materializer
import akka.stream.scaladsl.Source

import org.apache.zookeeper._
import org.apache.zookeeper.data.{ACL, Stat}

import scala.concurrent._

trait ZooKeeperClient {
  def close(): Unit

  def create[A](path: String, data: Array[Byte], acl: List[ACL], createMode: CreateMode, ctx: Option[A])(implicit ec: ExecutionContext): Future[CreateResponse[A]]
  def delete[A](path: String, version: Int, ctx: Option[A])(implicit ec: ExecutionContext): Future[VoidResponse[A]] 

  def exists(path: String)(implicit ec: ExecutionContext): Future[StatResponse[Nothing]]
  def watchExists[A](path: String, ctx: Option[A])(implicit ec: ExecutionContext): Future[(StatResponse[A], Source[(StatResponse[A], WatchedEvent), NotUsed])]

  def getACL[A](path: String, stat: Stat, ctx: Option[A])(implicit ec: ExecutionContext): Future[ACLResponse[A]]
  def setACL[A](path: String, acl: List[ACL], version: Int, ctx: Option[A])(implicit ec: ExecutionContext): Future[StatResponse[A]]

  def getChildren[A](path: String, ctx: Option[A])(implicit ec: ExecutionContext): Future[ChildrenResponse[A]]
  def watchChildren[A](path: String, ctx: Option[A])(implicit ec: ExecutionContext): Future[(ChildrenResponse[A], Source[(ChildrenResponse[A], WatchedEvent), NotUsed])]

  def getData(path: String)(implicit ec: ExecutionContext): Future[DataResponse[Nothing]]
  def setData[A](path: String, data: Array[Byte], version: Int, ctx: Option[A])(implicit ec: ExecutionContext): Future[StatResponse[A]]
  def watchData[A](path: String, ctx: Option[A])(implicit ec: ExecutionContext): Future[(DataResponse[A], Source[(DataResponse[A], WatchedEvent), NotUsed])]

  def sessionId(): Long
  def sessionTimeout(): Int
}

object ZooKeeperClient {
  def native(hosts: String, sessionTimeout: Int, canBeReadOnly: Boolean, watcher: Option[Watcher] = None)(implicit mat: Materializer) = {
    val zk = new ZooKeeper(hosts, sessionTimeout, watcher.getOrElse(null), canBeReadOnly)
    new NativeZookeeperClient(zk)
  }
}

case class ACLResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A], acl: List[ACL], stat: Stat)
case class ChildrenResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A], children: List[String], stat: Stat)
case class CreateResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A], name: String)
case class DataResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A], data: Array[Byte], stat: Stat)
case class VoidResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A])
case class StatResponse[A](rc: KeeperException.Code, path: String, ctx: Option[A], stat: Stat) 

//设置package包名称以及导入依赖的类
package com.quantifind.kafka.core

import com.quantifind.utils.ZkUtilsWrapper
import org.I0Itec.zkclient.ZkClient
import org.apache.zookeeper.data.Stat
import org.mockito.Matchers._
import org.mockito.Mockito._
import org.mockito.{Matchers => MockitoMatchers, Mockito}
import org.scalatest._

class StormOffsetGetterSpec extends FlatSpec with ShouldMatchers {

  trait Fixture {

    val mockedZkClient = Mockito.mock(classOf[ZkClient])
    val zkOffsetBase = "/stormconsumers"
    val mockedZkUtil =  Mockito.mock(classOf[ZkUtilsWrapper])

    val offsetGetter = new StormOffsetGetter(mockedZkClient, zkOffsetBase, mockedZkUtil)
  }

  "StormOffsetGetter" should "be able to extract topic from persisted spout state" in new Fixture {

    val testGroup = "testgroup"
    val testTopic = "testtopic"
    val spoutState = s"""{
                        "broker": {
                            "host": "kafka.sample.net",
                            "port": 9092
                        },
                        "offset": 4285,
                        "partition": 1,
                        "topic": "${testTopic}",
                        "topology": {
                            "id": "fce905ff-25e0 -409e-bc3a-d855f 787d13b",
                            "name": "Test Topology"
                        }
                       }"""
    val ret = (spoutState, Mockito.mock(classOf[Stat]))
    when(mockedZkUtil.readData(MockitoMatchers.eq(mockedZkClient), anyString)).thenReturn(ret)

    val topics = offsetGetter.getTopicList(testGroup)
    
    topics.size shouldBe 1
    topics(0) shouldBe testTopic
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.zk

import java.util.{List => JList}

import scala.collection.JavaConverters._

import org.apache.zookeeper.data.Stat
import org.apache.zookeeper.{AsyncCallback, KeeperException}

import com.twitter.util.{Promise, Return, Throw}


  protected def process(rc: Int, path: String)(result: => T) {
    KeeperException.Code.get(rc) match {
      case KeeperException.Code.OK => updateIfEmpty(Return(result))
      case code => updateIfEmpty(Throw(KeeperException.create(code, path)))
    }
  }
}

class StringCallbackPromise extends AsyncCallbackPromise[String] with AsyncCallback.StringCallback {
  def processResult(rc: Int, path: String, ctx: AnyRef, name: String) {
    process(rc, path) { name }
  }
}

class UnitCallbackPromise extends AsyncCallbackPromise[Unit] with AsyncCallback.VoidCallback {
  def processResult(rc: Int, path: String, ctx: AnyRef) {
    process(rc, path) { Unit }
  }
}

class ExistsCallbackPromise(znode: ZNode) extends AsyncCallbackPromise[ZNode.Exists]
    with AsyncCallback.StatCallback {
  def processResult(rc: Int, path: String, ctx: AnyRef, stat: Stat) {
    process(rc, path) {
      znode(stat)
    }
  }
}

class ChildrenCallbackPromise(znode: ZNode) extends AsyncCallbackPromise[ZNode.Children]
    with AsyncCallback.Children2Callback {
  def processResult(rc: Int, path: String, ctx: AnyRef, children: JList[String], stat: Stat) {
    process(rc, path) {
      znode(stat, children.asScala.toSeq)
    }
  }
}

class DataCallbackPromise(znode: ZNode) extends AsyncCallbackPromise[ZNode.Data]
    with AsyncCallback.DataCallback {
  def processResult(rc: Int, path: String, ctx: AnyRef, bytes: Array[Byte], stat: Stat) {
    process(rc, path) {
      znode(stat, bytes)
    }
  }
}