Request streaming support through circe-streaming #812

sergeykolbasov · 2017-07-13T18:31:52Z

Related to #667

Added a project finch-iteratee that depends on io.iteratee
Added default decoder atop of Enumeratee[Future, Buf, A] to finch-circe

With this implementation user would need to add following imports:

import io.finch.iteratee._

to get an access to enumeratorBody[A, CT] and enumeratorJsonBody[A] endpoints (beside adding a new dependency finch-iteratee).

codecov-io · 2017-07-13T18:42:37Z

Codecov Report

Merging #812 into master will decrease coverage by 0.31%.
The diff coverage is 71.11%.

@@            Coverage Diff             @@
##           master     #812      +/-   ##
==========================================
- Coverage   80.25%   79.94%   -0.32%     
==========================================
  Files          46       49       +3     
  Lines         699      738      +39     
  Branches       32       35       +3     
==========================================
+ Hits          561      590      +29     
- Misses        138      148      +10

Impacted Files	Coverage Δ
...amples/src/main/scala/io/finch/iteratee/Main.scala	`0% <0%> (ø)`
circe/src/main/scala/io/finch/circe/package.scala	`100% <100%> (ø)`	⬆️
...e/src/main/scala/io/finch/iteratee/Enumerate.scala	`100% <100%> (ø)`
...ut/src/main/scala/io/finch/argonaut/Decoders.scala	`100% <100%> (+16.66%)`	⬆️
circe/src/main/scala/io/finch/circe/Decoders.scala	`100% <100%> (ø)`	⬆️
...tee/src/main/scala/io/finch/iteratee/package.scala	`95.45% <95.45%> (ø)`
core/src/main/scala/io/finch/Endpoint.scala	`74.8% <0%> (+1.57%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 0adbae8...f16ee5f. Read the comment docs.

vkostyukov

I think it's really cool @imliar! Just some nits from my side around the structure. I like this direction and it seems the next good step(s) would be to allow serving Endpoint[Enumerator[Future, A] by (pretty much what Travis did).

Providing Enumeratee[Future, A, Buf] from within finch-circe and
Providing ToResponse[Enumerator[Future, A] from within finch-iteratee.

Let me If you want to do that as well (either as part of this PR or separately). This is a good work!

vkostyukov · 2017-07-13T19:58:55Z

build.sbt


 lazy val core = project
  .settings(moduleName := "finch-core")
  .settings(allSettings)

+lazy val streaming = project


Let's call it iteratee instead (since it basically provides support to it).

vkostyukov · 2017-07-13T19:59:39Z

circe/src/main/scala/io/finch/circe/Streaming.scala

+import io.iteratee.Enumeratee
+
+trait Streaming {
+  implicit def decoderEnumeratee[A : Decoder]: Enumeratee[Future, Buf, A] =


I think it's totally fine if we provide this instance under just io.finch.circe._.

vkostyukov · 2017-07-13T20:00:09Z

streaming/src/main/scala/io/finch/streaming/package.scala

+/**
+  * Streaming module
+  */
+package object streaming {


Should we extend Module[Fturue] (as Travis did here: #557)?

I guess there is no need for it

What about syntax?

Well, I guess it would save few symbols when lifting into a Enumerator but also add some overhead here and there, such as defining a monad, instead of picking it from imports.
Not sure if it worth it.

We can just point monad instance to one deinfed din catbird. I'm ok not doing this now (extending module), but I think it's a right thing to do at the end of the day.

Looks like we can just extend this: https://github.com/travisbrown/iteratee/blob/master/twitter/src/main/scala/io/iteratee/twitter/FutureModule.scala#L9

vkostyukov · 2017-07-13T20:01:18Z

streaming/src/main/scala/io/finch/streaming/package.scala

+    * An evaluating [[Endpoint]] that reads a required chunked streaming binary body, interpreted as
+    * an `Enumerator[Future, Buf]`. The returned [[Endpoint]] only matches chunked (streamed) requests.
+    */
+  def asyncBufBody: Endpoint[Enumerator[Future, Buf]] = new Endpoint[Enumerator[Future, Buf]] {


I think it's fine not to expose this endpoint. Let's just expose one that deals with A (not Buf).

Yep, instead I'll add an implicit identity Enumeratee[Future, Buf, Buf] in scope.
Just in case if someone would like to use this endpoint for uploading binary data

I like it. Good idea!

vkostyukov · 2017-07-13T20:03:00Z

streaming/src/main/scala/io/finch/streaming/package.scala

+    * An evaluating [[Endpoint]] that reads a required chunked streaming binary body, interpreted as
+    * an `Enumerator[Future, A]`. The returned [[Endpoint]] only matches chunked (streamed) requests.
+    */
+  def asyncJsonBody[A](implicit ee: Enumeratee[Future, Buf, A]): Endpoint[Enumerator[Future, A]] = {


This isn't really coupled to JSON as far as I can tell and can work with anything that's know how to go from Buf to A. How do you feel about renaming it to enumeratorBody?

True, agree on that

sergeykolbasov · 2017-07-13T20:31:11Z

@vkostyukov
I think I'll add it under the same PR here

vkostyukov · 2017-07-14T23:22:13Z

iteratee/src/main/scala/io/finch/iteratee/AsyncEncode.scala

+  def apply(enumerator: Enumerator[Future, A], cs: Charset): Enumerator[Future, Buf]
+}
+
+object AsyncEncode extends EncodeInstances {


Thinking about this more. Maybe it's an overkill?

Looks like we can go from Enumerator[Future, A] to Enumerator[Future, Buf] just by using Encode.Aux[CT, A] (it takes charset). Maybe we don't really need this type class? I understand it makes things symmetric, but I think if we remove it and rename AsyncDecode to something like Enumerate it should all line up pretty nicely.

vkostyukov

Thanks a lot @imliar! I think it looks pretty great already and I'm really happy with how things are lined up.

I added some small nits to the review. Once they are addressed, this is ready to get merged (I'm totally fine if we decide to test it in a separate commit). I'd like to see this go out public as part of 0.16-M2.

vkostyukov · 2017-07-15T17:02:22Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+
+trait IterateeInstances extends FutureModule  {
+
+  implicit def enumeratorToResponse[A, CT <: String](implicit


Why this has to go in its own trait (do we need to de-prioritize it?)?

@vkostyukov for the sake of API cleanness and readability only, to separate "internal" stuff from user API

vkostyukov · 2017-07-15T17:03:03Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+
+  import syntax._
+
+  private[finch] def enumeratorFromReader(reader: Reader): Enumerator[Future, Buf] = {


Can we embed this under the defined enumeratorBody endpoint? I don't think it's used anywhere else.

Used for testing

vkostyukov · 2017-07-15T17:03:44Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+  import syntax._
+
+  private[finch] def enumeratorFromReader(reader: Reader): Enumerator[Future, Buf] = {
+    reader.read(Int.MaxValue).intoEnumerator.flatMap({


nit: No need for wrapping (). Just .flatMap {} should work.

vkostyukov · 2017-07-15T17:04:32Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+    * An evaluating [[Endpoint]] that reads a required chunked streaming JSON body, interpreted as
+    * an `Enumerator[Future, A]`. The returned [[Endpoint]] only matches chunked (streamed) requests.
+    */
+  def enumeratorJsonBody[A](implicit ad: Enumerate.Aux[A, Application.Json]): Endpoint[Enumerator[Future, A]] =


vkostyukov · 2017-07-15T17:08:36Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+      response.setChunked(true)
+      response.contentType = w.value
+      val writer = response.writer
+      val iteratee = foreachM((buf: Buf) => writer.write(buf))


Should we add new-line delimiters between each streamed object? I think we do that for AsyncStream, but I'm not sure what the right answer here.

Dunno, what if it's a buf binary stream?

Also circe-streaming supports decoding of only arrays or objects, so they should be valid JSON structures. I assume that response structure and behaviour should be symmetrical

Yeah, I think I'm fine with that - let's just make sure to document it.

Thinking about it more, it's gonna be invalid construction currently, just concatenated JSON objects. May be \n is not a bad idea in the end.

FWIW, Twitter's streaming API is new-line delimited as well.

Yea, I know it, but confusing part for me is that circe-streaming works in a different manner, waiting for correct JSON structure as an input.
So when we're talking about JSON streaming, request and response formats are going to be different.

I think they don't have to. I'm pretty sure, circe-streaming (along with Jawn) can parse line-delimited JSON objects just fine.

Actually, I take that back - I'm not so sure. Let me play with it a little bit. If it doesn't work, I think, it's our responsibility to add/fix it in Circe.

@vkostyukov

https://github.com/circe/circe/blob/master/modules/streaming/src/main/scala/io/circe/streaming/ParsingEnumeratee.scala#L17

https://github.com/non/jawn/blob/master/README.md#parsing

My guess, we should have an ability to pass ValueStream mode (and other modes) as some configuration in circe-streaming.

vkostyukov · 2017-07-15T18:28:57Z

iteratee/src/main/scala/io/finch/iteratee/Enumerate.scala

+trait EnumerateInstances {
+  def instance[A, CT <: String]
+  (f: (Enumerator[Future, Buf], Charset) => Enumerator[Future, A]): Enumerate.Aux[A, CT] = new Enumerate[A] {
+    override type ContentType = CT


nit: I don't think it overrides anything.

vkostyukov · 2017-07-15T18:29:13Z

iteratee/src/main/scala/io/finch/iteratee/Enumerate.scala

+  (f: (Enumerator[Future, Buf], Charset) => Enumerator[Future, A]): Enumerate.Aux[A, CT] = new Enumerate[A] {
+    override type ContentType = CT
+
+    override def apply(enumerator: Enumerator[Future, Buf], cs: Charset): Enumerator[Future, A] = f(enumerator, cs)


Same here: no need for override.

sergeykolbasov · 2017-07-15T19:29:36Z

@vkostyukov I'd like to add tests under this PR. It'll take day or something. Is there any rush?

vkostyukov · 2017-07-16T19:45:36Z

@imliar No rush at all! Just let me know when you think it's ready!

vkostyukov · 2017-07-18T15:53:18Z

@imliar I have a quick question / topic for discussion. One of the pain points of AsyncStream was a difficulty of signaling mid-stream that we're no longer interested in sending/receiving data. Do you think there are ways for us to do that with iteratee? I'd imagine we need to use ensure (to close a reader) for an inbound Enumerator such that when we throw/raise (I guess, enumerate an error element, i.e., Future.exception) on it, it will close the reader. Do you think we can do the same for the outbound stream?

rpless

Wow, a lot of great work here @imliar!

I'm not sure I how I feel about adding iteratee as a dependency to the circe module, but I suppose its unavoidable if we want to provide a streaming mechanism for circe.

The only other thing we might want to add (probably doesn't have to be in this PR) is an enumeratorText, to match jsonAsyncStreamToResponse and textAsyncStreamToResponse that we provide in the core module.

sergeykolbasov · 2017-07-19T08:20:30Z

@vkostyukov

Could you clarify more, please.
Are we talking about "backpressure" in sense how it could be implemented in HTTP 1.1? You want to close a connection when there is an exception during reader.read() process?

sergeykolbasov · 2017-07-19T08:21:38Z

@rpless io.iteratee is already in use by circe-streaming anyway.

Not sure if it's meaningful to have default enumerator for text, I've followed a principle of YAGNI in this PR.

sergeykolbasov · 2017-07-19T09:14:54Z

Anyway, I think we have external dependency for now

vkostyukov · 2017-07-19T15:26:16Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+  import syntax._
+
+  private[finch] def enumeratorFromReader(reader: Reader): Enumerator[Future, Buf] = {
+    reader.read(Int.MaxValue).intoEnumerator.flatMap {


I mentioned that in the main thread, but I think we should do ensure(reader.discard()) here.

Not here, but after function call. There is no sense to add .ensure for every chunk

Oh yeah, that makes sense.

vkostyukov · 2017-07-19T15:30:33Z

@imliar I wouldn't think of that as a back-pressure it's more like an API for users to control the stream. And you're right - there is not much we can do in HTTP/1.1 besides just cutting the connection (and it's totally fine).

I think we should be fine on the outbound path and essentially failing an Enumerator should result in a closed connection. An inbound path is slightly more complicated, this is why I decided to start this discussion. Assuming we have ensure(reader.discard()) in place, I was wondering if we can trigger that mid-stream somehow.

sergeykolbasov · 2017-07-19T17:15:48Z

@vkostyukov I'm afraid there is not much space for inbound stream control.
Enumerator is immutable, enumeratee & iteratee have no backpressure, so there is no way to send exception back to enumerator. I mean, futures give some kind of control in terms of how fast reader is proccessed, but there is no way to stop processing at all on enumerator side.

May be @travisbrown could give some advice on that.

vkostyukov · 2017-07-19T17:32:59Z

I see. I think we still need ensure(reader.discard()) to make sure we cut the connection if one of the reader.read() failed (I guess it should get it triggered).

vkostyukov · 2017-07-19T17:34:31Z

@imliar Do you want to wait for circe/circe#712 to make sure we match inbound/outbound streaming approach or you think it's fine to ship this now as is?

vkostyukov

Just some nits from me - I really like where things end up!

vkostyukov · 2017-07-19T17:36:22Z

iteratee/src/main/scala/io/finch/iteratee/Enumerate.scala

+
+object Enumerate extends EnumerateInstances {
+
+  type Aux[A, CT <: String] = Enumerate[A] {type ContentType = CT}


Can we add implicitNoFound here to produce more friendly compile error message?

vkostyukov · 2017-07-19T17:37:41Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+      }
+
+      final override def item: RequestItem = items.BodyItem
+      final override def toString: String = "asyncJsonBody"


nit: I think it's named differently now.

vkostyukov · 2017-07-19T17:38:15Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+
+  implicit def enumeratorToJsonResponse[A](implicit
+                                           e: Encode.Aux[A, Application.Json],
+                                           w: Witness.Aux[Application.Json]


This indentation looks a bit weird to my taste.

sergeykolbasov · 2017-07-19T20:31:04Z

@vkostyukov I think it won't hurt nobody if we wait for circe first. Meanwhile could think about it closing of stream

vkostyukov · 2017-07-20T04:40:14Z

json-test/src/main/scala/io/finch/test/JsonLaws.scala

@@ -62,4 +101,9 @@ object JsonLaws {
    new DecodeJsonLaws[A] {
      val decode: Decode.Json[A] = implicitly[Decode.Json[A]]
    }
+
+  def enumeratorDecoding[A : Enumerate.Json]: EnumerateJsonLaws[A] =


enumerating?

vkostyukov · 2017-07-28T14:28:00Z

iteratee/src/main/scala/io/finch/iteratee/package.scala

+          EndpointResult.Skipped
+        } else {
+          val req = input.request
+          def enum = enumeratorFromReader(req.reader).ensureEval(Eval.later(Future.value(req.reader.discard())))


I'd probably prefer if enumeratorFromReader returned an enumerator that already ensures to discard the reader.

Update specs and some outdated code

sergeykolbasov · 2017-08-22T20:48:17Z

@vkostyukov there is a binary incompatibility between latest circe and argonaut (different jawn version dependencies). It hurts! And also breaks tests

travisbrown · 2017-08-22T21:08:27Z

@imliar I've got a PR open for that on the Argonaut side.

Is there evidence that using argonaut-jawn is significantly better in the context of Finch?

travisbrown · 2017-08-22T21:13:55Z

Also there is a jawn-argonaut artifact published from jawn that should be more or less a drop-in replacement for argonaut-jawn and has the most recent versions (also Erik just said he may be removing this in the longer term).

vkostyukov · 2017-08-22T21:19:42Z

I don't see a reason to stick with Jawn in the finch-argonaut context. If it helps, let's just drop this dependency now.

sergeykolbasov · 2017-08-23T11:11:52Z

Removed argonaut-jawn. Tests passed. Yay.

vkostyukov · 2017-08-23T16:25:32Z

Thanks a lot for all the hard work here, @imliar! I can "squash-and-merge" unless you're willing to update your branch (so it's one commit).

@rpless Mind taking a look as well?

rpless · 2017-08-23T17:21:49Z

@vkostyukov +1 from me. Great work @imliar!

sergeykolbasov · 2017-08-23T20:26:06Z

Happy to help 😄

sergeykolbasov force-pushed the request-streaming branch from b2a8368 to b8bb2f7 Compare July 13, 2017 18:42

sergeykolbasov force-pushed the request-streaming branch from b8bb2f7 to 6fe91d7 Compare July 13, 2017 18:45

vkostyukov requested changes Jul 13, 2017

View reviewed changes

sergeykolbasov force-pushed the request-streaming branch 2 times, most recently from 319d052 to e2a6b52 Compare July 14, 2017 22:59

vkostyukov reviewed Jul 14, 2017

View reviewed changes

vkostyukov requested changes Jul 15, 2017

View reviewed changes

vkostyukov reviewed Jul 15, 2017

View reviewed changes

sergeykolbasov mentioned this pull request Jul 17, 2017

Pass AsyncParser modes in circe-streaming circe/circe#711

Closed

vkostyukov requested a review from rpless July 18, 2017 17:06

rpless reviewed Jul 19, 2017

View reviewed changes

vkostyukov reviewed Jul 19, 2017

View reviewed changes

vkostyukov requested changes Jul 19, 2017

View reviewed changes

vkostyukov reviewed Jul 20, 2017

View reviewed changes

sergeykolbasov force-pushed the request-streaming branch from 3ed5e34 to 246fd7c Compare July 20, 2017 17:02

vkostyukov reviewed Jul 28, 2017

View reviewed changes

sergeykolbasov added 12 commits August 22, 2017 22:19

Remove AsyncEncode in favor of Encode. Rename AsyncDecode to Enumerate

17ec7b5

nitpicks fixes

ac793a4

Some basic specs for enumerator endpoints and ToResponse

2d4e8ba

AbstractSpec for Enumerate.Json

5a392be

Add newline delimeter to ToResponse[Enumerator, Application.Json]

8aeee6a

Fix comments

19894eb

Comment broken withToString test

e11f564

Enable enumeratorJsonBody.toString spec

05093fa

ensureEval with reader.discard()

97352e0

Move discard into enumeratorFromReader

e7e8976

ensureEval for response streaming

f7b62d3

Add iteratee example

1f9d149

sergeykolbasov force-pushed the request-streaming branch 2 times, most recently from 2822b71 to 1daecc8 Compare August 22, 2017 19:47

Upgrade to circe 0.9.0-M1

b6cfdd1

Update specs and some outdated code

sergeykolbasov force-pushed the request-streaming branch from 1daecc8 to b6cfdd1 Compare August 22, 2017 20:42

Rework argonaut, remove argonaut-jawn

f16ee5f

vkostyukov approved these changes Aug 23, 2017

View reviewed changes

vkostyukov merged commit 02990b4 into finagle:master Aug 23, 2017

vkostyukov mentioned this pull request Aug 23, 2017

Add module for streaming with iteratees [WIP] #557

Closed

sergeykolbasov mentioned this pull request Oct 8, 2017

v1.0 Roadmap #579

Open

9 tasks

jguitana pushed a commit to jguitana/finch that referenced this pull request Nov 24, 2017

Request streaming support through circe-streaming (finagle#812)

5aad9d0


		trait IterateeInstances extends FutureModule {

		implicit def enumeratorToResponse[A, CT <: String](implicit


		import syntax._

		private[finch] def enumeratorFromReader(reader: Reader): Enumerator[Future, Buf] = {


		object Enumerate extends EnumerateInstances {

		type Aux[A, CT <: String] = Enumerate[A] {type ContentType = CT}

Request streaming support through circe-streaming #812

Request streaming support through circe-streaming #812

Conversation

sergeykolbasov commented Jul 13, 2017 • edited Loading

codecov-io commented Jul 13, 2017 • edited Loading

Codecov Report

vkostyukov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov commented Jul 13, 2017

Choose a reason for hiding this comment

vkostyukov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov Jul 16, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov Jul 17, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov commented Jul 15, 2017

vkostyukov commented Jul 16, 2017

vkostyukov commented Jul 18, 2017

rpless left a comment

Choose a reason for hiding this comment

sergeykolbasov commented Jul 19, 2017

sergeykolbasov commented Jul 19, 2017 • edited Loading

sergeykolbasov commented Jul 19, 2017

vkostyukov Jul 19, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vkostyukov commented Jul 19, 2017

sergeykolbasov commented Jul 19, 2017 • edited Loading

vkostyukov commented Jul 19, 2017

vkostyukov commented Jul 19, 2017

vkostyukov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov commented Jul 19, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sergeykolbasov commented Aug 22, 2017

travisbrown commented Aug 22, 2017

travisbrown commented Aug 22, 2017

vkostyukov commented Aug 22, 2017

sergeykolbasov commented Aug 23, 2017

vkostyukov commented Aug 23, 2017

rpless commented Aug 23, 2017

sergeykolbasov commented Aug 23, 2017

sergeykolbasov commented Jul 13, 2017 •

edited

Loading

codecov-io commented Jul 13, 2017 •

edited

Loading

sergeykolbasov Jul 16, 2017 •

edited

Loading

sergeykolbasov Jul 17, 2017 •

edited

Loading

sergeykolbasov commented Jul 19, 2017 •

edited

Loading

vkostyukov Jul 19, 2017 •

edited

Loading

sergeykolbasov commented Jul 19, 2017 •

edited

Loading