@rossabaker Thanks for the tips.
I agree that Catchable might need some work in the case of failure, but I don’t think fail(…) should come into play in the case of a normal end of stream, right?
Same with the wrap - for a successful Task it should be fine.
The goal with Process[StreamAction, ???] is to conform to an already existing interface.
What I am confused and unsure about is the need to catch something in the case of a successful stream Process. Why wouldn’t I run into the same issues as above if I had a stream of Process(1, 2, 3).toSource: : Process[Task, Int] ?
Catchable that does the trick: implicit val catchable: Catchable[StreamAction] = new Catchable[StreamAction] {
override def fail[A](err: Throwable): StreamAction[A] =
EitherT.fromDisjunction(err.getMessage.left)
override def attempt[A](f: StreamAction[A]): StreamAction[\/[Throwable, A]] =
EitherT {
f.run.attempt.map( _ match {
case -\/(t) => t.left[A].right[StreamError]
case \/-(-\/(e)) => e.left[\/[Throwable,A]]
case \/-(\/-(a)) => a.right[Throwable].right[StreamError]
})
}
}Unhandled Exception when the stream ends normally...
DeepAsyncParser("results", UnwrapArray), and data/position information would be passed between the parsers
EOF token would immediately defer to the parent parser and all secondary output would be either ignored or returned separately in the terminating message
Either[ParseException, Either[J, Seq[J]]] or Either[ParseException, (Option[J], Seq[J])]
jawn.Facade and augmenting it with a tiny amount of additional data to make it work, no major changes to jawn required
jawn.Facade, take it's dual and use it as a output/serialization interface, and not just for parsing https://github.com/lihaoyi/upickle/blob/10904a40b38ec20d5ebb3139856a48c78f57c831/upickle/shared/src/main/scala/upickle/Writers.scala#L19
jawn.Facade is just a generic JSON visitor, and not something specific to parsing
ClassVisitors
{"$type": "pkg.Class, ...} the library can fall back to parsing an AST just for that section of the JSON, which would be transparent except for the performance cost
@travisbrown provisional benchmarks of AST-free serialization puts my WIP branch of uPickle about ~40% faster than circe on my current set of ad-hoc benchmarks, if the encoders/decoders are cached
playJson Read 2761067
playJson Write 3412630
circe Read 6005895
circe Write 5205007
upickleDefault Read 4543628
upickleDefault Write 3814459
upickleLegacy Read 8393416
upickleLegacy Write 7431523Circe is still significantly faster in the case where encoders/decoders are not cached, but I assume I just need to spend a bit of time micro-optimizing the encoder/decoder instantiation code and it's not a fundamental limitation (and more time optimizing should help the cached-encoder benchmark as well)
playJson Read 1975992
playJson Write 2811139
circe Read 4701980
circe Write 4252224
upickleDefault Read 2724334
upickleDefault Write 2443416
upickleLegacy Read 3142672
upickleLegacy Write 2878934Note that in that branch, String -> Case Class and Case Class -> String are both AST-free; my upickle Readers simply implement jawn.Facade, and the upickle Writers effectively extend jawn.Facade => Unit, and so actual definition of reader/writer instances for various types looks pretty similar to what you would see if you pattern matched over the AST (Reader example, Writer example) but it can be driven directly by the parser without any intermediate AST being constructed
{
"field1": "...",
"field2": "...",
"field3": "...",
"entries": [......]
}This json is presented as
fs2.Stream[F, Byte], and i need extract values of first 3 fields and need to get some stream of values from entries array, but i don't know how to use https://github.com/circe/circe-fs2 module here. Please tell me how can i do this or how can i get stream of json tokens for further parsing manually?
Hi,
I'm stuck with an error when parsing a json response from elastic search. This is my code.
def dropLeadingChars: Pipe[F, Byte, Byte] = {
def go(origStr: Stream[F, Byte], beginning: Boolean) : Pull[F, Byte, Unit] = {
logger.info(s"origStr inside dropLeadingChars -> $origStr")
origStr
.pull
.unconsN(if (beginning) 1 else CHUNK_SIZE, true)
.flatMap{
case Some((Chunk.Bytes(chunk, _, _), str)) if beginning && !chunk.headOption.contains('[') =>
Pull.output(Chunk.empty) >> go(str, true)
case Some((seg, str)) =>
logger.info(s"inside some seg is ${seg.toString} and str is $str")
Pull.output(seg) >> go(str, false)
case None => Pull.done
}
}
in => go(in, true).stream
}
val res2: Stream[F, String] = httpClient.getQueryStream(headers, subject.indexName, query, { resp =>
resp
.body
.through(dropLeadingChars)
.dropLastIf(_ == '}')
.dropLastIf(_ == '}')
.chunks
.unwrapJsonArray
.map{ json =>
json.as[Hit].fold(
err => s"Invalid json: $err for payload $json\n",
hit => {
logger.info(s"hit -> $hit")
val instance = hit._source.instance
logger.info(s"instnace -> $instance")
JsonNormalizer.normalizeOld(instance) ++ "\n"
}
)
}
})The error:
,"stack_trace":"org.typelevel.jawn.ParseException: expected true got 'text/p...' (line 1, column 2)\n\tat org.typelevel.jawn.Parser.die(Parser.scala:132)\
text/p part. Its application/json