Did this now:
def insertLeadsToDb: Stream[F, Unit] =
combineLeads.parEvalMapUnordered(100) { case (leadForm, lead) =>
Concurrent[F].delay(connectRepo.insertLead(lead, leadForm.connectCampaign.get.id))
}.void
am I discarding again? :D
You create
F[F[Int]]
dammit 😀
Thank you
There we go, awesome :D
You probably added
.void because it was complaining about type misalignment :)
I still have
void, because I don't care about the Int I guess :D
The whole story with Unit subsuming values in Scala is quite unfortunate. So you gotta be careful with that
Yeah, I've been bitten about that about 3 times now, since I started - I keep count :D
there's a warn-discard flag or something that helps in a lot of cases, but not all
The set of flags in that plugin is quite draconian, but it catches a vast array of cases when doing pure FP.
it seems to be a little bit slower that downloading in any other way (other http lib, etc.)
I've got that stream wrapped inside def run(): F[Unit] function:
val arrayStart = "[\n"
val arrayEnd = "\n]"
val chunkSize = 4096
fs2.io.readInputStream(
async.delay(new URL(uri.toString()).openConnection.getInputStream),
chunkSize
)
.map(_.toChar)
.through(rows[F, Char]())
.through(headers[F, String])
.map(_.toMap)
.map(cleanseCsvData)
.map(toJson)
.map(_.compactPrint)
.map { jsonString =>
Metrics.metrics.counter(taskId.taskId).inc()
jsonString
}
.intersperse("\n")
.cons1(arrayStart)
.append(Stream.eval(arrayEnd.pure))
.through(text.utf8Encode)
.through(Files[F].writeAll(outputFile.path, Seq(WRITE, APPEND, CREATE)))
.compile
.drainWhen I schedule it to fiber:
cancellable <- run().startOn(executionContext)and I cancel it later it doesn't end, a file gets completely downloaded and transformed.
What am I doing wrong?
@eltherion: Not sure what's "wrong" in your setup (also could depend on why you want to cancel at all), but you could try to use
Stream's interruptWhen with a Deferred that you complete once you want to cancel (or do onCancel), that should "stop" the stream reliably. But again, depends on your larger setup.
Btw: If you update to fs2-data's latest snapshot (RC1 to come today or tomorrow), then you can simplify the beginning:
fs2.io.readInputStream(...)
.through(text.utf8Decode)
.through(decodeUsingHeaders[CsvRow[String, String]])
.map(_.toMap)(As for the toMap, I just opened a PR that would allow to decode directly into a Map)
@ybasket:matrix.org
Thing is I tried it with Signal:
override def schedule(rawUri: RawUri): F[TaskId] = {
for {
signal <- SignallingRef[F, Boolean](false)
uri <- async.pure(Uri(rawUri.uri))
taskId <- TaskIdComp.create[F]
_ <- taskRunner.run(taskId, uri, signal).startOn(schedulerExecutionContext)
task = Task(
taskId = taskId,
uri = uri,
state = TaskState.Scheduled,
cancelable = Option(new Cancellable[F] {
def cancel: F[Unit] = {
signal.set(true)
}
}),
startTime = None,
endTime = None
)
_ <- tasks.addTask(task)
} yield taskIdInside taskRunner.run(taskId, uri, signal) I use .interruptWhen(signal) on a Stream
but then, when I try to cancel processing (by invoking another api endpoint):
override def cancelTask(taskId: TaskId): F[Option[CancellationResult]] = {
taskService.getTask(taskId).flatMap { taskOption =>
taskOption.fold(async.pure(Option.empty[CancellationResult])) { task =>
task.cancelable.cancel
}
}
}nothing happens... :(
Have you tried adding a
Sync[F].delay(println("Canceling")) in your cancel to see whether the signal is really triggered?
Btw: I would suggest to use http4s if your URIs are HTTP-based, stream it from there instead of using Java's blocking APIs. And in your
cancelTask, you can just use taskOption.traverse(_.cancelable.cancel) instead of the fold.
in my particular case, I'm doing something like that in my parsing library, and was able to implement a sort of "smart stack trace" by passing exceptions backwards in the "through" chain. That's done on my
Transformer[A, B] class, which now I'm trying to implement a toPipe[F] for
sort of a "directed bubble-up" of a normal exception
fs2 isn't that though
Hi Find this error could not find implicit value for parameter compiler: fs2.Stream.Compiler[[x]Any,G]
}.compile.drain. It is complaining "Compiler" at compile method? How to resolve this? Thanks. I searched that I need Sync[]. This is for fs2.1.1. (can not upgrade because of the old project)
}.compile.drain. It is complaining "Compiler" at compile method? How to resolve this? Thanks. I searched that I need Sync[]. This is for fs2.1.1. (can not upgrade because of the old project)
implicit def getJobResultResponseDecoder[F[_] : Sync](implicit cs: ContextShift[F]): EntityDecoder[F, GetJobResultResponse] =
EntityDecoder.decodeBy(MediaType.text.csv) { msg: Message[F] =>
EitherT {
Stream.resource(blockingExecutionContext).flatMap { blocker =>
val bodyStream = msg.body
val pipe: Pipe[F, Byte, Byte] = src =>
src
.through(text.utf8Decode)
.through(text.lines)
.through(text.utf8Encode)
val sink: Pipe[F, Byte, Unit] =
io.file.writeAll(Paths.get(getFilePath), blocker,
Seq(StandardOpenOption.CREATE, StandardOpenOption.APPEND)
)
val stream: Stream[F, Unit] =
bodyStream
.through(pipe)
.through(sink)
stream
}.head.compile.drain
Sync[F].pure(
GetJobResultResponse(locator, size)
.asRight[DecodeFailure]
)}