我正在尝试使用FS2(0.10.0)构建应用程序。我采用了这个例子from the documentation:
import fs2._
// import fs2._
import fs2.async
// import fs2.async
import scala.concurrent.ExecutionContext
// import scala.concurrent.ExecutionContext
import cats.effect.{ Effect, IO }
// import cats.effect.{Effect, IO}
type Row = List[String]
// defined type alias Row
trait CSVHandle {
def withRows(cb: Either[Throwable,Row] => Unit): Unit
}
// defined trait CSVHandle
def rows[F[_]](h: CSVHandle)(implicit F: Effect[F], ec: ExecutionContext): Stream[F,Row] =
for {
q <- Stream.eval(async.unboundedQueue[F,Either[Throwable,Row]])
_ <- Stream.suspend { h.withRows { e => async.unsafeRunAsync(q.enqueue1(e))(_ => IO.unit) }; Stream.emit(()) }
row <- q.dequeue.rethrow
} yield row
// rows: [F[_]](h: CSVHandle)(implicit F: cats.effect.Effect[F], implicit ec: scala.concurrent.ExecutionContext)fs2.Stream[F,Row]
但它编译失败:
type mismatch;
[error] found : fs2.Stream[F,Row]
[error] required: fs2.Stream[fs2.Pure,?]
[error] Expanded types:
[error] found : fs2.Stream[F,List[String]]
[error] required: fs2.Stream[fs2.Pure,?]"
[error] row <- q.dequeue.rethrow
我担心自己被卡住了,我不明白为什么会这样。任何的想法?
根据this Gitter conversation,示例中存在错误。改为使用:
def rows[F[_]](h: CSVHandle)(implicit F: Effect[F], ec: ExecutionContext): Stream[F,Row] =
for {
q <- Stream.eval(async.unboundedQueue[F,Either[Throwable,Row]])
_ <- Stream.eval { F.delay(h.withRows(e => async.unsafeRunAsync(q.enqueue1(e))(_ => IO.unit))) }
row <- q.dequeue.rethrow
} yield row