//设置package包名称以及导入依赖的类
package mesosphere.mesos.simulation

import javax.inject.Inject

import mesosphere.marathon.{ MarathonScheduler, MarathonSchedulerDriverHolder, SchedulerDriverFactory }
import org.apache.mesos.SchedulerDriver

class SimulatedSchedulerDriverFactory @Inject() (
  holder: MarathonSchedulerDriverHolder,
  newScheduler: MarathonScheduler)
    extends SchedulerDriverFactory {

  override def createDriver(): SchedulerDriver = {
    val driver = SimulatedDriverWiring.createDriver(newScheduler)
    holder.driver = Some(driver)
    driver
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.mesos.simulation

import akka.actor.Props
import org.apache.mesos.{ Scheduler, SchedulerDriver }

private class SimulatedDriverWiring(scheduler: Scheduler) {
  private lazy val schedulerActorProps = Props(new SchedulerActor(scheduler))
  private lazy val driverActorProps = Props(new DriverActor(schedulerActorProps))
  lazy val driver = new SimulatedDriver(driverActorProps)
}


object SimulatedDriverWiring {
  def createDriver(scheduler: Scheduler): SchedulerDriver = {
    new SimulatedDriverWiring(scheduler).driver
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.mesos.simulation

import akka.actor.{ Actor, Stash }
import akka.event.LoggingReceive
import mesosphere.marathon.stream._
import org.apache.mesos.Protos.{ FrameworkID, MasterInfo, Offer, TaskStatus }
import org.apache.mesos.{ Scheduler, SchedulerDriver }
import org.slf4j.LoggerFactory

import scala.collection.immutable.Seq

object SchedulerActor {
  private val log = LoggerFactory.getLogger(getClass)

  case class Registered(
    frameworkId: FrameworkID,
    master: MasterInfo)

  case class ResourceOffers(offers: Seq[Offer])
}

class SchedulerActor(scheduler: Scheduler) extends Actor with Stash {
  import SchedulerActor._

  var driverOpt: Option[SchedulerDriver] = None

  def receive: Receive = waitForDriver

  def waitForDriver: Receive = LoggingReceive.withLabel("waitForDriver") {
    case driver: SchedulerDriver =>
      log.info("received driver")
      driverOpt = Some(driver)
      context.become(handleCmds(driver))
      unstashAll()

    case _ => stash()
  }

  def handleCmds(driver: SchedulerDriver): Receive = LoggingReceive.withLabel("handleCmds") {
    case Registered(frameworkId, masterInfo) =>
      scheduler.registered(driver, frameworkId, masterInfo)

    case ResourceOffers(offers) =>
      scheduler.resourceOffers(driver, offers)

    case status: TaskStatus =>
      scheduler.statusUpdate(driver, status)
  }

  override def postStop(): Unit = {
    driverOpt.foreach { driver => scheduler.disconnected(driver) }
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.marathon

import javax.inject.Inject

import mesosphere.chaos.http.HttpConf
import mesosphere.marathon.storage.repository.FrameworkIdRepository
import org.apache.mesos.{ Scheduler, SchedulerDriver }
import org.slf4j.LoggerFactory

import scala.concurrent.Await

trait SchedulerDriverFactory {
  def createDriver(): SchedulerDriver
}

class MesosSchedulerDriverFactory @Inject() (
  holder: MarathonSchedulerDriverHolder,
  config: MarathonConf,
  httpConfig: HttpConf,
  frameworkIdRepository: FrameworkIdRepository,
  scheduler: Scheduler)

    extends SchedulerDriverFactory {

  private[this] val log = LoggerFactory.getLogger(getClass.getName)

  log.debug("using scheduler " + scheduler)

  
  override def createDriver(): SchedulerDriver = {
    implicit val zkTimeout = config.zkTimeoutDuration
    val frameworkId = Await.result(frameworkIdRepository.get(), zkTimeout).map(_.toProto)
    val driver = MarathonSchedulerDriver.newDriver(config, httpConfig, scheduler, frameworkId)
    holder.driver = Some(driver)
    driver
  }
} 

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

import org.apache.mesos.Protos.Offer
import org.apache.mesos.{Scheduler, SchedulerDriver}
import org.apache.mesos.{SchedulerDriver, Scheduler}
import org.apache.mesos.Protos._
import scala.collection.JavaConverters._
import scala.collection.mutable

  override def resourceOffers(driver: SchedulerDriver, offers: java.util.List[Offer]) {

    //for every available offer run tasks
    for (offer <- offers.asScala) {
      println(s"offer $offer")
      _tasks.dequeueFirst(value => true) map (cmdString => {
        val cmd = CommandInfo.newBuilder
          .setValue(cmdString)

        //our task will use one cpu
        val cpus = Resource.newBuilder.
          setType(org.apache.mesos.Protos.Value.Type.SCALAR)
          .setName("cpus")
          .setScalar(org.apache.mesos.Protos.Value.Scalar.newBuilder.setValue(1.0))
          .setRole("*")
          .build

        //generate random task id
        val id = "task" + System.currentTimeMillis()

        //create task with given command
        val task = TaskInfo.newBuilder
          .setCommand(cmd)
          .setName(id)
          .setTaskId(TaskID.newBuilder.setValue(id))
          .addResources(cpus)
          .setSlaveId(offer.getSlaveId)
          .build

        driver.launchTasks(offer.getId, List(task).asJava)
      })
    }
  }

  def submitTasks(tasks: String*) = {
    this.synchronized {
      this._tasks.enqueue(tasks: _*)
    }
  }

  override def reregistered(driver: SchedulerDriver, masterInfo: MasterInfo) {}

  override def registered(driver: SchedulerDriver, frameworkId: FrameworkID, masterInfo: MasterInfo) {}
} 

//设置package包名称以及导入依赖的类
package com.github.roberveral.mesosgocd.scheduler

import java.util

import com.github.roberveral.mesosgocd.spec.Task
import com.github.roberveral.mesosgocd.utils.OfferUtils
import org.apache.mesos.Protos
import org.apache.mesos.{Scheduler, SchedulerDriver}

import mesosphere.mesos.protos.Implicits._
import com.github.roberveral.mesosgocd.spec.Implicits._
import scala.collection.JavaConverters._


class GoCDScheduler(task: Task) extends Scheduler with OfferUtils {
  override def offerRescinded(schedulerDriver: SchedulerDriver, offerID: Protos.OfferID): Unit = {}

  override def disconnected(schedulerDriver: SchedulerDriver): Unit = {}

  override def reregistered(schedulerDriver: SchedulerDriver, masterInfo: Protos.MasterInfo): Unit = {}

  override def slaveLost(schedulerDriver: SchedulerDriver, slaveID: Protos.SlaveID): Unit = {}

  override def error(schedulerDriver: SchedulerDriver, s: String): Unit = {}

  override def statusUpdate(schedulerDriver: SchedulerDriver, taskStatus: Protos.TaskStatus): Unit =
    println(s"received status update $taskStatus")

  override def frameworkMessage(schedulerDriver: SchedulerDriver,
                                executorID: Protos.ExecutorID,
                                slaveID: Protos.SlaveID,
                                bytes: Array[Byte]): Unit = {}

  override def resourceOffers(schedulerDriver: SchedulerDriver, list: util.List[Protos.Offer]): Unit = {
    // https://github.com/apache/mesos/blob/1.2.0/include/mesos/mesos.proto
    val receivedOffers = list.asScala
    // Select a valid offer for the task to run
    val optionalAcceptOffer = receivedOffers.find(checkOffer(task, _))
    // If there is a valid offer, get the remaining offers
    val remainingOffers = receivedOffers diff optionalAcceptOffer.toList
    // Launche the task by accepting the offer
    optionalAcceptOffer.foreach(offer =>
      schedulerDriver.launchTasks(List(offer.getId).asJava,
        List(taskToProto(task).setSlaveId(offer.getSlaveId).build()).asJava))
    // Decline the other offers
    remainingOffers foreach(offer => schedulerDriver.declineOffer(offer.getId))
  }

  override def registered(schedulerDriver: SchedulerDriver,
                          frameworkID: Protos.FrameworkID,
                          masterInfo: Protos.MasterInfo): Unit =
    println(s"Framework registered with ID: $frameworkID")

  override def executorLost(schedulerDriver: SchedulerDriver,
                            executorID: Protos.ExecutorID,
                            slaveID: Protos.SlaveID,
                            i: Int): Unit = {}
}