{"action":"created","comment":{"url":"https://api.github.com/repos/akka/akka/pulls/comments/46044655","id":46044655,"diff_hunk":"@@ -9,6 +9,201 @@ is sometimes necessary to define new transformation stages either because some f\n stock operations, or for performance reasons. In this part we show how to build custom processing stages and graph\n junctions of various kinds.\n \n+Custom processing with GraphStage\n+=================================\n+\n+The :class:`GraphStage` abstraction can be used to create arbitrary graph processing stages with any number of input\n+or output ports. It is a counterpart of the ``FlowGraph.create()`` method which creates new stream processing\n+stages out from others. Where :class:`GraphStage` differs is that it creates a stage that is itself not divisible into\n+smaller ones, and allows state to be maintained inside it in a safe way.\n+\n+As a first motivating example, we will build a new :class:`Source` that will simply emit numbers from 1 until it is\n+cancelled. To start, we need to define the \"interface\" of our stage, which is called *shape* in Akka Streams terminology\n+(this is explained in more detail in the section :ref:`composition-java`).\n+\n+\n+.. includecode:: ../../../akka-samples/akka-docs-java-lambda/src/test/java/docs/stream/GraphStageDocTest.java#simple-source\n+\n+As you see, in itself the :class:`GraphStage` only defines the ports of this stage and a shape that contains the ports.\n+It also have a user implemented method called ``createLogic``. If you recall, stages are reusable in multiple\n+materializations, each resulting in a different executing entity. In the case of :class:`GraphStage` the actual running\n+logic is modeled as an instance of a :class:`GraphStageLogic` which will be created by the materializer by calling\n+the ``createLogic`` method.\n+\n+In order to emit from a :class:`Source` in a backpressured stream one needs first to have demand from the downstream.\n+To receive the necessary events one needs to register a subclass of :class:`AbstractOutHandler` to the output port\n+(:class:`Outlet`). This handler will receive events related to the lifecycle of the port. In our case we need to\n+override ``onPull()`` which indicates that we are free to emit a single element. There is another callback,\n+``onDownstreamFinish()`` which is called if the downstream cancelled. Since the default behavior of that callback is\n+to stop the stage, we don't need to override it. In the ``onPull`` callback we simply emit the next number.\n+\n+Instances of the above :class:`GraphStage` are subclasses of ``Graph<SourceShape<Int>,Unit>`` which means\n+that they are already usable in many situations, but do not provide the DSL methods we usually have for other\n+:class:`Source` s. In order to convert this :class:`Graph` to a proper :class:`Source` we need to wrap it using\n+``Source.fromGraph`` (see :ref:`composition-java` for more details about graphs and DSLs). Now we can use the\n+source as any other built-in one:\n+\n+.. includecode:: ../../../akka-samples/akka-docs-java-lambda/src/test/java/docs/stream/GraphStageDocTest.java#simple-source-usage\n+\n+Port states, AbstractInHandler and AbstractOutHandler\n+-----------------------------------------------------\n+\n+In order to interact with a port (:class:`Inlet` or :class:`Outlet`) of the stage we need to be able to receive events\n+and generate new events belonging to the port. From the :class:`GraphStageLogic` the following operations are available\n+on an output port:\n+\n+* ``push(out,elem)`` pushes an element to the output port. Only possible after the port has been pulled by downstream.\n+* ``complete(out)`` closes the output port normally.\n+* ``fail(out,exception)`` closes the port with a failure signal.\n+\n+\n+The events corresponding to an *output* port can be received in an :class:`AbstractOutHandler` instance registered to the\n+output port using ``setHandler(out,handler)``. This handler has two callbacks:\n+\n+* ``onPull()`` is called when the output port is ready to emit the next element, ``push(out, elem)`` is now allowed\n+  to be called on this port.\n+* ``onDownstreamFinish()`` is called once the downstream has cancelled and no longer allows messages to be pushed to it.\n+  No more ``onPull()`` will arrive after this event. If not overridden this will default to stopping the stage.\n+\n+Also, there are two query methods available for output ports:\n+\n+* ``isAvailable(out)`` returns true if the port can be pushed.\n+* ``isClosed`` returns true if the port is closed. At this point the port can not be pushed and will not be pulled anymore.\n+\n+The relationship of the above operations, events and queries are summarized in the state machine below. Green shows\n+the initial state while orange indicates the end state. If an operation is not listed for a state, then it is invalid\n+to call it while the port is in that state. If an event is not listed for a state, then that event cannot happen\n+in that state.\n+\n+|\n+\n+.. image:: ../images/outport_transitions.png\n+:align: center\n+\n+|\n+\n+The following operations are available for *input* ports:\n+\n+* ``pull(in)`` requests a new element from an input port. This is only possible after the port has been pushed by upstream.\n+* ``grab(in)`` acquires the element that has been received during an ``onPush()``. It cannot be called again until the\n+  port is pushed again by the upstream.\n+* ``cancel(in)`` closes the input port.\n+\n+The events corresponding to an *input* port can be received in an :class:`AbstractInHandler` instance registered to the\n+input port using ``setHandler(in, handler)``. This handler has three callbacks:\n+\n+* ``onPush()`` is called when the output port has now a new element. Now it is possible to aquire this element using\n+  ``grab()`` and/or call ``pull(in)`` on the port to request the next element. It is not mandatory to grab the\n+  element, but if it is pulled while the element has not been grabbed it will drop the buffered element.\n+* ``onUpstreamFinish()`` is called once the upstream has completed and no longer can be pulled for new elements.\n+  No more ``onPush()`` will arrive after this event. If not overridden this will default to stopping the stage.\n+* ``onUpstreamFailure()`` is called if the upstream failed with an exception and no longer can be pulled for new elements.\n+  No more ``onPush()`` will arrive after this event. If not overridden this will default to failing the stage.\n+\n+Also, there are three query methods available for input ports:\n+\n+* ``isAvailable(out)`` returns true if the port can be grabbed.\n+* ``hasBeenPulled(out)`` returns true if the port has been already pulled. Calling ``pull(in)`` in this state is illegal.\n+* ``isClosed`` returns true if the port is closed. At this point the port can not be pulled and will not be pushed anymore.\n+\n+The relationship of the above operations, events and queries are summarized in the state machine below. Green shows\n+the initial state while orange indicates the end state. If an operation is not listed for a state, then it is invalid\n+to call it while the port is in that state. If an event is not listed for a state, then that event cannot happen\n+in that state.\n+\n+|\n+\n+.. image:: ../images/inport_transitions.png\n+:align: center\n+\n+|\n+\n+Finally, there are two methods available for convenience to complete the stage and all of its ports:\n+\n+* ``completeStage()`` is equivalent to closing all output ports and cancelling all input ports.\n+* ``failStage(exception)`` is equivalent to failing all output ports and cancelling all input ports.\n+\n+\n+Completion\n+----------\n+\n+**This section is a stub and will be extended in the next release**\n+\n+Stages by default automatically stop once all of their ports (input and output) have been closed externally or internally.\n+It is possible to opt out from this behavior by overriding ``keepGoingAfterAllPortsClosed`` and returning true in\n+the :class:`GraphStageLogic` implementation. In this case the stage can be explicitly closed by calling ``completeStage()``\n+or ``failStage(exception)``. This feature has the risk of leaking streams and actors, therefore it should be used","path":"akka-docs-dev/rst/java/stream-customize.rst","position":127,"original_position":127,"commit_id":"24604c4b0f4f21642166c7bb68fe240bec4cdbdd","original_commit_id":"24604c4b0f4f21642166c7bb68fe240bec4cdbdd","user":{"login":"rkuhn","id":470469,"avatar_url":"https://avatars.githubusercontent.com/u/470469?v=3","gravatar_id":"","url":"https://api.github.com/users/rkuhn","html_url":"https://github.com/rkuhn","followers_url":"https://api.github.com/users/rkuhn/followers","following_url":"https://api.github.com/users/rkuhn/following{/other_user}","gists_url":"https://api.github.com/users/rkuhn/gists{/gist_id}","starred_url":"https://api.github.com/users/rkuhn/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rkuhn/subscriptions","organizations_url":"https://api.github.com/users/rkuhn/orgs","repos_url":"https://api.github.com/users/rkuhn/repos","events_url":"https://api.github.com/users/rkuhn/events{/privacy}","received_events_url":"https://api.github.com/users/rkuhn/received_events","type":"User","site_admin":false},"body":"or better “carries”","created_at":"2015-11-27T12:59:45Z","updated_at":"2015-11-27T12:59:45Z","html_url":"https://github.com/akka/akka/pull/19026#discussion_r46044655","pull_request_url":"https://api.github.com/repos/akka/akka/pulls/19026","_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/comments/46044655"},"html":{"href":"https://github.com/akka/akka/pull/19026#discussion_r46044655"},"pull_request":{"href":"https://api.github.com/repos/akka/akka/pulls/19026"}}},"pull_request":{"url":"https://api.github.com/repos/akka/akka/pulls/19026","id":51926404,"html_url":"https://github.com/akka/akka/pull/19026","diff_url":"https://github.com/akka/akka/pull/19026.diff","patch_url":"https://github.com/akka/akka/pull/19026.patch","issue_url":"https://api.github.com/repos/akka/akka/issues/19026","number":19026,"state":"open","locked":false,"title":"+str, doc: Documentation for GraphStage","user":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?v=3","gravatar_id":"","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"body":"With Java docs and stub sections for detailed subsections to be filled in in next milestone.","created_at":"2015-11-26T15:46:17Z","updated_at":"2015-11-27T12:59:45Z","closed_at":null,"merged_at":null,"merge_commit_sha":"6b41e814f7882528c9c9fc33fec799498b1ee78c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/akka/akka/pulls/19026/commits","review_comments_url":"https://api.github.com/repos/akka/akka/pulls/19026/comments","review_comment_url":"https://api.github.com/repos/akka/akka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/akka/akka/issues/19026/comments","statuses_url":"https://api.github.com/repos/akka/akka/statuses/24604c4b0f4f21642166c7bb68fe240bec4cdbdd","head":{"label":"drewhk:wip-document-graphstage-drewhk","ref":"wip-document-graphstage-drewhk","sha":"24604c4b0f4f21642166c7bb68fe240bec4cdbdd","user":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?v=3","gravatar_id":"","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"repo":{"id":4322312,"name":"akka","full_name":"drewhk/akka","owner":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?v=3","gravatar_id":"","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/drewhk/akka","description":"Akka Project","fork":true,"url":"https://api.github.com/repos/drewhk/akka","forks_url":"https://api.github.com/repos/drewhk/akka/forks","keys_url":"https://api.github.com/repos/drewhk/akka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/drewhk/akka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/drewhk/akka/teams","hooks_url":"https://api.github.com/repos/drewhk/akka/hooks","issue_events_url":"https://api.github.com/repos/drewhk/akka/issues/events{/number}","events_url":"https://api.github.com/repos/drewhk/akka/events","assignees_url":"https://api.github.com/repos/drewhk/akka/assignees{/user}","branches_url":"https://api.github.com/repos/drewhk/akka/branches{/branch}","tags_url":"https://api.github.com/repos/drewhk/akka/tags","blobs_url":"https://api.github.com/repos/drewhk/akka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/drewhk/akka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/drewhk/akka/git/refs{/sha}","trees_url":"https://api.github.com/repos/drewhk/akka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/drewhk/akka/statuses/{sha}","languages_url":"https://api.github.com/repos/drewhk/akka/languages","stargazers_url":"https://api.github.com/repos/drewhk/akka/stargazers","contributors_url":"https://api.github.com/repos/drewhk/akka/contributors","subscribers_url":"https://api.github.com/repos/drewhk/akka/subscribers","subscription_url":"https://api.github.com/repos/drewhk/akka/subscription","commits_url":"https://api.github.com/repos/drewhk/akka/commits{/sha}","git_commits_url":"https://api.github.com/repos/drewhk/akka/git/commits{/sha}","comments_url":"https://api.github.com/repos/drewhk/akka/comments{/number}","issue_comment_url":"https://api.github.com/repos/drewhk/akka/issues/comments{/number}","contents_url":"https://api.github.com/repos/drewhk/akka/contents/{+path}","compare_url":"https://api.github.com/repos/drewhk/akka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/drewhk/akka/merges","archive_url":"https://api.github.com/repos/drewhk/akka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/drewhk/akka/downloads","issues_url":"https://api.github.com/repos/drewhk/akka/issues{/number}","pulls_url":"https://api.github.com/repos/drewhk/akka/pulls{/number}","milestones_url":"https://api.github.com/repos/drewhk/akka/milestones{/number}","notifications_url":"https://api.github.com/repos/drewhk/akka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/drewhk/akka/labels{/name}","releases_url":"https://api.github.com/repos/drewhk/akka/releases{/id}","created_at":"2012-05-14T09:57:34Z","updated_at":"2015-02-17T12:40:08Z","pushed_at":"2015-11-27T11:04:50Z","git_url":"git://github.com/drewhk/akka.git","ssh_url":"git@github.com:drewhk/akka.git","clone_url":"https://github.com/drewhk/akka.git","svn_url":"https://github.com/drewhk/akka","homepage":"http://akka.io","size":168337,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"akka:release-2.3-dev","ref":"release-2.3-dev","sha":"45f52362752e862c801164a4ae4131f5c65dc083","user":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"repo":{"id":130013,"name":"akka","full_name":"akka/akka","owner":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/akka/akka","description":"Akka Project","fork":false,"url":"https://api.github.com/repos/akka/akka","forks_url":"https://api.github.com/repos/akka/akka/forks","keys_url":"https://api.github.com/repos/akka/akka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akka/akka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akka/akka/teams","hooks_url":"https://api.github.com/repos/akka/akka/hooks","issue_events_url":"https://api.github.com/repos/akka/akka/issues/events{/number}","events_url":"https://api.github.com/repos/akka/akka/events","assignees_url":"https://api.github.com/repos/akka/akka/assignees{/user}","branches_url":"https://api.github.com/repos/akka/akka/branches{/branch}","tags_url":"https://api.github.com/repos/akka/akka/tags","blobs_url":"https://api.github.com/repos/akka/akka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akka/akka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akka/akka/git/refs{/sha}","trees_url":"https://api.github.com/repos/akka/akka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akka/akka/statuses/{sha}","languages_url":"https://api.github.com/repos/akka/akka/languages","stargazers_url":"https://api.github.com/repos/akka/akka/stargazers","contributors_url":"https://api.github.com/repos/akka/akka/contributors","subscribers_url":"https://api.github.com/repos/akka/akka/subscribers","subscription_url":"https://api.github.com/repos/akka/akka/subscription","commits_url":"https://api.github.com/repos/akka/akka/commits{/sha}","git_commits_url":"https://api.github.com/repos/akka/akka/git/commits{/sha}","comments_url":"https://api.github.com/repos/akka/akka/comments{/number}","issue_comment_url":"https://api.github.com/repos/akka/akka/issues/comments{/number}","contents_url":"https://api.github.com/repos/akka/akka/contents/{+path}","compare_url":"https://api.github.com/repos/akka/akka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akka/akka/merges","archive_url":"https://api.github.com/repos/akka/akka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akka/akka/downloads","issues_url":"https://api.github.com/repos/akka/akka/issues{/number}","pulls_url":"https://api.github.com/repos/akka/akka/pulls{/number}","milestones_url":"https://api.github.com/repos/akka/akka/milestones{/number}","notifications_url":"https://api.github.com/repos/akka/akka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akka/akka/labels{/name}","releases_url":"https://api.github.com/repos/akka/akka/releases{/id}","created_at":"2009-02-16T12:51:54Z","updated_at":"2015-11-27T12:48:12Z","pushed_at":"2015-11-27T12:39:30Z","git_url":"git://github.com/akka/akka.git","ssh_url":"git@github.com:akka/akka.git","clone_url":"https://github.com/akka/akka.git","svn_url":"https://github.com/akka/akka","homepage":"http://akka.io","size":177745,"stargazers_count":4758,"watchers_count":4758,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1423,"mirror_url":null,"open_issues_count":776,"forks":1423,"open_issues":776,"watchers":4758,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/19026"},"html":{"href":"https://github.com/akka/akka/pull/19026"},"issue":{"href":"https://api.github.com/repos/akka/akka/issues/19026"},"comments":{"href":"https://api.github.com/repos/akka/akka/issues/19026/comments"},"review_comments":{"href":"https://api.github.com/repos/akka/akka/pulls/19026/comments"},"review_comment":{"href":"https://api.github.com/repos/akka/akka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/akka/akka/pulls/19026/commits"},"statuses":{"href":"https://api.github.com/repos/akka/akka/statuses/24604c4b0f4f21642166c7bb68fe240bec4cdbdd"}}}}