WIP: Server streaming callable improvements by igorbernstein2 · Pull Request #403 · googleapis/gax-java

igorbernstein2 · 2017-10-31T23:11:37Z

This is not ready for merging, its an early draft.

This is a rough draft of the ServerStreamingCallable api improvements that @garrettjonesgoogle & I were discussing. This PR is meant to get some early feedback on implementation direction and a sanity check. The code needs a lot of debugging, clean up, tests & doc but its feature complete.

Major changes:

add back pressure:
- add ResponseObserver that replaces ApiStreamObserver for ServerStreamingCallables
- add a StreamController that allows the user to manually adjust flow control
- update the GrpcDirectServerStreamingCallable to use the new interfaces
move the conversion of async callbacks to iterator code out of grpc and into gax
- add a ServerStream that replaces the old Iterator and duplicates the logic in grpc's ClientCalls
add syntatic sugar to convert a ServerStreamingCallable into UnaryCallables: first() & all()
deprecated all existing methods on ServerStreamingCallable

@garrettjonesgoogle, please take a look when you have a moment and let me know if I'm on the right track

igorbernstein2 · 2017-11-01T22:23:01Z

rebased

codecov-io · 2017-11-01T22:35:13Z

Codecov Report

Merging #403 into master will increase coverage by 0.85%.
The diff coverage is 77.64%.

@@ Coverage Diff @@## master #403 +/- ##
============================================
+ Coverage 70.53% 71.38% +0.85% - Complexity 634 680 +46 
============================================
Files 136 143 +7 Lines 2912 3334 +422 Branches 220 265 +45 ============================================
+ Hits 2054 2380 +326 - Misses 776 846 +70 - Partials 82 108 +26

Impacted Files	Coverage Δ	Complexity Δ
.../grpc/GrpcServerStreamingRequestParamCallable.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
.../java/com/google/api/gax/rpc/StreamController.java	`100% <100%> (ø)`	`2 <2> (?)`
...om/google/api/gax/rpc/ServerStreamingCallable.java	`43.75% <42.3%> (-56.25%)`	`5 <4> (+1)`
...va/com/google/api/gax/core/SequentialExecutor.java	`70.21% <70.21%> (ø)`	`4 <4> (?)`
...m/google/api/gax/rpc/ReapingStreamingCallable.java	`75.47% <75.47%> (ø)`	`6 <6> (?)`
...main/java/com/google/api/gax/rpc/ServerStream.java	`78.18% <78.18%> (ø)`	`6 <6> (?)`
...in/java/com/google/api/gax/rpc/StreamMediator.java	`80.9% <80.9%> (ø)`	`24 <24> (?)`
.../java/com/google/api/gax/rpc/SpoolingCallable.java	`86.36% <86.36%> (ø)`	`2 <2> (?)`
...a/com/google/api/gax/rpc/FirstElementCallable.java	`87.5% <87.5%> (ø)`	`2 <2> (?)`
...pi/gax/grpc/GrpcDirectServerStreamingCallable.java	`95.74% <95%> (-4.26%)`	`2 <0> (-1)`
... and 7 more

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 9b86a50...710d036. Read the comment docs.

garrettjonesgoogle

I haven't looked very deeply yet, but I would recommend merging from master sooner than later because I committed a large refactoring that will conflict with yours.

Sign in to view

+ * disable automatic flow control. The receiver can also save a reference to the instance and
+ * terminate the stream early using {@code cancel()}.
+ */
+ abstract class StreamController {


garrettjonesgoogle · 2017-11-02T16:51:15Z

Sorry I spoke too soon, you already rebased :-)

garrettjonesgoogle

Overall I like how this is turning out.

Sign in to view

+ * Wraps a GRPC ClientCall in a {@link StreamController}. It feeds events to a {@link
+ * ResponseObserver} and allows for back pressure.
+ */
+ static class GrpcStreamController<RequestT, ResponseT> extends StreamController {


Sign in to view

+ * control by calling {@code disableAutoInboundFlowControl()} in {@code onStart}. After which, the
+ * consumer must request responses by calling {@code request()}.
+ */
+public interface ResponseObserver<V> {


Sign in to view

+ *
+ * <p>By default, the stream uses automatic flow control, where the next response will be delivered
+ * as soon as the current one is processed by onResponse. A consumer can disable automatic flow
+ * control by calling {@code disableAutoInboundFlowControl()} in {@code onStart}. After which, the


Sign in to view

+ * <p>Can be called many times but is never called after {@link #onError(Throwable)} or {@link
+ * #onComplete()} are called.
+ *
+ * <p>Clients may may receive 0 or more onNext callbacks.


igorbernstein2 · 2017-11-03T14:23:54Z

Ok, I'll apply those changes & break this up into smaller chunks to make it easier to review:

nonfunctional updates like pulling server streaming tests into their own files
adding ResponseObserver & back pressure
adding ServerStream
adding first() & all()
It should ready for review on Monday morning

This introduces a richer callback interface `ResponseListener` that can be passed to ServerStreamingCallables. It allows the user to disable automatic flow control.

This allows the ServerStreamingCallables to be easily composable and allows the calls to be cleanly cancelled.

…ynchronization simplification

igorbernstein2 force-pushed the server-streaming-callable-improvements branch from e8863d1 to 40a8843Compare November 1, 2017 22:17

garrettjonesgoogle reviewed Nov 2, 2017
View reviewed changes

igorbernstein2 force-pushed the server-streaming-callable-improvements branch 2 times, most recently from 9ca1217 to 4c6b90fCompare November 2, 2017 23:39

garrettjonesgoogle reviewed Nov 3, 2017
View reviewed changes

igorbernstein2 added 21 commits November 16, 2017 17:13

Add support backpressure support for ServerStreamingCallables.
025f62d
This introduces a richer callback interface `ResponseListener` that can be passed to ServerStreamingCallables. It allows the user to disable automatic flow control.

Lift blocking server streaming iteration into gax.
6ada541
This allows the ServerStreamingCallables to be easily composable and allows the calls to be cleanly cancelled.

WIP: add first() & all() to ServerStreamingCallable
91bfa7c

missing copyrights
d839ebb

make async stream controller static
bdc545c

wip docs for iterable
ec0ab3f

minor simplification
c46d13a

minor tweaks on iterable
80f522c

Add tests for ServerStream
7f94430

ServerStream polish
0b4225e

docs for server streaming callable
00647db

fix bug in backwards compat layer
bf2e161

doc for FirstElementCallable
edf9071

Simplify the implementation of FirstElementCallable
548a4df

FirstElementCallable tests
e037f05

move StreamController to a toplevel type
9d1a885

fix test warnings in FirstElementCallableTest
8cbe3e9

Update SpoolingCallable to match FirstElementCallable: tests, doc & s…
5b2e673
…ynchronization simplification

add error propagation test for grpc streaming callable
63e74ee

more doc
89dc515

WIP: TitratingResponseObserver
e4d0cb6

igorbernstein2 added 5 commits November 16, 2017 17:13

doc fix
7453c38

Rename TitratingResponseObserver to StreamMediator
0611a8f

Remove message param for cancellations & cleanup some warnings
c0a3700

fixes
5e8fd76

Annotations
aa5dca1

igorbernstein2 force-pushed the server-streaming-callable-improvements branch from ab56d25 to aa5dca1Compare November 16, 2017 22:19

igorbernstein2 added 5 commits November 24, 2017 09:24

Fix visibility changes
3a7d695

Add ReapingStreamingCallable
1ab0d0e

lint
8c59262

fix gradle build to allow test runs from intellij
a4b21f9

better fakes
710d036

igorbernstein2 closed this Nov 29, 2017

igorbernstein2 mentioned this pull request Nov 29, 2017
Server streaming callable improvements #433
Closed

igorbernstein2 deleted the server-streaming-callable-improvements branch November 30, 2017 02:11

Navigation Menu

Uh oh!

WIP: Server streaming callable improvements - #403

WIP: Server streaming callable improvements#403
igorbernstein2 wants to merge 31 commits into
googleapis:masterfrom
igorbernstein2:server-streaming-callable-improvements

igorbernstein2 commented Oct 31, 2017•
edited
Loading

Uh oh!

igorbernstein2 commented Nov 1, 2017

Uh oh!

codecov-io commented Nov 1, 2017•
edited
Loading

Uh oh!

garrettjonesgoogle left a comment

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

garrettjonesgoogle commented Nov 2, 2017

Uh oh!

garrettjonesgoogle left a comment

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

igorbernstein2 commented Nov 3, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

igorbernstein2 commented Oct 31, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

igorbernstein2 commented Nov 1, 2017

Uh oh!

codecov-io commented Nov 1, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

garrettjonesgoogle left a comment

Choose a reason for hiding this comment

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

garrettjonesgoogle commented Nov 2, 2017

Uh oh!

garrettjonesgoogle left a comment

Choose a reason for hiding this comment

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

igorbernstein2 commented Nov 3, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

igorbernstein2 commented Oct 31, 2017•
edited
Loading

codecov-io commented Nov 1, 2017•
edited
Loading