//设置package包名称以及导入依赖的类
package com.twitter.finagle.serverset2.client.apache

import com.twitter.finagle.serverset2.client.{EventDeliveryThread, EventStats, WatchState}
import com.twitter.finagle.stats.{NullStatsReceiver, StatsReceiver}
import com.twitter.util.Var
import org.apache.zookeeper.{Watcher, WatchedEvent}

private[serverset2] class ApacheWatcher(statsIn: StatsReceiver = NullStatsReceiver)
    extends Watcher with EventStats {
  protected val stats = statsIn
  val state = Var[WatchState](WatchState.Pending)
  def process(event: WatchedEvent) = {
    event.getType match {
      case Watcher.Event.EventType.None =>
        EventDeliveryThread.offer(
          state,
          WatchState.SessionState(ApacheSessionState(event.getState)))
      case e =>
        EventDeliveryThread.offer(
          state,
          WatchState.Determined(EventFilter(ApacheNodeEvent(e))))
    }
  }
} 

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

import com.twitter.common.quantity.{Amount, Time => CommonTime}
import com.twitter.common.zookeeper.{ZooKeeperClient, ZooKeeperUtils}
import com.twitter.concurrent.{Offer, Broker, AsyncMutex}
import com.twitter.conversions.common.quantity._
import com.twitter.finagle.addr.StabilizingAddr.State._
import com.twitter.finagle.util.InetSocketAddressUtil
import com.twitter.util.Duration
import java.net.InetSocketAddress
import org.apache.zookeeper.Watcher.Event.KeeperState
import org.apache.zookeeper.{Watcher, WatchedEvent}
import scala.collection.JavaConverters._
import scala.collection._

private[finagle] class ZooKeeperHealthHandler extends Watcher {
  private[this] val mu = new AsyncMutex
  val pulse = new Broker[Health]()

  def process(evt: WatchedEvent) = for {
    permit <- mu.acquire()
    () <- evt.getState match {
      case KeeperState.SyncConnected => pulse ! Healthy
      case _ => pulse ! Unhealthy
    }
  } permit.release()
}

private[finagle] object DefaultZkClientFactory
  extends ZkClientFactory(Amount.of(
    ZooKeeperUtils.DEFAULT_ZK_SESSION_TIMEOUT.getValue.toLong,
    ZooKeeperUtils.DEFAULT_ZK_SESSION_TIMEOUT.getUnit).toDuration)

private[finagle] class ZkClientFactory(val sessionTimeout: Duration) {
  private[this] val zkClients: mutable.Map[Set[InetSocketAddress], ZooKeeperClient] = mutable.Map()

  def hostSet(hosts: String) = InetSocketAddressUtil.parseHosts(hosts).toSet

  def get(zkHosts: Set[InetSocketAddress]): (ZooKeeperClient, Offer[Health]) = synchronized {
    val client = zkClients.getOrElseUpdate(zkHosts, new ZooKeeperClient(
        Amount.of(sessionTimeout.inMillis.toInt, CommonTime.MILLISECONDS),
        zkHosts.asJava))
    // TODO: Watchers are tied to the life of the client,
    // which, in turn, is tied to the life of ZkClientFactory.
    // Maybe we should expose a way to unregister watchers.
    val healthHandler = new ZooKeeperHealthHandler
    client.register(healthHandler)
    (client, healthHandler.pulse.recv)
  }

  private[zookeeper] def clear() = synchronized { zkClients.clear() }
} 

//设置package包名称以及导入依赖的类
package org.kylin.scala.zookeeper

import org.apache.zookeeper.{WatchedEvent, Watcher, ZooKeeper}


object ZKApi {
  def main(args: Array[String]): Unit = {
    val zkAddress = "ip"
    val zkClientPort = "2181"
    val zkWatcher = new Watcher {
      override def process(event: WatchedEvent): Unit = println("watcher!")
    }
    val zkTest: ZooKeeper = new ZooKeeper(zkAddress, zkClientPort.toInt, zkWatcher)
    zkTest.getChildren("/", zkWatcher).toString.foreach(print)

  }
}