GitHub PullRequestReviewCommentEvent at 2014-04-24 17:56:53
Repository: reactive-streams
Actor: alexandru
Payload: {"comment":{"url":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/comments/11961782","id":11961782,"diff_hunk":"@@ -50,47 +48,63 @@ Compared to Rx, the SPI described here prescribes a mandatory, non-blocking way\n \r\n Iteratees are an abstraction used for consuming a stream, often for parsing it. In this sense they are not a stream transformation or combination tool in themselves.\r\n \r\n-### SPI Components ###\r\n-\r\n-The SPI consists of components that are required to be provided by Reactive Stream implementations but these interfaces should not be exposed to libraries or user code that *use* a Reactive Streams implementation. The reason for this is that the methods used on the SPI level have very strict and rather complex semantic requirements which are likely to be violated by end users.\r\n+### API Components ###\r\n \r\n-The components of the SPI are:\r\n+The API consists of the following components that are required to be provided by Reactive Stream implementations:\r\n \r\n  - Publisher\r\n  - Subscriber\r\n  - Subscription\r\n \r\n-A *Publisher* is a provider of a potentially unbounded number of sequenced elements, publishing them according to the demand received from its Subscriber(s). A Publisher can serve multiple subscribers subscribed dynamically at various points in time. In the case of multiple subscribers the Publisher should respect the processing rates of all of its subscribers (possibly allowing for a bounded drift between them). It must eventually clean up its resources after all of its subscribers have been unsubscribed and shut down. A Publisher will typically support fanning out to multiple Subscribers in order to support the dynamic assembly of processing networks from building blocks that can freely be shared.\r\n+A *Publisher* is a provider of a potentially unbounded number of sequenced elements, publishing them according to the demand received from its Subscriber(s). \r\n \r\n-A *Subscriber* is a component that accepts a sequenced stream of elements provided by a Publisher. At any given time a Subscriber might be subscribed to at most one Publisher. It provides the callback onNext to be called by the upstream Producer, accepting an element that is to be asynchronously processed or enqueued without blocking the Producer. \r\n+The protocol of a `Publisher`/`Subscriber` relationship is defined as:\r\n \r\n-A Subscriber communicates demand to the Publisher via a *Subscription* which is passed to the Subscriber after the subscription has been established. The Subscription exposes the requestMore(int) method that is used by the Subscriber to signal demand to the Publisher. For each of its subscribers the Publisher obeys the following invariant:\r\n+```\r\n+onSubscribe onNext* (onError | onComplete)?\r\n+```\r\n \r\n-*If N is the total number of demand tokens handed to the Publisher P by a Subscriber S during the time period up to a time T, then the number of onNext calls that are allowed to be performed by P on S before T must be less than or equal to N. The number of pending demand tokens must be tracked by the Producer separately for each of its subscribers.*\r\n+- The number of `onNext` events emitted by a `Publisher` to a `Subscriber` will at no point in time exceed the cumulative demand that has been signaled via that `Subscriber`’s `Subscription`.\r\n+- A `Publisher` can send less events that requested and end the `Subscription` by emitting `onComplete` or `onError`.\r\n+- Events sent to a `Subscriber` can only be sent sequentially (no concurrent notifications).\r\n+- If a `Publisher` fails it must emit an `onError`.\r\n+- If a `Publisher` terminates successfully (finite stream) it must emit an `onComplete`.\r\n+- If a Publisher signals either `onError` or `onComplete` on a `Subscriber`, that `Subscriber`’s `Subscription` must be considered canceled.\r\n+- Once a terminal state has been signaled (`onError`, `onNext`) no further events can be sent.\r\n+- Upon receiving a `Subscription.cancel` request it should stop sending events as soon as it can. \r\n+- Calling `onError` or `onComplete` is not required after having received a `Subscription.cancel`.\r\n+- The `Publisher.subscribe` method can be called as many times as wanted as long as it is with a different `Subscriber` each time. It is up to the `Publisher` whether underlying streams are shared or not. In other words, a `Publisher` can support multi-subscribe and then choose whether each `Subscription` is unicast or multicast.\r\n+- A `Publisher` can refuse subscriptions (calls to `subscribe`) if it is unable or unwilling to serve them (overwhelmed, fronting a single-use data source, etc) and can do so by calling `Subscriber.onError` instead of `Subscriber.onSubscribe` on the `Subscriber` instance calling `subscribe`.\r\n+- A `Publisher` should not throw an `Exception`. The only legal way to signal failure (or reject a `Subscription`) is via the `Subscriber.onError` method.\r\n \r\n-Subscribers that do not currently have an active subscription may subscribe to a Publisher. The only guarantee for subscribers attached at different points in time is that they all observe a common suffix of the stream, i.e. they receive the same elements after a certain point in time but it is not guaranteed that they see exactly the same number of elements. This obviously only holds if the subscriber does not cancel its subscription before the stream has been terminated.\r\n \r\n-> In practice there is a difference between the guarantees that different publishers can provide for subscribers attached at different points in time. For example Publishers serving elements from a strict collection (“cold”) might guarantee that all subscribers see *exactly* the same elements (unless unsubscribed before completion) since they can replay the elements from the collection at any point in time. Other publishers might represent an ephemeral source of elements (e.g. a “hot” TCP stream) and keep only a limited output buffer to replay for future subscribers.\r\n+A *`Subscriber`* is a component that accepts a sequenced stream of elements provided by a `Publisher`. At any given time a `Subscriber` might be subscribed to at most one `Publisher`. It provides the callback `onNext` to be called by the upstream `Publisher`, accepting an element that is to be processed or enqueued without blocking the `Publisher`. \r","path":"README.md","position":61,"original_position":60,"commit_id":"08744a5ef1ee8325b2f948d3989030cfedd4cb07","original_commit_id":"da058ccd0326d655ab2a62f7e6bc25165ef587bc","user":{"login":"alexandru","id":11753,"avatar_url":"https://avatars.githubusercontent.com/u/11753?","gravatar_id":"8a4b8068de29da6d2a4831cea986b53a","url":"https://api.github.com/users/alexandru","html_url":"https://github.com/alexandru","followers_url":"https://api.github.com/users/alexandru/followers","following_url":"https://api.github.com/users/alexandru/following{/other_user}","gists_url":"https://api.github.com/users/alexandru/gists{/gist_id}","starred_url":"https://api.github.com/users/alexandru/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexandru/subscriptions","organizations_url":"https://api.github.com/users/alexandru/orgs","repos_url":"https://api.github.com/users/alexandru/repos","events_url":"https://api.github.com/users/alexandru/events{/privacy}","received_events_url":"https://api.github.com/users/alexandru/received_events","type":"User","site_admin":false},"body":"I think @benjchristensen is thinking about a publisher producing a subscription that does this:\r\n\r\n```java\r\nnew Subscription {\r\n  // ...\r\n  public void request(int nr) {\r\n    AtomicInteger requestedNr = new AtomicInteger(0);\r\n\r\n    schedule {\r\n      while (requestedNr.getAndDecrement > 0) \r\n        subcriber.onNext(something)\r\n    }\r\n  }\r\n}\r\n```\r\n\r\nIn the above example `Subscription.request` is scheduled to run asynchronously, but then individual `onNext` events in a batch aren't. And because `onNext` can only happen in response to a `request` call, that must be asynchronous, then the publisher either has to block its loop, waiting for `request` to happen, or it has to schedule the batch of `onNext` to happen asynchronously after `request`.\r\n\r\nI think some examples would do well to shed light on such issues.","created_at":"2014-04-24T17:56:53Z","updated_at":"2014-04-24T17:56:53Z","html_url":"https://github.com/reactive-streams/reactive-streams/pull/37#discussion_r11961782","pull_request_url":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/37","_links":{"self":{"href":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/comments/11961782"},"html":{"href":"https://github.com/reactive-streams/reactive-streams/pull/37#discussion_r11961782"},"pull_request":{"href":"https://api.github.com/repos/reactive-streams/reactive-streams/pulls/37"}}}}