ericacm

/**
 * hosts(i) corresponds to serversKey(i) and serversVal(i)
 * 
 * hosts are "hostname:port" (same as cluster nodeId).  
 * Note: port is the server's client port.  port+1 and port+2 are used for serversVal.
 * 
 * serversKey are "server.1" (see Zookeeper Admin Guide)
 * serversVal are "hostname:nnnn:nnnn"
 */
case class ServerNames(hosts: IndexedSeq[String], 

ericacm

class ZookeeperService(hostInfo: HostInfo) 
  extends ClusterService with EmbeddedZookeeper with Logging {
    
  // For Standalone use the hostname, eg. localhost
  // For Replicated use hostname:port,hostname:port,...
  @Value("${zookeeperService.hosts:localhost}")
  var hosts: String = _  
  
  // This node's ZK server client port
  def clientPort: Int = hostInfo.appBasePort + 1

ericacm

def rmdir(dir: File) {
  if (dir.isDirectory) {
    for (entry <- dir.listFiles()) {
      if (entry.isDirectory) {
        rmdir(entry)
        entry.delete()
      }
      entry.delete()
    }
    dir.delete()

ericacm

def configureServer(serverNames: ServerNames): () => Unit = {
  // Recreate data directory
  if (dataDir == null || dataDir == "") {
    dataDir = "./zookeeper-" + nodeId
  }
  val dir = new File(dataDir)
  log.info("(Re)creating data directory: " + dataDir)
  rmdir(dir)
  dir.mkdirs()

ericacm

import org.apache.zookeeper.server.quorum.{QuorumPeerMain => ApacheQuorumPeerMain}
import org.apache.zookeeper.server.{ZooKeeperServerMain => ApacheZookeeperServerMain}

trait EmbeddedZookeeper {
  this: ClusterService with Logging =>
  
  @Value("${zookeeperService.server.enabled:false}")
  var isServerEnabled: Boolean = _
  
  // Location of server data directory

ericacm

/*
 * Topic Entity
 */
@Entity
@Table(uniqueConstraints = Array(new UniqueConstraint(columnNames=Array("application_id", "name"))))
@NamedQueries(Array(
  new NamedQuery(name="Topic.findAllByApplication", query="from Topic where application=:application"),
  new NamedQuery(name="Topic.findByApplicationAndName", query="from Topic where application=:application and name=:name")
))
class Topic {

ericacm

import io.Source
import java.io.{FileInputStream, InputStream}
import com.typesafe.config.{ConfigException, Config, ConfigFactory}
import com.foo.dbfs.{FileSystem, Factory}
import com.foo.util.Logging
import com.foo.util.Environment
 
/**
* CslConfig manages CSL Configuration.
*

ericacm

import sbt._
import Keys._
import com.typesafe.sbt.SbtMultiJvm
import com.typesafe.sbt.SbtMultiJvm.MultiJvmKeys.{ MultiJvm }

object ExampleBuild extends Build {

  lazy val buildSettings = Defaults.defaultSettings ++ multiJvmSettings ++ Seq(
    organization := "example",
    version := "1.0",

ericacm

class ZookeeperService(hostInfo: HostInfo) 
  extends ClusterService with EmbeddedZookeeper {

  def nodeId: String = hostInfo.hostname + ":" + hostInfo.basePort + 1  

  @Value("${zookeeperService.client.enabled:false}")
  var enabled: Boolean = _

  @Value("${zookeeperService.leaderPath:/ls}")
  var leaderPath: String = _

ericacm

val participants = clusterService.clusterStatus.participants
val workers = participants.map(nodeIdToActorRef(_))
workers.foreach(_ ! nextWork())