jchapuis

 implicit class RichLazyStringList(list: LazyList[String]) {
    def productWith(other: LazyList[String]): LazyList[String] =
      for {
        a <- list
        b <- other
      } yield a + b
  }
 
  LazyList
    .from(1)

jchapuis

trait Service extends Stoppable[Either[E, Unit]] {  
  def start(): Future[E, Unit]  
  def name: String 
}

jchapuis

def stopService(service: Service): Task[Either[ServiceStopError, Service]] =
  Task.deferFuture {
    service.triggerStop()
    service.stopped.map(_.map(_ => service))
  }.timeout(duration).onErrorRecover {
    case _: TimeoutException => ServiceStopTimeoutError(service, duration).asLeft
  }

jchapuis

object CoordinatedShutdownHttpHelpers {
  implicit class FluentHttpCoordinatedShutdown(coordinatedShutdown: CoordinatedShutdown)(
    implicit val system: ActorSystem
  ) {
    private implicit val executionContext: ExecutionContext = system.dispatcher

    def addHttpBinding(
      binding: Http.ServerBinding,
      pendingRequestsHardDeadline: FiniteDuration,
      unbindingDelay: FiniteDuration

jchapuis

import scala.concurrent.{ ExecutionContext, Future, Promise }  

/**
 * Stoppable[T]: represents a process with a 
 * termination function of type `T`
 */   
trait Stoppable[T] {  
  protected implicit def executionContext: ExecutionContext  
  private val stoppedPromise = Promise[T]()  
  private val stopTriggeredPromise = Promise[Unit]()  

jchapuis

implicit class FluentCoordinatedShutdown(coordinatedShutdown: CoordinatedShutdown) {

    /**
      * The first pre-defined phase that applications can add tasks to.
      */
    def beforeServiceUnbind(task: () => Future[Done], name: String): CoordinatedShutdown =
      withTask(CoordinatedShutdown.PhaseBeforeServiceUnbind, task, name)

    /**
      * Stop accepting new incoming connections.

jchapuis

"notify stopped with error in case of service stop timeout" in {
      implicit val scheduler = TestScheduler()
      val serviceA = Service.withName("A").withStopper(() => Future.never).build
      val serviceB = Service.withName("B").withStopper(() => Future.successful(())).build

      Given("a started composite with finite stopped timeout")
      implicit val stoppedTimeout = FiniteStoppedTimeout(30 seconds)
      val composite               = new CompositeService(serviceA, serviceB)
      composite.start()
      scheduler.tick(1 second)

jchapuis

Service
    .withName("safe")
    .withStarter{ ()=> Future.successful(()) }
    .withStopper{ ()=> Future.successful(()) }
    .build

jchapuis

  case class HttpServiceDefinition(
    route: Route,
    host: String,
    port: Int,
    pendingRequestsHardDeadline: FiniteDuration,
    unbindingDelay: FiniteDuration
  )

jchapuis

val akkaApplicationService: Service = builder
       .startSystemDependent(SomeEtcdService())
       .thenStartHttp(
          HttpServiceDefinition(
            new AkkaControllers().routes,
            serverConfig.host,
            serverConfig.port,
            serverConfig.hardDeadline,
            serverConfig.unbindingDelay
          )