This directory contains the example that distribute a zip file, install and start a service. This README explain how to quick-start this example.

In order to run the example:

1. Start a gearpump cluster, including Master and Workers.

2. Start the AppMaster:<br>

```bash

target/pack/bin/gear app -jar experiments/distributeservice/target/$SCALA_VERSION_MAJOR/gearpump-experiments-distributeservice_$VERSION.jar org.apache.gearpump.distributeservice.DistributeService -master 127.0.0.1:3000

```

3. Distribute the file:<br>

```bash

target/pack/bin/gear app -jar experiments/distributeservice/target/$SCALA_VERSION_MAJOR/gearpump-experiments-distributeservice_$VERSION.jar org.apache.gearpump.distributeservice.DistributeServiceClient -master 127.0.0.1:3000 -appid $APPID -file ${File_Path}

```

package org.apache.gearpump.experiments.distributeservice

import java.io.File

import akka.actor.{Deploy, Props}

import akka.remote.RemoteScope

import com.typesafe.config.{ConfigFactory, Config}

import org.apache.gearpump.cluster.ClientToMaster.ShutdownApplication

import org.apache.gearpump.cluster.appmaster.ExecutorSystemScheduler.{ExecutorSystemJvmConfig, StartExecutorSystemTimeout, ExecutorSystemStarted}

import org.apache.gearpump.cluster.{ExecutorContext, ApplicationMaster, Application, AppMasterContext}

import org.apache.gearpump.experiments.distributeservice.DistServiceAppMaster.{DistributeFile, FileContainer, GetFileContainer}

import org.apache.gearpump.util._

import org.slf4j.Logger

import akka.pattern.{ask, pipe}

import scala.concurrent.Future

class DistServiceAppMaster(appContext : AppMasterContext, app : Application) extends ApplicationMaster {

import appContext._

import context.dispatcher

implicit val timeout = Constants.FUTURE_TIMEOUT

private val LOG: Logger = LogUtil.getLogger(getClass, app = appId)

private var currentExecutorId = 0

private var fileServerPort = -1

val rootDirectory = new File("/tmp")

val host = context.system.settings.config.getString(Constants.NETTY_TCP_HOSTNAME)

val server = context.actorOf(Props(classOf[FileServer], rootDirectory, host , 0))

override def preStart(): Unit = {

LOG.info(s"Distribute Service AppMaster started")

ActorUtil.launchExecutorOnEachWorker(masterProxy, getExecutorJvmConfig, self)

}

(server ? FileServer.GetPort).asInstanceOf[Future[FileServer.Port]] pipeTo self

override def receive: Receive = {

case ExecutorSystemStarted(executorSystem) =>

import executorSystem.{address, worker, resource => executorResource}

val executorContext = ExecutorContext(currentExecutorId, worker.workerId, appId, self, executorResource)

//start executor

val executor = context.actorOf(Props(classOf[DistServiceExecutor], executorContext, app.userConfig)

.withDeploy(Deploy(scope = RemoteScope(address))), currentExecutorId.toString)

executorSystem.bindLifeCycleWith(executor)

currentExecutorId += 1

case StartExecutorSystemTimeout =>

LOG.error(s"Failed to allocate resource in time")

masterProxy ! ShutdownApplication(appId)

context.stop(self)

case FileServer.Port(port) =>

this.fileServerPort = port

case GetFileContainer =>

val name = Math.abs(new java.util.Random().nextLong()).toString

sender ! new FileContainer(s"http://$host:$fileServerPort/$name")

case DistributeFile(url, fileName) =>

context.children.foreach(_ ! DistributeFile(url, fileName))

}

private def getExecutorJvmConfig: ExecutorSystemJvmConfig = {

val config: Config = Option(app.clusterConfig).map(_.getConfig).getOrElse(ConfigFactory.empty())

val jvmSetting = Util.resolveJvmSetting(config.withFallback(context.system.settings.config)).executor

ExecutorSystemJvmConfig(jvmSetting.classPath, jvmSetting.vmargs,

appJar, username, config)

}

}

object DistServiceAppMaster {

case object GetFileContainer

case class FileContainer(url: String)

case class DistributeFile(url: String, fileName: String)

}

package org.apache.gearpump.experiments.distributeservice

import java.io.File

import akka.actor.Actor

import org.apache.commons.io.FileUtils

import org.apache.gearpump.cluster.{UserConfig, ExecutorContext}

import org.apache.gearpump.experiments.distributeservice.DistServiceAppMaster.DistributeFile

import org.apache.gearpump.util.{FileServer, LogUtil}

import org.slf4j.Logger

class DistServiceExecutor(executorContext: ExecutorContext, userConf : UserConfig) extends Actor {

import executorContext._

private val LOG: Logger = LogUtil.getLogger(getClass, executor = executorId, app = appId)

LOG.info(s"ShellExecutor started!")

override def receive: Receive = {

case DistributeFile(url, fileName) =>

val tempFile = File.createTempFile(s"executor$executorId", fileName)

val client = FileServer.newClient

val bytes = client.get(url).get

FileUtils.writeByteArrayToFile(tempFile, bytes)

LOG.info(s"executor $executorId retrieve file $fileName to local ${tempFile.getPath}")

}

}

package org.apache.gearpump.experiments.distributeservice

import org.apache.gearpump.cluster.client.ClientContext

import org.apache.gearpump.cluster.{AppJar, UserConfig, Application}

import org.apache.gearpump.cluster.main.{ParseResult, CLIOption, ArgumentsParser}

import org.apache.gearpump.util.LogUtil

import org.slf4j.Logger

object DistributeService extends App with ArgumentsParser {

private val LOG: Logger = LogUtil.getLogger(getClass)

override val options: Array[(String, CLIOption[Any])] = Array(

"master" -> CLIOption[String]("<host1:port1,host2:port2,host3:port3>", required = true)

)

def application(config: ParseResult) : Application = {

Application("DistributedShell", classOf[DistServiceAppMaster].getName, UserConfig.empty)

}

LOG.info(s"Distribute Service submitting application...")

val config = parse(args)

val context = ClientContext(config.getString("master"))

implicit val system = context.system

val appId = context.submit(application(config))

context.close()

LOG.info(s"Distribute Service Application started with appId $appId !")

}

package org.apache.gearpump.experiments.distributeservice

import java.io.File

import org.apache.commons.io.FileUtils

import org.apache.gearpump.cluster.client.ClientContext

import org.apache.gearpump.cluster.main.{CLIOption, ArgumentsParser}

import org.apache.gearpump.experiments.distributeservice.DistServiceAppMaster.{DistributeFile, FileContainer, GetFileContainer}

import org.apache.gearpump.util.{FileServer, Constants}

import org.slf4j.{LoggerFactory, Logger}

import akka.pattern.ask

import scala.concurrent.Future

import scala.util.{Failure, Success}

object DistributeServiceClient extends App with ArgumentsParser{

implicit val timeout = Constants.FUTURE_TIMEOUT

import scala.concurrent.ExecutionContext.Implicits.global

private val LOG: Logger = LoggerFactory.getLogger(getClass)

override val options: Array[(String, CLIOption[Any])] = Array(

"master" -> CLIOption[String]("<host1:port1,host2:port2,host3:port3>", required = true),

"appid" -> CLIOption[Int]("<the distributed shell appid>", required = true),

"file" -> CLIOption[String]("<file path>", required = true)

)

val config = parse(args)

val context = ClientContext(config.getString("master"))

val appid = config.getInt("appid")

val file = new File(config.getString("file"))

val appMaster = context.resolveAppID(appid)

(appMaster ? GetFileContainer).asInstanceOf[Future[FileContainer]].map { container =>

val bytes = FileUtils.readFileToByteArray(file)

val result = FileServer.newClient.save(container.url, bytes)

result match {

case Success(_) =>

appMaster ! DistributeFile(container.url, file.getName)

context.close()

case Failure(ex) => throw ex

}

}

}

.aggregate(core, streaming, fsio, examples_kafka, sol, wordcount, complexdag, services, external_kafka, examples, distributedshell, distributeservice)

lazy val distributeservice = Project(

id = "gearpump-experiments-distributeservice",

base = file("experiments/distributeservice"),

settings = commonSettings

) dependsOn(core % "test->test;compile->compile")