Flow.Processor for Operation submission
Douglas Surber
douglas.surber at oracle.com
Thu Oct 12 20:04:54 UTC 2017
Forgot to answer “what would a Publisher do about it”.
Both cases are programming errors. They are not data errors or caused by an unexpected case. They are strictly the result of an error in program logic. At some level it would be reasonable to throw Error as they should never happen. If it does happen the app is broken and the code must change. So arguably there is no way to recover. The Publisher should abandon whatever request it is handling and try again either on the same request or a new one, hoping that execution does not follow the same erroneous code path.
Douglas
> On Oct 12, 2017, at 12:53 PM, Douglas Surber <douglas.surber at oracle.com> wrote:
>
> There are two cases for values that would be improper to pass to onNext. In both cases the values are themselves proper. The error is passing them to onNext.
>
> 1) Passing an Operation that is a member of OperationGroup A to an operationProcessor for OperationGroup B.
>
> 2) Passing an Operation that has already been submitted.
>
> Code example:
>
> void foo(Connection conn1, Connection conn2) {
> String updateSql = null;
> Flow.Subscription aSubscription = null;
> conn1.countOperation(updateSql).submit(); // this is ok
> Flow.Processor<Operation<Object>, Submission<Object>> p = conn1.operationProcessor();
> p.onSubscribe(aSubscription);
> p.onNext(conn1.countOperation(updateSql)); // this is ok
> conn2.countOperation(updateSql).submit(); // this is ok
> p.onNext(conn2.countOperation(updateSql)); // error case 1. Op created by conn2 passed to Processor for conn1.
> Operation x = conn1.countOperation(updateSql);
> x.submit();
> p.onNext(x); // error case 2. Op created by conn1 but already submitted
> }
>
> Note: Connection inherits from OperationGroup.
>
> Note 2: OperationGroup is a tool for managing errors. By default if an Operation fails all subsequent Operations in the same OperationGroup are skipped, are completed exceptionally with SqlSkippedException. So you can use OperationGroups to control which Operations are skipped and hence when execution resumes. OperationGroups have other capabilities. OperationGroups are Operation factories. Since Connections need to be Operation factories Connections are OperationGroups. There are more or less reasonable use cases for using all of the capabilities of OperationGroup on Connection.
>
> Note 3: A Connection close Operation is never skipped.
>
> Douglas
>
>> On Oct 12, 2017, at 11:20 AM, Viktor Klang <v at lightbend.com> wrote:
>>
>> Hi Douglas,
>>
>> Would you mind elaborating a bit about these improper values? I.e. how do they manifest, and what would a Publisher do about it?
>>
>> (I'm super excited about the prospect of having an async JDBC!)
>>
>> Best regards,
>> √
>>
>> On Oct 12, 2017 17:56, "Douglas Surber" <douglas.surber at oracle.com <mailto:douglas.surber at oracle.com>> wrote:
>> There are two error cases I need to specify. Each error case occurs when the Publisher calls Subscriber.onNext with an improper value. So far as I can comprehend in both cases the Subscriber calls Subscription.cancel. That’s all well and good but I do not see any way for the Subscriber to report which of the two cases provoked the call to cancel. What am I missing?
>>
>> Further it is at least locally reasonable that the Subscriber tolerate these improper values, reporting them as errors but allowing the Publisher to send more values anyway. If we did choose to do this I don’t see any way for the Subscriber to inform the Publisher that anything untoward happened at all.
>>
>> Douglas
>>
>>
>>> On Oct 12, 2017, at 8:20 AM, Konrad “ktoso” Malawski <ktosopl at gmail.com <mailto:ktosopl at gmail.com>> wrote:
>>>
>>> Hi Douglas, David,
>>>
>>> After reading the JavaDoc it’s not clear to me that onNext cannot throw. The JavaDoc says “the resulting behavior is not guaranteed” but it doesn’t say that onNext cannot throw. Can you point to something that has more detail?
>>>
>>>
>>> The full set of rules of Reactive Streams, and thus also, Flow.* semantics is
>>> available on the reactive streams site here: https://github.com/reactive-streams/reactive-streams-jvm#specification <https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_reactive-2Dstreams_reactive-2Dstreams-2Djvm-23specification&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=8PKpeAVkkcN6hA6GJ2zcEVqI3ohcI5138zxnEdjgetA&e=>
>>>
>>> We (specifically; Viktor, initiator of RS, CCed here) have been in talks with Doug Lea to include the actual specification rules
>>> in JavaDoc of Flow as we felt this would happen (that people would not realize there’s a full well-defined spec behind it).
>>> I hope we’ll be able to get that done, as otherwise many people will be left unaware of the rules that govern Flow implementations.
>>> There also is a TCK available <https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_reactive-2Dstreams_reactive-2Dstreams-2Djvm_tree_master_tck&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=mGgS1KAtK30Xw6rPq4K6cOFAL4UN3rz6mI8wLqQiUUA&e=> as testng tests. I’ve been slow to update it <https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_reactive-2Dstreams_reactive-2Dstreams-2Djvm_pull_398&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=xC07YG9iqW5h0mXN1exu72kW4TUTNr_YmhxG94BMKqw&e=> to directly work on Flow.* types – very sorry about that, and will try to get it completed very soon so Flow impls can use it directly.
>>>
>>> Glad to see many RS contributors joining the discussion here, and hopefully we’ll be able to help out making efficient use of Flow in at least some parts of the API.
>>>
>>>
>>> Similar APIs, that might be useful as examples in the discussion (I hope that’s helpful):
>>>
>>> I have yet to read up on details about what Operation/OperationGroup/Submission etc really mean semantically,
>>> but if you want to model a back-pressured stream that back-pressures the incoming operations and emits Submissions (are they like “results”? I’ll read up more),
>>> then modeling it as a Processor is indeed a good idea. This is how we have modeled the back-pressured HTTP-client in Akka HTTP, the type is (translated):
>>> Processor<HttpRequest, HttpResponse>, which may be backed by multiple connections or by a single one internally, so the pulling in requests is handled appropriately.
>>> Looking at the API proposed here it seems that’s what you’re after; so I can confirm we have such API in the wild since some years and it functions well.
>>> The API I talk about is here https://github.com/akka/akka-http/blob/master/akka-http-core/src/main/scala/akka/http/javadsl/Http.scala#L380 <https://urldefense.proofpoint.com/v2/url?u=https-3A__github.com_akka_akka-2Dhttp_blob_master_akka-2Dhttp-2Dcore_src_main_scala_akka_http_javadsl_Http.scala-23L380&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=Ji0jhIlx0uZcHaIXJmiv3P03oR_O4bq2HzBL1FprLoQ&e=> (assume Flow is exactly like a Processor)
>>>
>>> Another example of reactive-streams with JDBC in action is Slick, and it’s wrappers in Alpakka, here’s examples: https://developer.lightbend.com/docs/alpakka/current/slick.html <https://urldefense.proofpoint.com/v2/url?u=https-3A__developer.lightbend.com_docs_alpakka_current_slick.html&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=QowbfwYtawd90nnK4Ubzra0xVXUigKlX4jX5YCgSEds&e=>
>>> It’s pretty simple, you can click on the Java tab for inspiration (Sink == Subscriber, Source == Publisher, Flow == Processor), and basically the simplest version of an RS integration.
>>>
>>> Hope this helps,
>>> happy to help with further design or general reactive-streams questions (we’re trying to figure out how to help here the most efficiently currently).
>>>
>>>
>>> --
>>> Cheers,
>>> Konrad 'ktoso <https://urldefense.proofpoint.com/v2/url?u=http-3A__kto.so&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=x4S1kDPFg6JwQ9mRlkvN22YRVOpNwiB-5BgAITRIsv8&e=>' Malawski
>>> Akka <https://urldefense.proofpoint.com/v2/url?u=http-3A__akka.io_&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=5b5-GC2k85BKW51yEAecgwQYujsGU-PGmIBOts8xY6w&e=> @ Lightbend <https://urldefense.proofpoint.com/v2/url?u=http-3A__lightbend.com_&d=DwMFaQ&c=RoP1YumCXCgaWHvlZYR8PQcxBKCX5YTpkKY057SbK10&r=ChRVNBZ3Ru5F7CzL9kG_sNBRUO0uuqD6z6ltcMO-LbA&m=fU-qyMXO_EKnNqN2Bjq1yF0W8Dd6ATBG1CsLuDUF6ks&s=CBQKBaKMiqEV5R_eUsO3lULxZXmBerKDlw0Gsyw2pG0&e=>
>>> On October 12, 2017 at 23:54:21, Douglas Surber (douglas.surber at oracle.com <mailto:douglas.surber at oracle.com>) wrote:
>>>
>>>> After reading the JavaDoc it’s not clear to me that onNext cannot throw. The JavaDoc says “the resulting behavior is not guaranteed” but it doesn’t say that onNext cannot throw. Can you point to something that has more detail?
>>>>
>>>> Douglas
>>>>
>>>>
>>>>> On Oct 12, 2017, at 7:33 AM, Douglas Surber <douglas.surber at oracle.com <mailto:douglas.surber at oracle.com>> wrote:
>>>>>
>>>>> Thanks for the correction with regard to onNext.
>>>>>
>>>>>> On Oct 11, 2017, at 3:54 PM, Dávid Karnok <akarnokd at gmail.com <mailto:akarnokd at gmail.com> <mailto:akarnokd at gmail.com <mailto:akarnokd at gmail.com>>> wrote:
>>>>>>
>>>>>>> Calling Subscription.onNext with any
>>>>>>> * Operation that is not a member of this OperationGroup, that is
>>>>>>> * was not created by calling one of the Operation factory methods on this
>>>>>>> * OperationGroup, will throw IllegalArgumentException.
>>>>>>
>>>>>> Processor.onNext must not throw, therefore an invalid onNext item should cancel the Subscription and signal onError to the Publisher side of the Processor.
>>>>>>
>>>>>>> Still thinking about how to publish rows.
>>>>>>
>>>>>> In rxjava-jdbc, ResultSet is used as a lambda parameter and the instance shown to the lambda should not escape it, i.e.,
>>>>>>
>>>>>> <T> Flow.Publisher<T> convert(Function<Row, T> mapper);
>>>>>>
>>>>>> 2017-10-11 22:05 GMT+02:00 Douglas Surber <douglas.surber at oracle.com <mailto:douglas.surber at oracle.com> <mailto:douglas.surber at oracle.com <mailto:douglas.surber at oracle.com>> <mailto:douglas.surber at oracle.com <mailto:douglas.surber at oracle.com> <mailto:douglas.surber at oracle.com <mailto:douglas.surber at oracle.com>>>>:
>>>>>> I have found what at first glance appears to be a non-hacky way to use Flow to channel Operations to an OperationGroup. It’s not perfect. The limitation that published Operations must be member Operations of the Operation group is unfortunate but absolutely necessary. Otherwise I think it is ok.
>>>>>>
>>>>>> Delete Connection.onSubscribe.
>>>>>>
>>>>>> Add the following to OperationGroup:
>>>>>>
>>>>>> /**
>>>>>> * Returns a Flow.Processor that subscribes to a sequence of Operations and
>>>>>> * publishes a sequence of corresponding Submissions. The Operations must
>>>>>> * be members of this OperationGroup. Calling Subscription.onNext with any
>>>>>> * Operation that is not a member of this OperationGroup, that is
>>>>>> * was not created by calling one of the Operation factory methods on this
>>>>>> * OperationGroup, will throw IllegalArgumentException. The method
>>>>>> * Subscription.onNext will call submit on each Operation it is passed
>>>>>> * and publish the resulting Submission. Since an Operation can only be
>>>>>> * submitted once, submitting an Operation and calling onNext with that
>>>>>> * submitted Operation will throw IllegalStateException.
>>>>>> *
>>>>>> * Each call to this method returns a new Flow.processor. The Submissions
>>>>>> * published to each Processor are exactly those generated by calling
>>>>>> * submit on the Operations passed as arguments to onNext on the same Processor.
>>>>>> *
>>>>>> * Note: If any Operation is submitted directly, that is by calling submit
>>>>>> * rather than passing it to onNext, the Submission returned by the submit
>>>>>> * call will not be published.
>>>>>> *
>>>>>> * @return a Flow.Processor that accepts Operations and generates Submissions
>>>>>> */
>>>>>> public Flow.Processor<Operation<T>, Submission<T>> operationProcessor();
>>>>>>
>>>>>> Comments?
>>>>>>
>>>>>> Still thinking about how to publish rows.
>>>>>>
>>>>>>
>>>>>> Douglas
>>>>>>
>>>>>>
>>>>>>
>>>>>> --
>>>>>> Best regards,
>>>>>> David Karnok
>>
>
More information about the jdbc-spec-discuss
mailing list