scala - ZIO 和多个回调
问题描述
我对使用 ZIO 很陌生。我目前正在用 Scala 编写一个加密交易机器人,同时我也在尝试学习 ZIO。现在我正在打开一个 websocket,这个 websocket 会提供多个回调,直到它关闭,我正在努力将其集成到我的代码中。我当前的代码:
object Main extends zio.App with Logging {
def run(args: List[String]): URIO[Any with Console, ExitCode] = Configuration.getConfiguration.fold(onError, start).exitCode
private val interval: CandlestickInterval = CandlestickInterval.ONE_MINUTE
private def onError(exception: ConfigurationException): ZIO[Any, Throwable, Unit] = {
logger.info("Could not initialize traderbot!")
logger.error(exception.getMessage)
IO.succeed()
}
private final def start(configuration: Configuration): ZIO[Any, Throwable, Unit] = {
for {
binanceClient <- IO.succeed(BinanceApiClientFactory.newInstance(configuration.apiKey, configuration.secret))
webSocketClient <- IO.succeed(binanceClient.newWebSocketClient())
candlesticks <- Task.effectAsync[CandlestickEvent] {
callback =>
webSocketClient.onCandlestickEvent(
"adaeur",
interval, d => callback(IO.succeed(d))
)
})
// TODO Calculate RSI from candlesticks.
} yield candlesticks
}
}
我想继续接收烛台事件并保持功能正常。我看到了一些关于 Zio Streams 的东西,但是我找不到处理重复回调并且易于理解的示例。现在我不能在 de 中使用我的烛台代码来理解。
谢谢你的时间!
解决方案
不幸的是,ZIO
使用时无法处理多个回调,effectAsync
因为数据类型基于单个成功或失败值。
您可以改用ZStream
它具有类似形状的运算符,可以多次调用:
private final def start(configuration: Configuration): ZStream[Any, Throwable, Unit] = {
val candlesticks = ZStream.unwrap(
IO.effectTotal {
val client = BinanceApiClientFactory
.newInstance(configuration.apiKey, configuration.secret)
.newWebSocketClient()
// This variant accepts a return value in the `Left` which
// is called when during shutdown to make sure that the websocket is
// cleaned up
ZStream.effectAsyncInterrupt { cb =>
val closeable = webSocketClient.onCancelstickEvent(
"adaeur",
interval,
d => cb(IO.succeed(d)
)
Left(UIO(closeable.close()))
}
)
for {
candlestick <- candlesticks
// TODO Calculate RSI from candlesticks.
} yield ()
}
推荐阅读
- reactjs - 部署的 Heroku 应用在桌面上运行良好,但在移动设备上只有空白屏幕
- reactjs - 通过函数在子组件中向下传递变量的道具类型应该是什么
- java - [CustomClassMapper]:在 Firebase Firestore 的类上找不到 ... 的设置器/字段
- python - 一次为多个设备的主机名更改创建ansible playbook
- postgresql - 为什么独特的约束在 Ecto 中不起作用?
- python - 为什么在管道组件上调用通信()而不是最后一个会产生损坏的输出?
- linux - 如何在 bash 脚本中优化 cp 函数?
- wordpress - WooCommerce 根据变化以编程方式设置属性
- javascript - 我正在尝试添加一个带有 HTML、CSS 和 Javascript 的下拉菜单
- javascript - 从计数器添加数字,但也使用新计数器创建新按钮