Scala Subscribe类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Scala›Scala教程

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

本文整理汇总了Scala中akka.cluster.pubsub.DistributedPubSubMediator.Subscribe类的典型用法代码示例。如果您正苦于以下问题：Scala Subscribe类的具体用法？Scala Subscribe怎么用？Scala Subscribe使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

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

示例1: UserSocket

//设置package包名称以及导入依赖的类
package actors

import actors.UserSocket.{ChatMessage, Message}
import actors.UserSocket.Message.messageReads
import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Publish, Subscribe}
import akka.event.LoggingReceive
import play.api.libs.json.{Writes, JsValue, Json}
import play.twirl.api.HtmlFormat

import scala.xml.Utility

object UserSocket {
  def props(user: String)(out: ActorRef) = Props(new UserSocket(user, out))

  case class Message(msg: String)

  object Message {
    implicit val messageReads = Json.reads[Message]
  }

  case class ChatMessage(user: String, text: String)

  object ChatMessage {
    implicit val chatMessageWrites = new Writes[ChatMessage] {
      def writes(chatMessage: ChatMessage): JsValue = {
        Json.obj(
          "type" -> "message",
          "user" -> chatMessage.user,
          "text" -> multiLine(chatMessage.text)
        )
      }
    }

    private def multiLine(text: String) = {
      HtmlFormat.raw(text).body.replace("\n", "<br/>")
    }
  }
}

class UserSocket(uid: String, out: ActorRef) extends Actor with ActorLogging {

  val topic = "chat"

  val mediator = DistributedPubSub(context.system).mediator

  mediator ! Subscribe(topic, self)

  def receive = LoggingReceive {
    case js: JsValue =>
      js.validate[Message](messageReads)
        .map(message => Utility.escape(message.msg))
        .foreach { msg => mediator ! Publish(topic, ChatMessage(uid, msg))}

    case c:ChatMessage => out ! Json.toJson(c)
  }
}

开发者ID:onegrx，项目名称:playakkachat，代码行数:59，代码来源:UserSocket.scala

示例2: UserSocketSpec

//设置package包名称以及导入依赖的类
package actors

import actors.UserSocket.{ChatMessage, Message}
import akka.actor._
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{SubscribeAck, Subscribe}
import akka.testkit.TestProbe
import org.specs2.mutable._
import play.api.libs.json._

import scala.language.postfixOps

class UserSocketSpec extends Specification {
  val UserId = "user1"

  "A user socket" should {
    val topic = "chat"

    "send chat message to all subscribers" in new AkkaTestkitSpecs2Support {
      implicit val messageWrites = Json.writes[Message]

      val mediator = DistributedPubSub(system).mediator
      val browser = TestProbe()
      val chatMember1 = TestProbe()
      val chatMember2 = TestProbe()
      mediator ! Subscribe(topic, chatMember1.ref)
      mediator ! Subscribe(topic, chatMember2.ref)
      val socket = system.actorOf(UserSocket.props("user1")(browser.ref), "userSocket")
      val message = "hello"

      socket ! Json.toJson(Message(message))

      chatMember1.ignoreMsg({ case SubscribeAck => true })
      chatMember1.expectMsg(ChatMessage(UserId, message))
      chatMember2.ignoreMsg({ case SubscribeAck => true })
      chatMember2.expectMsg(ChatMessage(UserId, message))
    }

    "forward chat message to browser" in new AkkaTestkitSpecs2Support {
      val browser = TestProbe()
      val socket = system.actorOf(UserSocket.props(UserId)(browser.ref), "userSocket")
      val chatMessage = ChatMessage(UserId, "There is important thing to do!")

      socket ! chatMessage

      browser.expectMsg(Json.toJson(chatMessage))
    }
  }
}

开发者ID:onegrx，项目名称:playakkachat，代码行数:50，代码来源:UserSocketSpec.scala

示例3: topics

//设置package包名称以及导入依赖的类
package org.a4r.actors.pubsub

import akka.actor.Actor
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Subscribe, SubscribeAck}
import org.a4r.Logged
import org.a4r.messages.MessageTopic


trait DistributedSubscriber extends Actor with Logged {

  def topics:Seq[MessageTopic]

  import akka.cluster.pubsub.DistributedPubSubMediator.Subscribe
  val mediator = DistributedPubSub(context.system).mediator

  info(s"subscribing to topics $topics")
  topics.map(topic => {
    mediator ! Subscribe(topic.name, self)
  })

  def wrappedReceive:Receive

  override def receive: Receive = {
    case ack:SubscribeAck =>
      info(s"Am now subscribed: $ack")
    case msg =>
      wrappedReceive(msg)
  }

}

开发者ID:brendanobra，项目名称:a4r，代码行数:32，代码来源:DistributedSubscriber.scala

示例4: AkkaToZmqFlow

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

import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Subscribe
import akka.stream.scaladsl.Sink
import com.trueaccord.scalapb.GeneratedMessage
import tmt.app.library.Connector
import tmt.app.utils.ActorRuntime
import tmt.demo.zeromq_drivers.ZmqPublisherFactory

class AkkaToZmqFlow(actorRuntime: ActorRuntime, zmqPublisherFactory: ZmqPublisherFactory) {
  import actorRuntime._

  def connect[Msg <: GeneratedMessage](
    subscriberTopic: String,
    publishingPort: Int
  ) = {
    val (sourceLinkedRef, source) = Connector.coupling[Msg](Sink.asPublisher(fanout = false))
    DistributedPubSub(system).mediator ! Subscribe(subscriberTopic, sourceLinkedRef)
    val zmqPublisher = zmqPublisherFactory.make[Msg](publishingPort)

    zmqPublisher
      .publish(source)
      .onComplete { x =>
        zmqPublisher.shutdown()
      }
  }
}

开发者ID:mushtaq，项目名称:tcs-jeromq，代码行数:29，代码来源:AkkaToZmqFlow.scala

示例5: FileProcessor

//设置package包名称以及导入依赖的类
package processing

import akka.actor.{ Actor, ActorLogging, ActorRef, ActorSystem, Props }
import common.CommonMessages.{ KillYourself, ProcessLine, fileProcessingTopic }
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{ Subscribe, SubscribeAck }

object FileProcessor {
  def props(): Props = Props(new FileProcessor)
  def createFileProcessor(system: ActorSystem): ActorRef = {
    system.actorOf(FileProcessor.props())
  }
}

class FileProcessor extends Actor with ActorLogging {

  val mediator = DistributedPubSub(context.system).mediator
  mediator ! Subscribe(fileProcessingTopic, self)

  override def receive: Receive = {
    case SubscribeAck(Subscribe(topic, None, `self`)) ?
      log.info(s"subscribing to $topic")

    case msg: ProcessLine =>
      log.info(msg.toString)

    case KillYourself =>
      context stop self

  }
}

开发者ID:jmarin，项目名称:akka-cluster-example，代码行数:32，代码来源:FileProcessor.scala

示例6: Notification

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

import akka.actor.Actor
import akka.cluster.Cluster
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Subscribe, SubscribeAck}

case class Notification(title: String, body: String)

class NotificationSubscriber extends Actor {
  val mediator = DistributedPubSub(context.system).mediator
  // subscribe to the topic named "notifications"
  mediator ! Subscribe("notification", self)

  val cluster = Cluster(context.system)
  val clusterAddress = cluster.selfUniqueAddress

  def receive = {
    case notification: Notification ?
      println(s"Got notification in node $clusterAddress => $notification")
    case SubscribeAck(Subscribe("notification", None, `self`)) ?
      println("subscribing");
  }
}

开发者ID:PacktPublishing，项目名称:Akka-Cookbook，代码行数:25，代码来源:NotificationSubscriber.scala

示例7: UserActor

//设置package包名称以及导入依赖的类
package actor

import akka.actor.{Actor, ActorRef, ActorSelection, Props}
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Subscribe, SubscribeAck}
import models.{ToPublish, _}
import play.api.libs.json.Json
import models.Formats._
import models.MsgTypes._

class UserActor(out: ActorRef) extends Actor {

  val mediator: ActorRef = DistributedPubSub(context.system).mediator
  mediator ! Subscribe("content", self)

  val publisher: ActorSelection = context.system.actorSelection("/user/publisher")
  val playlist: ActorSelection = context.system.actorSelection("/user/playlist")

  playlist ! GetPlaylist

  def receive = {
    case SubscribeAck(Subscribe("content", None, `self`)) ? println("subscribing")

    case msg: Payload if msg.event == PUBLISH =>
      publisher ! ToPublish(msg)
    case ToPublish(payload) =>
      out ! payload

    case msg: Payload if msg.event == PLAY || msg.event == STOP =>
      playlist ! msg
    case msg: Payload if msg.event == NEXT =>
      playlist ! PlayNext
    case msg: Payload if msg.event == ADD =>
      playlist ! AddToPlaylist(msg.content.get)
    case msg: Payload if msg.event == CLEAR =>
      playlist ! ClearPlaylist
      publisher ! ToPublish(msg)
    case msg: Playlist =>
      out ! Payload(PLAYLIST, Some(Json.toJson(msg).toString()))

    case Payload(event, content) =>
      out ! Payload("response", content)
  }
}

object UserActor {
  def props(out: ActorRef) = Props(new UserActor(out))
}

开发者ID:oen9，项目名称:bard-api，代码行数:49，代码来源:UserActor.scala

示例8: Notification

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


import akka.actor.Actor
import akka.cluster.Cluster
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Subscribe, SubscribeAck}

case class Notification(title: String, body: String)

class NotificationSubscriber extends Actor {
  val mediator = DistributedPubSub(context.system).mediator
  mediator ! Subscribe("notification", self)
  val cluster = Cluster(context.system)
  val clusterAddress = cluster.selfUniqueAddress

  def receive = {
    case notification: Notification =>
      println(s"Got notification in node $clusterAddress => $notification")
        case SubscribeAck(Subscribe("notification", None, `self`)) ? println("subscribing");
  }
}

开发者ID:michelnossin，项目名称:ndb，代码行数:23，代码来源:NotificationSubscriber.scala

示例9: UserSocketSpec

//设置package包名称以及导入依赖的类
package actors

import actors.UserSocket.{ChatMessage, Message}
import akka.actor._
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{SubscribeAck, Subscribe}
import akka.testkit.TestProbe
import org.specs2.mutable._
import play.api.libs.json._

import scala.language.postfixOps

class UserSocketSpec extends Specification {
  val UserId = "user1"

  "A user socket" should {
    val topic = "chat" 

    "send chat message to all subscribers" in new AkkaTestkitSpecs2Support {
      implicit val messageWrites = Json.writes[Message]

      val mediator = DistributedPubSub(system).mediator
      val browser = TestProbe()
      val chatMember1 = TestProbe()
      val chatMember2 = TestProbe()
      mediator ! Subscribe(topic, chatMember1.ref)
      mediator ! Subscribe(topic, chatMember2.ref)
      val socket = system.actorOf(UserSocket.props("user1")(browser.ref), "userSocket")
      val message = "hello"

      socket ! Json.toJson(Message(message))

      chatMember1.ignoreMsg({ case SubscribeAck => true })
      chatMember1.expectMsg(ChatMessage(UserId, message))
      chatMember2.ignoreMsg({ case SubscribeAck => true })
      chatMember2.expectMsg(ChatMessage(UserId, message))
    }

    "forward chat message to browser" in new AkkaTestkitSpecs2Support {
      val browser = TestProbe()
      val socket = system.actorOf(UserSocket.props(UserId)(browser.ref), "userSocket")
      val chatMessage = ChatMessage(UserId, "There is important thing to do!")

      socket ! chatMessage

      browser.expectMsg(Json.toJson(chatMessage))
    }
  }
}

开发者ID:GoranSchumacher，项目名称:massive-actors，代码行数:50，代码来源:UserSocketSpec.scala

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Scala PostgresDriver类代码示例发布时间：2022-05-23

Scala Position类代码示例发布时间：2022-05-23

剪的笔顺,诠释剪的笔画,认识剪的部首

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：8786|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：6428|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：5336|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：5809|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：5689|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：6056|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：5656|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：5126|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：5476|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：4844|2022-11-06

客服电话

电子邮件

Scala Subscribe类代码示例

示例1: UserSocket

示例2: UserSocketSpec

示例3: topics

示例4: AkkaToZmqFlow

示例5: FileProcessor

示例6: Notification

示例7: UserActor

示例8: Notification

示例9: UserSocketSpec

请发表评论

全部评论

上一篇：

下一篇：

PacktPublishing/Python-Machine-Learning-

Matlab图像处理—锐化滤波器

sussillo/hfopt-matlab: A parallel, cpu-b

鲁东大学一米网:Win7系统USB驱动器RAM的操

CVE-2022-34529

剪的笔顺,诠释剪的笔画,认识剪的部首

florent37/ViewAnimator: A fluent Android

florent37/Shrine-MaterialDesign2: implem

CVE-2020-36276

六六分期app的软件客服如何联系？(六六分期

doraiso/Mastodon

关于我们

产品与服务

解决方案

139-2527-9053