• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

Scala TopicProducer类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Scala中com.lightbend.lagom.scaladsl.broker.TopicProducer的典型用法代码示例。如果您正苦于以下问题:Scala TopicProducer类的具体用法?Scala TopicProducer怎么用?Scala TopicProducer使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了TopicProducer类的12个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Scala代码示例。

示例1: HelloServiceImpl

//设置package包名称以及导入依赖的类
package sample.helloworld.impl

import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import sample.helloworld.api.model.GreetingMessage
import sample.helloworld.api.HelloService


class HelloServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends HelloService {

  override def hello(id: String) = ServiceCall { _ =>
    // Look up the Hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id, None))
  }

  override def useGreeting(id: String) = ServiceCall { request =>
    // Look up the Hello entity for the given ID.
         val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }

  override def greetingsTopic(): Topic[GreetingMessage] = {
    TopicProducer.singleStreamWithOffset {
      offset =>
        persistentEntityRegistry.eventStream(HelloEventTag.instance, offset)
          .map(ev => (convertEvent(ev), offset))
    }
  }

  private def convertEvent(helloEvent: EventStreamElement[HelloEvent]): GreetingMessage = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => GreetingMessage(msg)
    }
  }

} 
开发者ID:knoldus,项目名称:lagom-scala-wordcount.g8,代码行数:44,代码来源:HelloServiceImpl.scala


示例2: BasketServiceImpl

//设置package包名称以及导入依赖的类
package demo.impl.basket

import akka.NotUsed
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.PersistentEntity.InvalidCommandException
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import demo.api.basket.{Basket, BasketService, ExtraTransportExceptions, Item}

import scala.collection.immutable
import scala.concurrent.ExecutionContext

class BasketServiceImpl(persistentEntities: PersistentEntityRegistry)(implicit ec: ExecutionContext)
  extends BasketService with ExtraTransportExceptions{
  override def getBasket(basketId: String): ServiceCall[NotUsed, Basket] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetBasket)
  }

  override def addItem(basketId: String): ServiceCall[Item, NotUsed] = ServiceCall { item =>
    persistentEntities.refFor[BasketEntity](basketId).ask(AddItem(item))
      .map(_ => NotUsed)
      .recoverWith {
        case e: InvalidCommandException => throw BadRequest(e.message)
      }
  }

  override def getTotal(basketId: String): ServiceCall[NotUsed, Int] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetTotal)
  }

  override def clearAll(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(ClearAll)
      .map(x => NotUsed)
      .recoverWith {
        case e: InvalidCommandException => throw BadRequest(e.message)
      }
  }

  override def placeOrder(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(PlaceOrder)
      .map(x => NotUsed)
      .recoverWith {
        case e: InvalidCommandException => throw BadRequest(e.message)
      }
  }

  override def placedOrders: Topic[demo.api.basket.OrderPlaced] =
      TopicProducer.taggedStreamWithOffset(BasketEntityEvent.Tag.allTags.to[immutable.Seq]) { (tag, offset) =>
        persistentEntities.eventStream(tag, offset).collect {
          case EventStreamElement(t, OrderPlaced(id, basket), o)  => {
            demo.api.basket.OrderPlaced(id, basket) -> o
          }
        }
    }
} 
开发者ID:tommpy,项目名称:demo-lagom-checkout,代码行数:57,代码来源:BasketServiceImpl.scala


示例3: BasketServiceImpl

//设置package包名称以及导入依赖的类
package demo.impl.basket

import akka.NotUsed
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.PersistentEntity.InvalidCommandException
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import demo.api.basket.{Basket, BasketService, ExtraTransportExceptions, Item}

import scala.collection.immutable
import scala.concurrent.ExecutionContext

class BasketServiceImpl(persistentEntities: PersistentEntityRegistry)(implicit ec: ExecutionContext)
  extends BasketService with ExtraTransportExceptions{
  override def getBasket(basketId: String): ServiceCall[NotUsed, Basket] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetBasket)
  }

  override def addItem(basketId: String): ServiceCall[Item, NotUsed] = ServiceCall { item =>
    persistentEntities.refFor[BasketEntity](basketId).ask(AddItem(item))
      .map(_ => NotUsed)
      .recoverWith {
        case e: InvalidCommandException => throw BadRequest(e.message)
      }
  }

  override def getTotal(basketId: String): ServiceCall[NotUsed, Int] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetTotal)
  }

  override def clearAll(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(ClearAll)
      .map(x => NotUsed)
      .recoverWith {
        case e: InvalidCommandException => throw BadRequest(e.message)
      }
  }

  override def placeOrder(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(PlaceOrder).map(_ => NotUsed)
  }

  override def placedOrders: Topic[demo.api.basket.OrderPlaced] =
      TopicProducer.taggedStreamWithOffset(BasketEntityEvent.Tag.allTags.to[immutable.Seq]) { (tag, offset) =>
        persistentEntities.eventStream(tag, offset).collect {
          case EventStreamElement(t, OrderPlaced(id, basket), o)  => {
            demo.api.basket.OrderPlaced(id, basket) -> o
          }
        }
    }
} 
开发者ID:tommpy,项目名称:demo-lagom-checkout,代码行数:53,代码来源:BasketServiceImpl.scala


示例4: BasketServiceImpl

//设置package包名称以及导入依赖的类
package demo.impl.basket

import akka.NotUsed
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import demo.api.basket.{Basket, BasketService, Item}

import scala.collection.immutable
import scala.concurrent.ExecutionContext

class BasketServiceImpl(persistentEntities: PersistentEntityRegistry)(implicit ec: ExecutionContext)
  extends BasketService {
  override def getBasket(basketId: String): ServiceCall[NotUsed, Basket] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetBasket)
  }

  override def addItem(basketId: String): ServiceCall[Item, NotUsed] = ServiceCall { item =>
    persistentEntities.refFor[BasketEntity](basketId).ask(AddItem(item)).map(_ => NotUsed)
  }

  override def getTotal(basketId: String): ServiceCall[NotUsed, Int] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(GetTotal)
  }

  override def clearAll(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(ClearAll).map(x => NotUsed)
  }

  override def placeOrder(basketId: String): ServiceCall[NotUsed, NotUsed] = ServiceCall { req =>
    persistentEntities.refFor[BasketEntity](basketId).ask(PlaceOrder).map(_ => NotUsed)
  }

  override def placedOrders: Topic[demo.api.basket.OrderPlaced] =
      TopicProducer.taggedStreamWithOffset(BasketEntityEvent.Tag.allTags.to[immutable.Seq]) { (tag, offset) =>
        persistentEntities.eventStream(tag, offset).collect {
          case EventStreamElement(t, OrderPlaced(id, basket), o)  => {
            demo.api.basket.OrderPlaced(id, basket) -> o
          }
        }
    }
} 
开发者ID:tommpy,项目名称:demo-lagom-checkout,代码行数:44,代码来源:BasketServiceImpl.scala


示例5: MonitorServiceImpl

//设置package包名称以及导入依赖的类
package org.wex.cmsfs.monitor.impl

import java.util.UUID

import akka.persistence.query.Offset
import com.datastax.driver.core.utils.UUIDs
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import org.wex.cmsfs.monitor.api
import org.wex.cmsfs.monitor.api._

import scala.concurrent.{ExecutionContext, Future}

class MonitorServiceImpl(registry: PersistentEntityRegistry)(implicit ec: ExecutionContext) extends MonitorService {
  override def createMonitor: ServiceCall[Monitor, Monitor] = ServiceCall { m =>
    val id = UUIDs.timeBased()
    val monitor = Monitor(Some(id), m.cron, m.mode, m.monitorId, m.connectorId, MonitorStatus.Created)
    val command = CreateMonitor(monitor)
    entityRef(id).ask(command).map(_ => monitor)
  }

  override def monitorCollectEvents = TopicProducer.taggedStreamWithOffset(MonitorEvent.Tag.allTags.toList) { (tag, offset) =>
    println(11)
    registry.eventStream(tag, offset).mapAsync(1)(p => Future{
      println(offset);
      (api.MonitorCreated(UUIDs.random(), 1, 1), offset)
    })
  }


    private def convertEvent(eventStreamElement: EventStreamElement[MonitorEvent]): Future[(api.MonitorEvent, Offset)] = {
      eventStreamElement match {
        case EventStreamElement(_, MonitorCreated(m), offset) =>
          Future((api.MonitorCreated(m.id.get, m.monitorId, m.connectorId), offset))
      }
    }

  private def entityRef(id: UUID) = entityRefString(id.toString)

  private def entityRefString(id: String) = registry.refFor[MonitorEntity](id)
} 
开发者ID:shinhwazx160,项目名称:test,代码行数:43,代码来源:MonitorServiceImpl.scala


示例6: AddServiceImpl

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

import akka.Done
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import com.orion.add.api
import com.orion.add.api.AddService


class AddServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends AddService {
  override def add(): ServiceCall[api.AddNumber, String] = ServiceCall { request =>
    val ref = persistentEntityRegistry.refFor[NumberEntity]("constant")

    ref.ask(AddNumber(request.number))
  }

  override def curry(): ServiceCall[api.CurryWith, Done] = ServiceCall { request =>
    val ref = persistentEntityRegistry.refFor[NumberEntity]("constant")

    ref.ask(CurryWith(request.number))
  }

  override def additionTopic(): Topic[api.CurriedWith] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(NumberEvent.Tags, fromOffset)
        .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(numberEvent: EventStreamElement[NumberEvent]): api.CurriedWith = {
    numberEvent.event match {
      case CurriedWith(num) => api.CurriedWith(num)
    }
  }
} 
开发者ID:utsavgupta,项目名称:add-lagom,代码行数:38,代码来源:AddServiceImpl.scala


示例7: HelloServiceImpl

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

import com.example.hello.api
import com.example.hello.api.{HelloService}
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}


class HelloServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends HelloService {

  override def hello(id: String) = ServiceCall { _ =>
    // Look up the Hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id))
  }

  override def useGreeting(id: String) = ServiceCall { request =>
    // Look up the Hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }


  override def greetingsTopic(): Topic[api.GreetingMessageChanged] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(HelloEvent.Tag, fromOffset)
          .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(helloEvent: EventStreamElement[HelloEvent]): api.GreetingMessageChanged = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => api.GreetingMessageChanged(helloEvent.entityId, msg)
    }
  }
} 
开发者ID:namelos,项目名称:lagom-spike,代码行数:43,代码来源:HelloServiceImpl.scala


示例8: HelloServiceImpl

//设置package包名称以及导入依赖的类
package sample.helloworld.impl

import akka.{Done, NotUsed}
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import sample.helloworld.api.model.GreetingMessage
import sample.helloworld.api.HelloService


class HelloServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends HelloService {

  override def hello(id: String): ServiceCall[NotUsed, _root_.sample.helloworld.impl.Hello#ReplyType] = ServiceCall { _ =>
    // Look up the Hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id, None))
  }

  override def useGreeting(id: String): ServiceCall[GreetingMessage, Done] = ServiceCall { request =>
    // Look up the Hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[HelloEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }

  override def greetingsTopic(): Topic[GreetingMessage] = {
    TopicProducer.singleStreamWithOffset {
      offset =>
        persistentEntityRegistry.eventStream(HelloEventTag.instance, offset)
          .map(ev => (convertEvent(ev), offset))
    }
  }

  private def convertEvent(helloEvent: EventStreamElement[HelloEvent]): GreetingMessage = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => GreetingMessage(msg)
    }
  }

} 
开发者ID:knoldus,项目名称:lagom-play-spike.g8,代码行数:45,代码来源:HelloServiceImpl.scala


示例9: GithubServiceImpl

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

import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}
import com.onedrop.github
import com.onedrop.github.api.GithubService


class GithubServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends GithubService {

  override def getOrg(org: String) = ServiceCall { _ =>
    // Look up the github entity for the given ID.
    val ref = persistentEntityRegistry.refFor[GithubEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id))
  }

  override def getMembers(org: String) = ServiceCall { request =>
    // Look up the github entity for the given ID.
    val ref = persistentEntityRegistry.refFor[GithubEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }


  override def getRepos(org: String): Topic[github.api.GreetingMessageChanged] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(GithubEvent.Tag, fromOffset)
          .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(helloEvent: EventStreamElement[GithubEvent]): github.api.GreetingMessageChanged = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => github.api.GreetingMessageChanged(helloEvent.entityId, msg)
    }
  }
} 
开发者ID:jewelsjacobs,项目名称:one-drop-test,代码行数:43,代码来源:GithubServiceImpl.scala


示例10: LagomhelloServiceImpl

//设置package包名称以及导入依赖的类
package de.breitbandig.lagomhello.impl

import de.breitbandig.lagomhello.api
import de.breitbandig.lagomhello.api.{LagomhelloService}
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}


class LagomhelloServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends LagomhelloService {

  override def hello(id: String) = ServiceCall { _ =>
    // Look up the lagom-hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[LagomhelloEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id))
  }

  override def useGreeting(id: String) = ServiceCall { request =>
    // Look up the lagom-hello entity for the given ID.
    val ref = persistentEntityRegistry.refFor[LagomhelloEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }


  override def greetingsTopic(): Topic[api.GreetingMessageChanged] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(LagomhelloEvent.Tag, fromOffset)
          .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(helloEvent: EventStreamElement[LagomhelloEvent]): api.GreetingMessageChanged = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => api.GreetingMessageChanged(helloEvent.entityId, msg)
    }
  }
} 
开发者ID:aanno,项目名称:lagom-hello,代码行数:43,代码来源:LagomhelloServiceImpl.scala


示例11: AlphaApplication

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.scaladsl.testkit

import akka.persistence.query.Offset
import akka.stream.scaladsl.Source
import akka.stream.testkit.scaladsl.TestSink
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.cassandra.CassandraPersistenceComponents
import com.lightbend.lagom.scaladsl.playjson.EmptyJsonSerializerRegistry
import com.lightbend.lagom.scaladsl.server.{ LagomApplication, LagomApplicationContext, LagomServer, LocalServiceLocator }
import com.lightbend.lagom.scaladsl.testkit.services.{ AlphaEvent, AlphaService }
import org.scalatest.{ AsyncWordSpec, Matchers }
import play.api.libs.ws.ahc.AhcWSComponents

abstract class AlphaApplication(context: LagomApplicationContext)
  extends LagomApplication(context)
  with CassandraPersistenceComponents
  with TestTopicComponents
  with AhcWSComponents {

  override lazy val lagomServer = serverFor[AlphaService](new AlphaServiceImpl())

  override lazy val jsonSerializerRegistry = EmptyJsonSerializerRegistry
}

class AlphaServiceImpl extends AlphaService {
  override def messages: Topic[AlphaEvent] =
    TopicProducer.singleStreamWithOffset { offset =>
      val events = (1 to 10).filter(_ % 2 == 0).map(AlphaEvent.apply)
      Source(events).map(event => (event, Offset.sequence(event.message / 2)))
    }
}

class TopicPublishingSpec extends AsyncWordSpec with Matchers {

  "The AlphaService" should {
    "publish events on alpha topic" in ServiceTest.withServer(ServiceTest.defaultSetup) { ctx =>
      new AlphaApplication(ctx) with LocalServiceLocator
    } { server =>

      implicit val system = server.actorSystem
      implicit val mat = server.materializer

      val client: AlphaService = server.serviceClient.implement[AlphaService]
      val source = client.messages.subscribe.atMostOnceSource

      source.runWith(TestSink.probe[AlphaEvent])
        .request(1)
        .expectNext should ===(AlphaEvent(2))

    }
  }

} 
开发者ID:lagom,项目名称:lagom,代码行数:55,代码来源:TopicPublishingSpec.scala


示例12: ReferentialwebcrawlerServiceImpl

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

import com.andycot.referentialwebcrawler.api
import com.andycot.referentialwebcrawler.api.{ReferentialwebcrawlerService}
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}


class ReferentialwebcrawlerServiceImpl(persistentEntityRegistry: PersistentEntityRegistry) extends ReferentialwebcrawlerService {

  override def hello(id: String) = ServiceCall { _ =>
    // Look up the referential-webcrawler entity for the given ID.
    val ref = persistentEntityRegistry.refFor[ReferentialwebcrawlerEntity](id)

    // Ask the entity the Hello command.
    ref.ask(Hello(id))
  }

  override def useGreeting(id: String) = ServiceCall { request =>
    // Look up the referential-webcrawler entity for the given ID.
    val ref = persistentEntityRegistry.refFor[ReferentialwebcrawlerEntity](id)

    // Tell the entity to use the greeting message specified.
    ref.ask(UseGreetingMessage(request.message))
  }


  override def greetingsTopic(): Topic[api.GreetingMessageChanged] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(ReferentialwebcrawlerEvent.Tag, fromOffset)
          .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(helloEvent: EventStreamElement[ReferentialwebcrawlerEvent]): api.GreetingMessageChanged = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) => api.GreetingMessageChanged(helloEvent.entityId, msg)
    }
  }
} 
开发者ID:frferrari,项目名称:referential-webcrawler,代码行数:43,代码来源:ReferentialwebcrawlerServiceImpl.scala



注:本文中的com.lightbend.lagom.scaladsl.broker.TopicProducer类示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Scala Observer类代码示例发布时间:2022-05-23
下一篇:
Scala Actor类代码示例发布时间:2022-05-23
热门推荐
热门话题
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap