Akka Stream的Keep right/left/both如何产生不同的输出？
Akka Stream是一种用于构建可扩展、高吞吐量的流处理应用程序的工具包。它基于Actor模型，提供了一种声明式的方式来处理数据流，并支持并发、容错和可伸缩性。
在Akka Stream中，Keep是一个用于指定在流操作中如何处理结果的方法。它有三个选项：Keep.right、Keep.left和Keep.both，它们可以产生不同的输出。
- Keep.right：当使用Keep.right时，流操作将返回右侧的结果，忽略左侧的结果。这意味着只保留了右侧的结果，并将其作为流操作的输出。这通常用于只关心最终结果的情况。
- Keep.left：当使用Keep.left时，流操作将返回左侧的结果，忽略右侧的结果。这意味着只保留了左侧的结果，并将其作为流操作的输出。这通常用于需要访问流的状态或其他中间结果的情况。
- Keep.both：当使用Keep.both时，流操作将返回左侧和右侧的结果。这意味着保留了左侧和右侧的结果，并将它们作为元组的形式返回。这通常用于需要同时访问流的状态和最终结果的情况。
通过使用不同的Keep选项，可以根据具体的需求来选择保留哪些结果作为流操作的输出。
对于Akka Stream的更详细了解和使用示例，可以参考腾讯云的相关产品：腾讯云流计算（Tencent Cloud StreamCompute），该产品提供了基于Akka Stream的流处理服务，可帮助用户快速构建和部署流处理应用程序。详情请参考：腾讯云流计算产品介绍，了解如何使用Akka Stream进行流处理。
相关·内容
我正在尝试理解Keep在Akka streams中是如何工作的。阅读What does Keep in akka stream mean中的答案，我理解它有助于控制我们从物化器的左侧/右侧/两侧获得结果。然而，我仍然不能构建一个例子，因为我可以改变left/right的值并得到不同的结果。 例如, implicit val system: ActorSystem = ActorSystem("Playground")
implicit val materializer: ActorMaterializer = ActorMaterializer()
val sentence
浏览 17提问于2020-07-02得票数 0
回答已采纳
1回答
保持组合的意义？
、、
我试图在akka中的Keep组合下创建以下示例：
import java.nio.file.Paths
import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, IOResult}
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.util.ByteString
import scala.concurrent.Future
import scala.util.{Failur
浏览 0提问于2019-04-12得票数 4
回答已采纳
2回答
我对akka流很陌生，我想了解流中的物化是如何工作的。
//Print sum of elements from 1 to 10
val newSource = Source(1 to 10)
val flow = Flow[Int].fold(0)((a, b) => a + b)
val sink = Sink.foreach(println)
val sumFuture = newSource.via(flow).toMat(sink)(Keep.left).run()
它使用55和Keep.left打印值Keep.right。两者有何不同？
我想看看Keep.
浏览 4提问于2020-12-22得票数 2
回答已采纳
我正在尝试客户端的websocket，方法是在上跟踪文档。
示例代码是：
import akka.actor.ActorSystem
import akka.Done
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws._
import scala.concurrent.Future
object WebSock
浏览 2提问于2016-10-31得票数 11
3回答
我正在学习akka流，并在代码中遇到Keep.left和Keep.right：
implicit val system = ActorSystem("KafkaProducer")
implicit val materializer = ActorMaterializer()
val source = Source(List("a", "b", "c"))
val sink = Sink.fold[String, String]("")(_ + _)
val runnable: RunnableGraph[Fu
浏览 11提问于2017-11-04得票数 6
回答已采纳
1回答
开始使用akka-streams，我想构建一个简单的例子。在chrome中，使用web套接字插件，我可以通过通过wss://ws.blockchain.info/inv和发送2个命令连接到像这个流那样的流。
{"op":"ping"}
{"op":"unconfirmed_sub"}将在chromes插件窗口中显示结果。
我试图在akka流中实现相同的功能，但遇到了一些问题：
执行了两个命令，但实际上没有得到流输出。
同一个命令执行两次( ping命令)
在遵循或教程时，下面是我的适应性：
obje
浏览 1提问于2017-04-24得票数 7
回答已采纳
1回答
在Play应用程序中，我创建了一个socket服务器：
package controllers
import play.api.mvc._
import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.ActorSystem
import akka.stream.Materializer
class Application @Inject() (cc: ControllerComponents)(implicit system: ActorSystem, mat: Materiali
浏览 15提问于2020-06-03得票数 0
回答已采纳
1回答
如何将接收到的消息广播到两个不同的流 我使用akka流websocket客户端来请求和接收数据websocket服务器。使用从websocket接收到的数据，我想广播到两个不同的流中。下图应能清楚地说明场景： ? 正如您在图像上看到的，它应该随后广播到两个不同的流，以分离汇点。 websocket客户端可以按如下方式创建： import akka.actor.ActorSystem
import akka.Done
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.strea
浏览 20提问于2019-04-15得票数 2
回答已采纳
1回答
我需要创建一个具有以下接口的函数：
import akka.kafka.scaladsl.Consumer.Control
object ItemConversionFlow {
def build(config: StreamConfig): Flow[Item, OtherItem, Control] = {
// Implementation goes here
}
我的问题是，我不知道如何定义流，使其适合上面的接口。
当我做这样的事情时
val flow = Flow[Item]
.map(item => doConversion(item)
.fil
浏览 5提问于2018-01-17得票数 2
我想了解一下akka流中的Keep.both，但我在网上找不到简单的。
谁能提供一个关于Keep.right和Keep.both的非常简单的例子？
我试过了：
implicit val system = ActorSystem("KafkaProducer")
implicit val materializer = ActorMaterializer()
val source = Source.single("Hello")_
val sink = Sink.fold[String, String]("")(_ +
浏览 3提问于2019-04-11得票数 0
我正在尝试将一个Akka Source分成两个单独的。
val requestFlow = Flow[BodyPartEntity].to(Sink.seq) // convert to Seq[BodyPartEntity]
val dataFlow = Flow[BodyPartEntity].to(Sink.asPublisher(fanout = false)) // convert to Publisher[BodyPartEntity]
implicit class EitherSourceExtension[L, R, Mat](source: Source[
浏览 2提问于2021-11-29得票数 0
回答已采纳
下面是来自的代码片段
val sinkUnderTest = Flow[Int].map(_.toString).toMat(Sink.fold("")(_ + _))(Keep.right)
val (ref, future) = Source.actorRef(3, OverflowStrategy.fail)
.toMat(sinkUnderTest)(Keep.both).run()
ref ! 1
ref ! 2
ref ! 3
ref ! akka.actor.Status.Success("done")
val result = Awai
浏览 6提问于2017-08-30得票数 1
回答已采纳
我正在尝试使用方法创建一个对象。某种形式的东西
import akka.stream.OverflowStrategy.fail
import akka.stream.scaladsl.Source
case class Weather(zip : String, temp : Double, raining : Boolean)
val weatherSource = Source.actorRef[Weather](Int.MaxValue, fail)
val sunnySource = weatherSource.filter(!_.raining)
...
我的问题是：如何向基
浏览 0提问于2015-06-11得票数 30
回答已采纳
我未能实现Sink.seq，当到了实现的时候，我失败了，只有这个例外
akka.stream.SubscriptionWithCancelException$StageWasCompleted$:
以下是github：上的完整源代码
我试图将所有被从websocket中推入的元素聚合到一个Sink.seq中。我必须先进行一些json转换，然后才能在Sink.seq中积累一些东西。
val endSink: Sink[WalletNotification[_], Future[Seq[WalletNotification[_]]]] =
Sink.seq[WalletNotifica
浏览 7提问于2021-12-19得票数 0
回答已采纳
2回答
在akka中，我希望将元素放入流中并返回一个对象。我知道元素可能是运行图形的源。但是如何将元素放到运行时并返回对象呢？
import akka.actor.ActorSystem
import akka.stream.QueueOfferResult.{Dropped, Enqueued, Failure, QueueClosed}
import akka.stream.{ActorMaterializer, OverflowStrategy}
import akka.stream.scaladsl.{Keep, Sink, Source}
import scala.Array.range
浏览 5提问于2020-02-06得票数 2
回答已采纳
我有一个演员，它每2秒产生一个NotUsed。也许它没有任何意义，但它只是为了测试目的。 import akka.NotUsed
import akka.actor.{Actor, ActorLogging, Props}
import akka.stream.{ActorMaterializer, KillSwitches}
import akka.stream.scaladsl.{Keep, Sink, Source}
import com.sweetsoft.FsmSystem.{Add, StartTicker, StopTicker}
import scala.concurren
浏览 13提问于2019-05-12得票数 3
回答已采纳
我为Akka编写了一个简单的测试代码(用F#编写，但Scala版本不匹配)：
var source = Source.From(Enumerable.Range(1, 3));
var flow = Flow.FromFunction(new Func<int, string>(x => (x * 2).ToString()));
var sink = Sink.ForEach<string>(output.Add);
var runnable = source.Via(flow).To(sink);
由于Via helper方法只
浏览 3提问于2017-05-17得票数 3
回答已采纳
如何通过Akka WebSocket发送活动WebSocket连接的实时计数？看来Akka改变了Akka的前置阶段的运行。
在下面的第一个代码块中，只有当计数是唯一的连接时，计数才会立即发送。任何在计数超过1的情况下进行连接的客户端，在下一个客户端连接之前都不会接收更新。
第二个代码块删除了WebSocket代码，因此只使用Akka，并且每个流订阅都会收到一个即时计数，如stdout所示。
使用Akka HTTP WebSocket
运行此代码并使用多个浏览器窗口访问将在其浏览器控制台中显示这一点。
import akka.actor.ActorSystem
import akka.http.
浏览 1提问于2017-09-06得票数 1
我是Akka Stream的新手，我想学习如何为我的一个项目处理TCP套接字。我从中摘取了这段代码。
import akka.stream.scaladsl.Framing
val connections: Source[IncomingConnection, Future[ServerBinding]] =
Tcp().bind(host, port)
connections.runForeach { connection =>
println(s"New connection from: ${connection.remoteAddress}")
v
浏览 16提问于2021-02-13得票数 0
回答已采纳
2回答
我正在尝试运行一个简单的示例，但没有成功。我可以创建一个源代码，运行流，然后创建一个文件，但是ByteString不会被写入到文件中。然而，如果我试图打印流输出到控制台，我可以这样做。我是不是漏掉了什么？
import akka.stream._
import akka.stream.scaladsl._
import akka.{ NotUsed, Done}
import akka.actor.ActorSystem
import akka.util.ByteString
import scala.concurrent._
import scala.concurrent.duration
浏览 3提问于2016-09-19得票数 2
1回答
我在用可重新启动的源和接收器处理akka流中的错误。
object Main extends App {
implicit val system: ActorSystem = ActorSystem("akka-streams-system")
val restartSettings =
RestartSettings(1.seconds, 10.seconds, 0.2d)
val restartableSource = RestartSource.onFailuresWithBackoff(restartSettings) {() => {
浏览 8提问于2022-01-10得票数 1
回答已采纳
2回答
我是Akka的新手，并开发了一个示例Akka WebSocket服务器，该服务器使用BroadcastHub (基于Akka 的示例)将文件内容流到客户端。
假设客户端消耗的速度与服务器一样快，我如何度量吞吐量(消息/秒)？
// file source
val fileSource = FileIO.fromPath(Paths.get(path)
// Akka file source
val theFileSource = fileSource
.toMat(BroadcastHub.sink)(Keep.right)
.run
//Akka kafka file source
浏览 2提问于2018-03-09得票数 4
回答已采纳
1回答
我想使用将元素从一个Source复制到另一个Source，但它不像我所期望的那样工作。代码示例，它从Java Source创建一个Akka InputStream，并进行一些转换，并使用alsoTo创建s1的副本。
import java.io.ByteArrayInputStream
import java.nio.charset.StandardCharsets
import akka.actor.ActorSystem
import akka.stream.IOResult
import akka.stream.scaladsl.{Sink, Source, StreamConvert
浏览 5提问于2020-08-14得票数 0
回答已采纳
1回答
协议交换成功与否
、、、
我有下面的代码片段，它没有编译：
import akka.actor.ActorSystem
import akka.Done
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws._
import scala.concurrent._
import scala.concurrent.duration._
obj
浏览 1提问于2019-04-20得票数 2
回答已采纳
1回答
我正在尝试构建一个包含一个简单循环的Akka流。在阅读了文档之后，我尝试将示例代码作为起始库复制，但这也不起作用。代码编译(在包含了示例中缺少的源代码之后)，但没有输出任何内容。看起来好像有什么东西在永远地向后压，但我不明白为什么。
这是我的代码，任何帮助都将不胜感激：
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.stream.scaladsl._
import akka.stream.ClosedShape
object
浏览 5提问于2016-03-04得票数 0
回答已采纳
我正在尝试通过WebSockets来使用一个流，我想解析接收到的消息并在它们到达Sink之前对它们进行转换。但是，每当我使用不接受Message作为输入的Source或Sink时，我就会不断地收到错误。
根据
因此，将WebSocket连接建模为您连接FlowMessage、Message、Mat到或FlowMessage、Message、Mat (连接SourceMessage、Mat和SinkMessage、Mat )的对象。
我仍然不确定我是否正确地理解了这一点。我的困惑是:使用Akka的源、流和Sinks必须始终传递Message类型吗？有办法绕过它吗？最重要的是，这里的最佳实
浏览 3提问于2020-07-09得票数 0
回答已采纳
2回答
我不知道如何立即停止akka stream Runnable Graph？如何使用killswitch来实现这一点？我创建akka streams才几天。在我的例子中，我从文件中读取行，并在flow中执行一些操作，然后写入接收器。我想要做的是，在我想要的时候立即停止读取文件，我希望这可能会停止整个运行的图形。任何关于这方面的想法都将非常感谢。
提前谢谢。
浏览 9提问于2016-08-03得票数 5
回答已采纳
目标是实现一个具有此签名的函数。
def bindedSinkAndSource[A]:(Sink[A, Any], Source[A, Any]) = ???
返回的源发出接收器接收到的任何信息。
我的主要目标是通过指令实现websocket转发器。
转发器图是：
leftReceiver ~> rightEmitter
leftEmitter <~ rightReceiver
其中，leftReceiver和leftEmiter是左端点处理程序流的进出；rightReceiver和rightEmitter是右端点处理程序流的进进出出。
例如：
import akka.NotUs
浏览 4提问于2016-09-21得票数 2
回答已采纳
2回答
akka流toMat
、、、
我正在努力理解toMat在akka流中的作用。例如：
val sink1:Sink[Int, Future[Int]]=Sink.fold[Int,Int](0)(_ + _)
val flow=Flow[Int].fold[Int](0){(x,y)=> x+y}
val runnable = Source (1 to 10).viaMat(flow)(Keep.right).toMat(sink1)(Keep.both)
viaMat和via的用途是什么？
toMat在viaMat到toMat之间做什么？
keep.both的用途是什么，这是否意味着我可以从以前的和当
浏览 2提问于2016-03-05得票数 21
回答已采纳
1回答
我有一个来自像这样的web套接字的akka流，并且希望构建一个可重用的图形阶段(inlet：这个流，FlowShape：添加一个额外的字段到JSON中，指定原点：
{
...,
"origin":"blockchain.info"
}
和卡夫卡的outlet。
我面临以下三个问题：
无法从web套接字流中创建自定义Inlet。
无法将kafka直接集成到流中(请参阅下面的代码)
不确定要添加附加字段的转换器是否需要反序列化json以添加原点。
示例项目(仅流)如下所示：
import system.dispatcher
implicit va
浏览 2提问于2017-05-02得票数 0
回答已采纳
1回答
我在用。
private val (queueSource, connectionPool) = Source.queue[(HttpRequest, Promise[HttpResponse])](queueSize, OverflowStrategy.backpressure).async
.viaMat(poolFlow)(Keep.both)
.toMat(
Sink.foreach({
case ((Success(resp), p)) =>
p.success(resp)
case ((Fai
浏览 3提问于2017-12-09得票数 4
1回答
我试图使用AKKA从一个kafka主题中消费，并使用KPL发送到一个。
我为卡夫卡主题创造了一个来源。
ActorSystem actorSystem = ActorSystemFactory.getClassicActorSystem();
KafkaService kafkaService = new kafkaService(actorSystem);
Source<Message, Consumer.Control> kafkaSource = kakfaService.createSource(runtimeConfiguration.getSourceKafkaTop
浏览 12提问于2022-09-27得票数 0
回答已采纳
2回答
当MyActor接收到Start消息时，它运行一个Akka Stream，并将收到的每一项发布到Akka Event Stream。
class MyActor (implicit system: ActorSystem, materialize: Materializer, ec: ExecutionContext) extends Actor {
override def receive: Receive = {
case Start =>
someSource
.toMat(Sink.foreach(item => system.ev
浏览 0提问于2020-04-30得票数 1
回答已采纳
我想做一些服务器端事件(SSE)到一个网络应用程序。我想我把上证所的水管都修好了。我现在需要在房屋的Akka HTTP端创建一个Source。
我发现你可以这样做：
val source = Source.actorRef(5, akka.stream.OverflowStrategy.dropTail)
我想要做的是以某种方式“发布”到这个来源，大概是通过向演员发送一条信息。我从文档中看到，这个调用创建了Source<T,ActorRef>。
如何获得这个ActorRef实例以便向它发送消息？
浏览 1提问于2017-11-02得票数 2
回答已采纳
1回答
你如何节流在最新的阿克卡(2.4.6)？我想节流Http客户端流，将请求的数量限制在每秒3个请求。我在网上发现了下面的例子，但是对于老Akka和akka-streams API变化太大了，我想不出如何重写它。
def throttled[T](rate: FiniteDuration): Flow[T, T] = {
val tickSource: Source[Unit] = TickSource(rate, rate, () => ())
val zip = Zip[T, Unit]
val in = UndefinedSource[T]
val out = Unde
浏览 2提问于2016-05-24得票数 4
回答已采纳
1回答
我正在从Packt了解Akka流通过Akka Cookbook模块。我运行了示例并得到了以下内容：
为了让参与者系统在流处理完成后退出，我添加了以下回调：
// Future[IOResult]
val future = stream.run()
future.onComplete(_ => system.terminate())
但这一次，应用程序直接退出，没有任何控制台输出：
我想出的一个解决办法是添加Thread.sleep(10000)
我想解释一下这种行为。
浏览 1提问于2018-02-27得票数 0
回答已采纳
1回答
我正在使用Akka stream处理一个包含1839行的CSV文件。我已经添加了计数器来计算处理的行数。 这是我的源代码，我已经确保输入文件中的每一行都少于700个字符。 case class ParsedLine(input: String, field1: String, field2: String, field3: String)
val counter0 = new AtomicInteger()
val counter1 = new AtomicInteger()
val lineSource = FileIO
.fromPath(Paths.get(InputFi
浏览 13提问于2021-07-17得票数 1
回答已采纳
1回答
我使用将数据推送到某个客户端。
这就是我到目前为止所做的：
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.http.javadsl.ConnectHttp;
import akka.http.javadsl.Http;
浏览 3提问于2017-10-06得票数 1
1回答
我试图返回流操作的结果，在本例中是：
与列表之和
平方值
平方值
其表现为：
.fold(0, (aggr, next) -> aggr + next)
.map(x -> x * x)
.map(x -> x * x)
访问我使用的值
final AtomicInteger returnValue = new AtomicInteger();
其次是：
.to(Sink.foreach(x -> {
returnValue.set(x);
浏览 17提问于2021-04-12得票数 0
回答已采纳
我使用Source.queue对HttpRequests进行排队，并在客户端限制它，以便从远程服务器下载文件。我知道Source.queue不是线程安全的，我们需要使用MergeHub来使它成为线程安全。下面是使用Source.queue和cachedHostConnectionPool的代码片段。
import java.io.File
import akka.actor.Actor
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuildin
浏览 0提问于2017-09-22得票数 3
我似乎无法使用在ServerBinding上调用terminate()的推荐方法从连接的服务器端终止websocket连接。
下面包含了建立连接、从服务器端终止连接、等待终止、然后从客户端发送消息的代码。终止后，服务器将成功地处理所有消息。
如何从服务器端终止web套接字连接？
public class AkkaWebSocketServerTerminateTest {
private static final Logger LOGGER = LoggerFactory.getLogger(AkkaWebSocketServerTerminateTest.class);
@Test
pu
浏览 1提问于2018-10-25得票数 0
我已经将websocket客户端连接到akka-http websocket服务器，我如何收听服务器上发生的连接关闭事件(即服务器关闭/服务器关闭websocket连接)？
object Client extends App {
implicit val actorSystem = ActorSystem("akka-system")
implicit val flowMaterializer = ActorMaterializer()
val config = actorSystem.settings.config
val interface = con
浏览 1提问于2016-06-09得票数 3
回答已采纳
在处理各种背压场景时，我实现了这样一种情况:一个订阅者使用缓冲区的速度很慢，而另一个订阅者则消耗掉扔给它的任何东西。使用Scala和Akka Streams。如果愿意，您可以查看代码以及运行它的测试。
我通常尝试开发一个RxJava版本来进行比较，但我在这个版本上卡住了。在Akka Streams中，我可以构建一个图，其中一个源在2个通道上广播，并从这些通道消耗一个慢接收器和一个快速接收器。每个通道可以独立地应用缓冲和节流。在RxJava中，有用于广播的share操作符，但是缓冲和节流逻辑不在Subscriber上，而是在Observable上。因此，我不确定如何应用缓冲和节流，而不影响两个订
浏览 1提问于2016-09-02得票数 2
1回答
我正在为一个相当挑剔的协议(SIP)处理服务器，它需要本地IP和端口作为报头结构的一部分。我正在使用Akka流作为其惊人之处，但我缺少了与BSD的getsockname功能相当的功能。在面向Akka演员的连接中，连接发送一条方便的消息，报告本地IP和端口，但我找不到从Akka Streams版本获得此消息的方法。流直接连接到某些流以供进一步处理，但没有空间容纳包含连接的本地端的IP和端口的消息。
val connection = Tcp().outgoingConnection("www.google.com", 80)
val test = List("GET /
浏览 0提问于2018-03-23得票数 0
回答已采纳
1回答
我很难理解NotUsed和Done在Akka流中的目的和意义。
让我们看看以下两个简单的例子：
使用NotUsed :
implicit val system = ActorSystem("akka-streams")
implicit val materializer = ActorMaterializer()
val myStream: RunnableGraph[NotUsed] =
Source.single("stackoverflow")
.map(s => s.toUpperCase())
.to(Sink.foreach(pr
浏览 4提问于2017-10-24得票数 5
回答已采纳
我正在尝试使用Akka和WebSockets实现服务器推送，并编写了以下程序：
我尝试使用handleMessagesWithSinkSource，使用ActorSource作为源，忽略了Sink；我的问题是没有发送到底层Actor的消息被推送到web套接字输出流中，并且web套接字客户端没有接收到任何东西。为了让它按预期工作，我遗漏了什么；GET /send-42应该将一个带有42的TextMessage推送到输出通道中，并被WS客户机接收。
亲切的问候
浏览 0提问于2020-01-15得票数 1
1回答
我试图使用Akka HTTP和Akka流来运行刮板。我从一堆索引页面开始，解析它们中的链接，然后获取每个链接并解析该页面，以返回一堆单独的链接。所以，像这样：
fetch-top-level-page -> list-of-links-to-child-pages -> fetch-child-page -> list-of-links-in-child-page
我的问题是我连一页都找不到。每一个顶级URL，我都试着用一个死信来获取结果，没有什么比这更深入的了。
在这个示例代码中，我所要做的就是将HttpRequest发送到池中，将其转换为HttpResponse，并通过将
浏览 1提问于2018-03-07得票数 1
回答已采纳
我开始学习Akka，遇到了一个很难解决的挑战，尽管我已经涉猎了文档和相关的Stakoverflow问题：
基于Akka网站上的示例，我在Scala中使用以下代码片段作为基础：
val flow: Flow[Message, Message, Future[Done]] =
Flow.fromSinkAndSourceMat(printSink, Source.maybe)(Keep.left)
val (upgradeResponse, closed) =
Http().singleWebSocketRequest(WebSocketRequest("ws://loc
浏览 3提问于2021-04-17得票数 0
回答已采纳
我希望实现一个服务，其中许多客户端可以使用WebSocket连接到服务器。服务器应该能够在任意内部事件上向所有连接的客户端广播消息。到目前为止，我有这样的代码：
import akka.http.scaladsl.server.RouteResult.route2HandlerFlow
import akka.http.scaladsl.server.Directives._
implicit val system = ActorSystem("Server")
implicit val mat = ActorMaterializer()
// The source to b
浏览 0提问于2018-05-13得票数 1
回答已采纳
1回答
我想引用流中的物化值。下面是代码片段，但它没有编译，错误： type mismatch;
found : (akka.NotUsed, scala.concurrent.Future[akka.Done])
required: (Playground.DomainObj, scala.concurrent.Future[akka.Done]) 代码： import akka.actor.ActorSystem
import akka.stream.scaladsl._
import scala.concurrent.Future
import akka.NotUsed
import
浏览 6提问于2020-08-16得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
