SubmissionPublisher - Subscriber#onComplete() not invoked when publisher is closed

[Michael McMahon]

Maybe the spec could be tighter around this, but it's not unreasonable 
that there is a
delay in receiving onComplete() notification because of the subscriber 
controlled flow control.
Notifying onError() is not subject to flow control; so you might expect 
that it would be triggered immediately.

Michael.

On 21/02/2017, 11:32, Pavel Bucek wrote:
> SubmissionPublisher#closeExceptionally does trigger 
> Subscriber#onError, but based on javadoc, I cannot really be sure that 
> it will be called, since it contains exactly the same wording as 
> SubmissionPublisher#close
>
> /** * Unless already closed, issues {@link * 
> Flow.Subscriber#onError(Throwable) onError} signals to current * 
> subscribers with the given error, and disallows subsequent * attempts 
> to publish. Future subscribers also receive the given * error. Upon 
> return, this method does <em>NOT</em>guarantee * that all subscribers 
> have yet completed. * * @param error the {@code onError} argument sent 
> to subscribers * @throws NullPointerException if error is null */
>
> So, Pavel, if that is not a bug, how can the SubmissionPublisher be 
> closed in a way that subscribers are notified?
>
> Thanks for the link to the other mailing list - do I understand it 
> correctly that I should move this thread there?
>
> Thanks and regards,
> Pavel
>
>
> On 21/02/2017 12:15, Pavel Rappo wrote:
>> I believe, the most appropriate place for concurrency-related 
>> questions is
>>
>>      http://altair.cs.oswego.edu/mailman/listinfo/concurrency-interest
>>
>> As for the question itself. I don't think this behaviour is a bug.
>> SubmissionPublisher.close() seems to be a graceful way of shutting 
>> down (in
>> contrast with SubmissionPublisher.closeExceptionally()), akin to 
>> putting a EOF
>> on an input stream.
>>
>> My reading of the javadoc is that after SubmissionPublisher.close has 
>> been
>> invoked, the publisher will no longer accept any attempts to publish 
>> items and
>> will call Subscriber.onClose() *eventually*.
>>
>>> On 21 Feb 2017, at 09:24, Pavel Bucek <pavel.bucek at oracle.com> wrote:
>>>
>>> there is a formatting issue in the code snippet, publisher.close() 
>>> should be on the new line:
>>>
>>> {
>>>     SubmissionPublisher<String> publisher =new SubmissionPublisher<>();
>>>     publisher.subscribe(new Flow.Subscriber<String>() {
>>>         @Override public void onSubscribe(Flow.Subscription 
>>> subscription) { }
>>>
>>>         @Override public void onNext(String item) { }
>>>
>>>         @Override public void onError(Throwable throwable) {
>>>             System.out.println("onError()");
>>>         }
>>>
>>>         @Override public void onComplete() {
>>>             System.out.println("onComplete()");
>>>         }
>>>     });
>>>     publisher.submit("item");// if this is commented out, 
>>> #onComplete is invoked.
>>>
>>>     publisher.close();
>>> }
>>>
>>>
>>> On 21/02/2017 10:16, Pavel Bucek wrote:
>>>> Hi all,
>>>>
>>>> firstly - please let me know if this is is a wrong place to send 
>>>> this; I wasn't able to find list specific to concurrency.
>>>>
>>>> Consider following example:
>>>>
>>>> {
>>>>     SubmissionPublisher<String> publisher =new 
>>>> SubmissionPublisher<>();
>>>>     publisher.subscribe(new Flow.Subscriber<String>() {
>>>>         @Override public void onSubscribe(Flow.Subscription 
>>>> subscription) { }
>>>>
>>>>         @Override public void onNext(String item) { }
>>>>
>>>>         @Override public void onError(Throwable throwable) {
>>>>             System.out.println("onError()");
>>>>         }
>>>>
>>>>         @Override public void onComplete() {
>>>>             System.out.println("onComplete()");
>>>>         }
>>>>     });
>>>>     publisher.submit("item");// if this is commented out, 
>>>> #onComplete is invoked. publisher.close();
>>>> }
>>>>
>>>> I'd expect that Subscriber#onComplete is invoked after calling 
>>>> publisher.close(), but it is not happening. Curiously, when I 
>>>> comment out 'publisher.submit("item")', Subscriber#onComplete is 
>>>> indeed invoked.
>>>>
>>>> SubmissionPublisher#close() javadoc says:
>>>>
>>>> /** * Unless already closed, issues {@link * 
>>>> Flow.Subscriber#onComplete() onComplete} signals to current * 
>>>> subscribers, and disallows subsequent attempts to publish. * Upon 
>>>> return, this method does <em>NOT</em>guarantee that all * 
>>>> subscribers have yet completed. */
>>>>
>>>> So it seems like it will be invoked in different thread or 
>>>> something like that, but it is not invoked ever (or more precisely 
>>>> - not during 10 second after the publisher is closed. There is 
>>>> nothing else running on that particular jvm instance).
>>>>
>>>> Also, publisher#isClosed() returns true and 
>>>> publisher#getNumberOfSubscribers() returns 0.
>>>>
>>>> I'm using Java(TM) SE Runtime Environment (build 
>>>> 9-ea+157-jigsaw-nightly-h6115-20170219)
>>>>
>>>> What am I doing wrong?
>>>>
>>>> Thanks and regards, Pavel
>>>>
>