We have defined all of our worker services so let's take a look at Manager and how it drives the baking process by calling other services in the right order. Let's start with the service definition:

trait ManagerService extends Service {
  def bake(count: Int): ServiceCall[NotUsed, Done]
  def sell(count: Int): ServiceCall[NotUsed, Int]
  def report: ServiceCall[NotUsed, Int]

  override def descriptor: Descriptor = {
    import Service._
    named("Bakery").withCalls(
      restCall(Method.POST, "/bake/:count", bake _),
      restCall(Method.POST, "/sell?count", sell _),
      pathCall("/report", report)
    )
  }
}

We define three methods:

• bake, for initiating a baking process for a number of cookies
• sell, for selling cookies if there is enough stock
• report, for checking the number of cookies currently in stock

We map them to two rest calls and a path call. We're using one path and one query parameter just to demonstrate the possibilities offered by the Lagom's descriptor DSL.

Let's get on with the service implementation:

class ManagerServiceImpl(boyService: BoyService,
                         chefService: ChefService,
                         cookService: CookService,
                         bakerService: BakerService,
                         as: ActorSystem)
    extends ManagerService {

  private val count: AtomicInteger = new AtomicInteger(0)

  private val logger = Logger("Manager")

  ...
}

We have to provide all of the services we're about to call as constructor parameters so that we can wire them together later in the definition of the application. We also define logger and count, which will hold the current number of cookies. In a real project, we would implement an event-sourced approach to the internal state of the Manager, but here we're just keeping it in memory for simplicity.

The report and sell methods are implemented by checking the internal state and modifying it if appropriate:

override def sell(cnt: Int): ServiceCall[NotUsed, Int] =
  ServiceCall { _ =>
   if (cnt > count.get()) {
     Future.failed(new IllegalStateException(s"Only $count cookies on sale"))
  } else {
    count.addAndGet(-1 * cnt)
    Future.successful(cnt)
  }
}

override def report: ServiceCall[NotUsed, Int] = ServiceCall { _ =>
  Future.successful(count.get())
}

The bake method is implemented by actually calling other services:

override def bake(count: Int): ServiceCall[NotUsed, Done] = ServiceCall { _ =>
  val sl = shoppingList(count)
  logger.info(s"Shopping list: $sl")
  for {
    groceries <- boyService.shop.invoke(sl)
    done <- chefService.mix.invoke(groceries)
  } yield {
    logger.info(s"Sent $groceries to Chef")
    done
  }
}

Here, we generate a shopping list based on the number of cookies we requested to be baked. Then, in a for-comprehension, we're calling boyService and chefService. With the call of the chef service, we need to return because it is going to take some time for the chef to make the dough.

We already defined the listener for Dough, which is sent back by Chef via the message topic, so we just need to define the flow to handle the incoming messages:

private lazy val chefFlow: Flow[Dough, Done, NotUsed] = Flow[Dough]
  .map { dough: Dough =>
    val fut = cookService.cook.invoke(dough)
    val src = Source.fromFuture(fut)
    val ready: Future[Source[ReadyCookies, NotUsed]] =
      bakerService.bake.invoke(src)
    Source.fromFutureSource(ready)
  }
  .flatMapConcat(identity)
  .map(count.addAndGet(cookies.count))
  .map(_ => Done)

Here again, we're representing the possible one-liner as a couple of statements so that it is easy to spot what is going on: we define Flow, which transforms dough into Future[RawCookies] by calling cookService. bakerService is a streaming one so it expects Source[RawCookies, _] and we create it from Future. The invocation of bakerService returns Future[Source[ReadyCookies, _]] so we convert Future into Source again and then flatten Source[Source[ReadyCookies, _]] with flatMapConcat. Finally, we change the service's internal state and return Done as expected by the subscription method.

It's time to build ManagerApplication together! We need to provide references to all of the services we've used in ManagerImpl. Of course, we'll use serviceClient for this:

abstract class ManagerApplication(context: LagomApplicationContext)
    extends LagomApplication(context) with LagomKafkaClientComponents {
  lazy val boyService: BoyService = serviceClient.implement[BoyService]
  lazy val chefService: ChefService = serviceClient.implement[ChefService]
  lazy val cookService: CookService = serviceClient.implement[CookService]
  lazy val bakerService: BakerService = serviceClient.implement[BakerService]
  override lazy val lagomServer: LagomServer =
    serverFor[ManagerService](wire[ManagerServiceImpl])
}

ManagerServiceImpl itself is constructed using the Macwire as before.