raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Merge pull request #1129 from akka/wip-IO

Browse source 
Wip io
This commit is contained in:
Roland Kuhn 2013-02-13 08:29:51 -08:00
commit 12de575139
43 changed files with 11621 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

1
akka-docs/rst/dev/index.rst
View file

@ -6,6 +6,7 @@ Information for Developers
building-akka
multi-jvm-testing
io-layer
developer-guidelines
documentation
team

116
akka-docs/rst/dev/io-layer.rst Normal file
View file

@ -0,0 +1,116 @@
.. _io-layer:
################
I/O Layer Design
################
The ``akka.io`` package has been developed in collaboration between the Akka
and `spray.io`_ teams. Its design incorporates the experiences with the
``spray-io`` module along with improvements that were jointly developed for
more general consumption as an actor-based service.
Requirements
============
In order to form a general and extensible IO layer basis for a wide range of
applications, with Akka remoting and spray HTTP being the initial ones, the
following requirements were established as key drivers for the design:
* scalability to millions of concurrent connections
* lowest possible latency in getting data from an input channel into the
target actor’s mailbox
* maximal throughput
* optional back-pressure in both directions (i.e. throttling local senders as
well as allowing local readers to throttle remote senders, where allowed by
the protocol)
* a purely actor-based API with immutable data representation
* extensibility for integrating new transports by way of a very lean SPI; the
goal is to not force I/O mechanisms into a lowest common denominator but
instead allow completely protocol-specific user-level APIs.
Basic Architecture
==================
Each transport implementation will be made available as a separate Akka
extension, offering an :class:`ActorRef` representing the initial point of
contact for client code. This "manager" accepts requests for establishing a
communications channel (e.g. connect or listen on a TCP socket). Each
communications channel is represented by one dedicated actor, which is exposed
to client code for all interaction with this channel over its entire lifetime.
The central element of the implementation is the transport-specific “selector”
actor; in the case of TCP this would wrap a :class:`java.nio.channels.Selector`.
The channel actors register their interest in readability or writability of
their channel by sending corresponding messages to their assigned selector
actor. However, the actual channel reading and writing is performed by the
channel actors themselves, which frees the selector actors from time-consuming
tasks and thereby ensures low latency. The selector actor's only responsibility
is the management of the underlying selector's key set and the actual select
operation, which is the only operation to typically block.
The assignment of channels to selectors is performed by the manager actor and
remains unchanged for the entire lifetime of a channel. Thereby the management
actor "stripes" new channels across one or more selector actors based on some
implementation-specific distribution logic. This logic may be delegated (in
part) to the selectors actors, which could, for example, choose to reject the
assignment of a new channel when they consider themselves to be at capacity.
The manager actor creates (and therefore supervises) the selector actors, which
in turn create and supervise their channel actors. The actor hierarchy of one
single transport implementation therefore consists of three distinct actor
levels, with the management actor at the top-, the channel actors at the leaf-
and the selector actors at the mid-level.
Back-pressure for output is enabled by allowing the user to specify within its
:class:`Write` messages whether it wants to receive an acknowledgement for
enqueuing that write to the O/S kernel. Back-pressure for input is enabled by
sending the channel actor a message which temporarily disables read interest
for the channel until reading is re-enabled with a corresponding resume command.
In the case of transports with flow control—like TCP—the act of not
consuming data at the receiving end (thereby causing them to remain in the
kernels read buffers) is propagated back to the sender, linking these two
mechanisms across the network.
Design Benefits
===============
Staying within the actor model for the whole implementation allows us to remove
the need for explicit thread handling logic, and it also means that there are
no locks involved (besides those which are part of the underlying transport
library). Writing only actor code results in a cleaner implementation,
while Akka’s efficient actor messaging does not impose a high tax for this
benefit. In fact the event-based nature of I/O maps so well to the actor model
that we expect clear performance and especially scalability benefits over
traditional solutions with explicit thread management and synchronization.
Another benefit of supervision hierarchies is that clean-up of resources comes
naturally: shutting down a selector actor will automatically clean up all
channel actors, allowing proper closing of the channels and sending the
appropriate messages to user-level client actors. DeathWatch allows the channel
actors to notice the demise of their user-level handler actors and terminate in
an orderly fashion in that case as well; this naturally reduces the chances of
leaking open channels.
The choice of using :class:`ActorRef` for exposing all functionality entails
that these references can be distributed or delegated freely and in general
handled as the user sees fit, including the use of remoting and life-cycle
monitoring (just to name two).
How to go about Adding a New Transport
======================================
The best start is to study the TCP reference implementation to get a good grip
on the basic working principle and then design an implementation, which is
similar in spirit, but adapted to the new protocol in question. There are vast
differences between I/O mechanisms (e.g. compare file I/O to a message broker)
and the goal of this I/O layer is explicitly **not** to shoehorn all of them
into a uniform API, which is why only the basic architecture ideas are
documented here.
.. _spray.io: http://spray.io

7831
akka-docs/rst/images/tcp-message-protocol.graffle Normal file
View file

File diff suppressed because it is too large Load diff

3
akka-docs/rst/images/tcp-message-protocol_bind-connect-maxed-out.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 6.7 KiB

3
akka-docs/rst/images/tcp-message-protocol_binding.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 7.7 KiB

3
akka-docs/rst/images/tcp-message-protocol_closing.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 5 KiB

3
akka-docs/rst/images/tcp-message-protocol_establishing-incoming.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 9 KiB

3
akka-docs/rst/images/tcp-message-protocol_establishing-outgoing.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 9 KiB

3
akka-docs/rst/images/tcp-message-protocol_noticing-close.svg Normal file
View file

@ -0,0 +1,3 @@
<?xml version="1.0"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xl="http://www.w3.org/1999/xlink" version="1.1" viewBox="25 1593 490 155" width="490pt" height="155pt"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:date>2013-01-23 14:28Z</dc:date><!-- Produced by OmniGraffle Professional 5.3.6 --></metadata><defs><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="NegativeControls_Marker" viewBox="-1 -5 2 10" markerWidth="2" markerHeight="10" color="black"><g><line x1="0" y1="-4" x2="0" y2="4" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="FilledArrow_Marker" viewBox="-1 -4 10 8" markerWidth="10" markerHeight="8" color="black"><g><path d="M 8 0 L 0 -3 L 0 3 Z" fill="currentColor" stroke="currentColor" stroke-width="1"/></g></marker><font-face font-family="Helvetica" font-size="8" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="532.22656" cap-height="719.72656" ascent="770.01953" descent="-229.98047" font-weight="bold"><font-face-src><font-face-name name="Helvetica-Bold"/></font-face-src></font-face></defs><g stroke="none" stroke-opacity="1" stroke-dasharray="none" fill="none" fill-opacity="1"><title>Canvas 1</title><g><title>Layer 1</title><rect x="36" y="1620" width="468" height="18" fill="#ccc"/><rect x="36" y="1620" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 1623)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="176.46582" y="10" textLength="105.06836">Connection Established</tspan></text><text transform="translate(90 1650)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="18.490234" y="10" textLength="35.019531">Handler</tspan></text><text transform="translate(360 1650)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="20.923828" y="10" textLength="6.1083984">T</tspan><tspan font-family="Helvetica" font-size="10" font-weight="500" x="25.923828" y="10" textLength="61.152344">cpConnection</tspan></text><line x1="126" y1="1665" x2="126" y2="1719" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="414" y1="1665" x2="414" y2="1698.5" marker-end="url(#NegativeControls_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="414" y1="1683" x2="135.89999" y2="1683" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(135 1673.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="4.4414062" y="8" textLength="50.246094">Closed / Confi</tspan><tspan font-family="Helvetica" font-size="8" font-weight="500" x="54.6875" y="8" textLength="49.796875">rmedClosed / </tspan><tspan font-family="Helvetica" font-size="8" font-weight="500" x="104.046875" y="8" textLength="152.51172">Aborted / PeerClosed / ErrorClosed(cause)</tspan></text><rect x="36" y="1719" width="468" height="18" fill="#ccc"/><rect x="36" y="1719" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 1722)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="168.68506" y="10" textLength="120.62988">No Connection Established</tspan></text><text transform="translate(36 1605)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="bold" x="0" y="10" textLength="183.35449">Noticing that a Connection was closed</tspan></text></g></g></svg>
Side by side

After

Width:  |  Height:  |  Size: 4.4 KiB

3
akka-docs/rst/images/tcp-message-protocol_receiving.svg Normal file
View file

@ -0,0 +1,3 @@
<?xml version="1.0"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xl="http://www.w3.org/1999/xlink" version="1.1" viewBox="25 972 490 173" width="490pt" height="173pt"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:date>2013-01-23 14:28Z</dc:date><!-- Produced by OmniGraffle Professional 5.3.6 --></metadata><defs><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="FilledArrow_Marker" viewBox="-1 -4 10 8" markerWidth="10" markerHeight="8" color="black"><g><path d="M 8 0 L 0 -3 L 0 3 Z" fill="currentColor" stroke="currentColor" stroke-width="1"/></g></marker><font-face font-family="Helvetica" font-size="8" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="532.22656" cap-height="719.72656" ascent="770.01953" descent="-229.98047" font-weight="bold"><font-face-src><font-face-name name="Helvetica-Bold"/></font-face-src></font-face></defs><g stroke="none" stroke-opacity="1" stroke-dasharray="none" fill="none" fill-opacity="1"><title>Canvas 1</title><g><title>Layer 1</title><rect x="36" y="999" width="468" height="18" fill="#ccc"/><rect x="36" y="999" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 1002)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="176.46582" y="10" textLength="105.06836">Connection Established</tspan></text><text transform="translate(234 1029)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="9.876953" y="10" textLength="6.1083984">T</tspan><tspan font-family="Helvetica" font-size="10" font-weight="500" x="14.876953" y="10" textLength="47.246094">cpSelector</tspan></text><text transform="translate(90 1029)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="18.490234" y="10" textLength="35.019531">Handler</tspan></text><text transform="translate(360 1029)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="20.923828" y="10" textLength="6.1083984">T</tspan><tspan font-family="Helvetica" font-size="10" font-weight="500" x="25.923828" y="10" textLength="61.152344">cpConnection</tspan></text><line x1="126" y1="1044" x2="126" y2="1116" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="270" y1="1044" x2="270" y2="1116" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="414" y1="1044" x2="414" y2="1116" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="270" y1="1062" x2="404.09998" y2="1062" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(333 1052.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="3.9746094" y="8" textLength="64.05078">ChannelReadable</tspan></text><line x1="414" y1="1098" x2="279.90002" y2="1098" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(279 1088.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="4.0976562" y="8" textLength="45.804688">ReadInterest</tspan></text><path d="M 414 1080 L 275.5 1080 C 275.5 1074.5 264.5 1074.5 264.5 1080 L 135.89999 1080" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(135 1070.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="4.375" y="8" textLength="54.25">Received(data)</tspan></text><rect x="36" y="1116" width="468" height="18" fill="#ccc"/><rect x="36" y="1116" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 1119)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="176.46582" y="10" textLength="105.06836">Connection Established</tspan></text><text transform="translate(36 984)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="bold" x="0" y="10" textLength="162.80762">Receiving Data from a Connection</tspan></text></g></g></svg>
Side by side

After

Width:  |  Height:  |  Size: 4.9 KiB

3
akka-docs/rst/images/tcp-message-protocol_unbinding.svg Normal file
View file

@ -0,0 +1,3 @@
<?xml version="1.0"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xl="http://www.w3.org/1999/xlink" version="1.1" viewBox="25 243 490 173" width="490pt" height="173pt"><metadata xmlns:dc="http://purl.org/dc/elements/1.1/"><dc:date>2013-01-22 22:05Z</dc:date><!-- Produced by OmniGraffle Professional 5.3.6 --></metadata><defs><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><font-face font-family="Helvetica" font-size="8" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="522.94922" cap-height="717.28516" ascent="770.01953" descent="-229.98047" font-weight="500"><font-face-src><font-face-name name="Helvetica"/></font-face-src></font-face><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="FilledArrow_Marker" viewBox="-1 -4 10 8" markerWidth="10" markerHeight="8" color="black"><g><path d="M 8 0 L 0 -3 L 0 3 Z" fill="currentColor" stroke="currentColor" stroke-width="1"/></g></marker><marker orient="auto" overflow="visible" markerUnits="strokeWidth" id="NegativeControls_Marker" viewBox="-1 -5 2 10" markerWidth="2" markerHeight="10" color="black"><g><line x1="0" y1="-4" x2="0" y2="4" fill="none" stroke="currentColor" stroke-width="1"/></g></marker><font-face font-family="Helvetica" font-size="10" units-per-em="1000" underline-position="-75.683594" underline-thickness="49.316406" slope="0" x-height="532.22656" cap-height="719.72656" ascent="770.01953" descent="-229.98047" font-weight="bold"><font-face-src><font-face-name name="Helvetica-Bold"/></font-face-src></font-face></defs><g stroke="none" stroke-opacity="1" stroke-dasharray="none" fill="none" fill-opacity="1"><title>Canvas 1</title><g><title>Layer 1</title><text transform="translate(126 300)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="12.657715" y="10" textLength="23.891602">User </tspan><tspan font-family="Helvetica" font-size="10" font-weight="500" x="36.002441" y="10" textLength="23.339844">Actor</tspan></text><text transform="translate(171 341.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="6.263672" y="8" textLength="32.472656">Unbound</tspan></text><line x1="162" y1="315" x2="162" y2="387" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="162" y1="333" x2="368.09998" y2="333" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(333 323.5)" fill="black"><tspan font-family="Helvetica" font-size="8" font-weight="500" x="5.3242188" y="8" textLength="25.351562">Unbind</tspan></text><line x1="378" y1="351" x2="171.90001" y2="351" marker-end="url(#FilledArrow_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><line x1="378" y1="315" x2="378" y2="366.5" marker-end="url(#NegativeControls_Marker)" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><rect x="36" y="270" width="468" height="18" fill="#ccc"/><rect x="36" y="270" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 273)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="198.42627" y="10" textLength="61.14746">Server Bound</tspan></text><text transform="translate(342 300)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="10.4311523" y="10" textLength="6.1083984">T</tspan><tspan font-family="Helvetica" font-size="10" font-weight="500" x="15.431152" y="10" textLength="46.137695">cpListener</tspan></text><rect x="36" y="387" width="468" height="18" fill="#ccc"/><rect x="36" y="387" width="468" height="18" stroke="black" stroke-linecap="round" stroke-linejoin="round" stroke-width="1"/><text transform="translate(41 390)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="500" x="192.58887" y="10" textLength="72.822266">Server Unbound</tspan></text><text transform="translate(36 255)" fill="black"><tspan font-family="Helvetica" font-size="10" font-weight="bold" x="0" y="10" textLength="91.68457">Unbinding a Server</tspan></text></g></g></svg>
Side by side

After

Width:  |  Height:  |  Size: 4.5 KiB

3
akka-docs/rst/images/tcp-message-protocol_writing.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 7 KiB

248
akka-docs/rst/scala/io-old.rst Normal file
View file

@ -0,0 +1,248 @@
.. _io-scala-old:
.. warning::
This is the documentation of the old IO implementation that is considered now deprecated. Please take a look
at new IO API: :ref:`io-scala`
IO (Scala)
==========
Introduction
------------
This documentation is in progress and some sections may be incomplete. More will be coming.
Components
----------
ByteString
^^^^^^^^^^
A primary goal of Akka's IO support is to only communicate between actors with immutable objects. When dealing with network IO on the jvm ``Array[Byte]`` and ``ByteBuffer`` are commonly used to represent collections of ``Byte``\s, but they are mutable. Scala's collection library also lacks a suitably efficient immutable collection for ``Byte``\s. Being able to safely and efficiently move ``Byte``\s around is very important for this IO support, so ``ByteString`` was developed.
``ByteString`` is a `Rope-like <http://en.wikipedia.org/wiki/Rope_(computer_science)>`_ data structure that is immutable and efficient. When 2 ``ByteString``\s are concatenated together they are both stored within the resulting ``ByteString`` instead of copying both to a new ``Array``. Operations such as ``drop`` and ``take`` return ``ByteString``\s that still reference the original ``Array``, but just change the offset and length that is visible. Great care has also been taken to make sure that the internal ``Array`` cannot be modified. Whenever a potentially unsafe ``Array`` is used to create a new ``ByteString`` a defensive copy is created. If you require a ``ByteString`` that only blocks a much memory as necessary for it's content, use the ``compact`` method to get a ``CompactByteString`` instance. If the ``ByteString`` represented only a slice of the original array, this will result in copying all bytes in that slice.
``ByteString`` inherits all methods from ``IndexedSeq``, and it also has some new ones. For more information, look up the ``akka.util.ByteString`` class and it's companion object in the ScalaDoc.
``ByteString`` also comes with it's own optimized builder and iterator classes ``ByteStringBuilder`` and ``ByteIterator`` which provides special features in addition to the standard builder / iterator methods:
Compatibility with java.io
..........................
A ``ByteStringBuilder`` can be wrapped in a `java.io.OutputStream` via the ``asOutputStream`` method. Likewise, ``ByteIterator`` can we wrapped in a ``java.io.InputStream`` via ``asInputStream``. Using these, ``akka.io`` applications can integrate legacy code based on ``java.io`` streams.
Encoding and decoding of binary data
....................................
``ByteStringBuilder`` and ``ByteIterator`` support encoding and decoding of binary data. As an example, consider a stream of binary data frames with the following format:
.. code-block:: text
frameLen: Int
n: Int
m: Int
n times {
a: Short
b: Long
}
data: m times Double
In this example, the data is to be stored in arrays of ``a``, ``b`` and ``data``.
Decoding of such frames can be efficiently implemented in the following fashion:
.. includecode:: code/docs/io/BinaryCoding.scala
:include: decoding
This implementation naturally follows the example data format. In a true Scala application, one might, of course, want use specialized immutable Short/Long/Double containers instead of mutable Arrays.
After extracting data from a ``ByteIterator``, the remaining content can also be turned back into a ``ByteString`` using the ``toSeq`` method
.. includecode:: code/docs/io/BinaryCoding.scala
:include: rest-to-seq
with no copying from bytes to rest involved. In general, conversions from ByteString to ByteIterator and vice versa are O(1) for non-chunked ByteStrings and (at worst) O(nChunks) for chunked ByteStrings.
Encoding of data also is very natural, using ``ByteStringBuilder``
.. includecode:: code/docs/io/BinaryCoding.scala
:include: encoding
The encoded data then can be sent over socket (see ``IOManager``):
.. includecode:: code/docs/io/BinaryCoding.scala
:include: sending
IO.Handle
^^^^^^^^^
``IO.Handle`` is an immutable reference to a Java NIO ``Channel``. Passing mutable ``Channel``\s between ``Actor``\s could lead to unsafe behavior, so instead subclasses of the ``IO.Handle`` trait are used. Currently there are 2 concrete subclasses: ``IO.SocketHandle`` (representing a ``SocketChannel``) and ``IO.ServerHandle`` (representing a ``ServerSocketChannel``).
IOManager
^^^^^^^^^
The ``IOManager`` takes care of the low level IO details. Each ``ActorSystem`` has it's own ``IOManager``, which can be accessed calling ``IOManager(system: ActorSystem)``. ``Actor``\s communicate with the ``IOManager`` with specific messages. The messages sent from an ``Actor`` to the ``IOManager`` are handled automatically when using certain methods and the messages sent from an ``IOManager`` are handled within an ``Actor``\'s ``receive`` method.
Connecting to a remote host:
.. code-block:: scala
val address = new InetSocketAddress("remotehost", 80)
val socket = IOManager(actorSystem).connect(address)
.. code-block:: scala
val socket = IOManager(actorSystem).connect("remotehost", 80)
Creating a server:
.. code-block:: scala
val address = new InetSocketAddress("localhost", 80)
val serverSocket = IOManager(actorSystem).listen(address)
.. code-block:: scala
val serverSocket = IOManager(actorSystem).listen("localhost", 80)
Receiving messages from the ``IOManager``:
.. code-block:: scala
def receive = {
case IO.Listening(server, address) =>
println("The server is listening on socket " + address)
case IO.Connected(socket, address) =>
println("Successfully connected to " + address)
case IO.NewClient(server) =>
println("New incoming connection on server")
val socket = server.accept()
println("Writing to new client socket")
socket.write(bytes)
println("Closing socket")
socket.close()
case IO.Read(socket, bytes) =>
println("Received incoming data from socket")
case IO.Closed(socket: IO.SocketHandle, cause) =>
println("Socket has closed, cause: " + cause)
case IO.Closed(server: IO.ServerHandle, cause) =>
println("Server socket has closed, cause: " + cause)
}
IO.Iteratee
^^^^^^^^^^^
Included with Akka's IO support is a basic implementation of ``Iteratee``\s. ``Iteratee``\s are an effective way of handling a stream of data without needing to wait for all the data to arrive. This is especially useful when dealing with non blocking IO since we will usually receive data in chunks which may not include enough information to process, or it may contain much more data than we currently need.
This ``Iteratee`` implementation is much more basic than what is usually found. There is only support for ``ByteString`` input, and enumerators aren't used. The reason for this limited implementation is to reduce the amount of explicit type signatures needed and to keep things simple. It is important to note that Akka's ``Iteratee``\s are completely optional, incoming data can be handled in any way, including other ``Iteratee`` libraries.
``Iteratee``\s work by processing the data that it is given and returning either the result (with any unused input) or a continuation if more input is needed. They are monadic, so methods like ``flatMap`` can be used to pass the result of an ``Iteratee`` to another.
The basic ``Iteratee``\s included in the IO support can all be found in the ScalaDoc under ``akka.actor.IO``, and some of them are covered in the example below.
Examples
--------
Http Server
^^^^^^^^^^^
This example will create a simple high performance HTTP server. We begin with our imports:
.. includecode:: code/docs/io/HTTPServer.scala
:include: imports
Some commonly used constants:
.. includecode:: code/docs/io/HTTPServer.scala
:include: constants
And case classes to hold the resulting request:
.. includecode:: code/docs/io/HTTPServer.scala
:include: request-class
Now for our first ``Iteratee``. There are 3 main sections of a HTTP request: the request line, the headers, and an optional body. The main request ``Iteratee`` handles each section separately:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request
In the above code ``readRequest`` takes the results of 3 different ``Iteratees`` (``readRequestLine``, ``readHeaders``, ``readBody``) and combines them into a single ``Request`` object. ``readRequestLine`` actually returns a tuple, so we extract it's individual components. ``readBody`` depends on values contained within the header section, so we must pass those to the method.
The request line has 3 parts to it: the HTTP method, the requested URI, and the HTTP version. The parts are separated by a single space, and the entire request line ends with a ``CRLF``.
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request-line
Reading the request method is simple as it is a single string ending in a space. The simple ``Iteratee`` that performs this is ``IO.takeUntil(delimiter: ByteString): Iteratee[ByteString]``. It keeps consuming input until the specified delimiter is found. Reading the HTTP version is also a simple string that ends with a ``CRLF``.
The ``ascii`` method is a helper that takes a ``ByteString`` and parses it as a ``US-ASCII`` ``String``.
Reading the request URI is a bit more complicated because we want to parse the individual components of the URI instead of just returning a simple string:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request-uri
For this example we are only interested in handling absolute paths. To detect if we the URI is an absolute path we use ``IO.peek(length: Int): Iteratee[ByteString]``, which returns a ``ByteString`` of the request length but doesn't actually consume the input. We peek at the next bit of input and see if it matches our ``PATH`` constant (defined above as ``ByteString("/")``). If it doesn't match we throw an error, but for a more robust solution we would want to handle other valid URIs.
Next we handle the path itself:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-path
The ``step`` method is a recursive method that takes a ``List`` of the accumulated path segments. It first checks if the remaining input starts with the ``PATH`` constant, and if it does, it drops that input, and returns the ``readUriPart`` ``Iteratee`` which has it's result added to the path segment accumulator and the ``step`` method is run again.
If after reading in a path segment the next input does not start with a path, we reverse the accumulated segments and return it (dropping the last segment if it is blank).
Following the path we read in the query (if it exists):
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-query
It is much simpler than reading the path since we aren't doing any parsing of the query since there is no standard format of the query string.
Both the path and query used the ``readUriPart`` ``Iteratee``, which is next:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-uri-part
Here we have several ``Set``\s that contain valid characters pulled from the URI spec. The ``readUriPart`` method takes a ``Set`` of valid characters (already mapped to ``Byte``\s) and will continue to match characters until it reaches on that is not part of the ``Set``. If it is a percent encoded character then that is handled as a valid character and processing continues, or else we are done collecting this part of the URI.
Headers are next:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-headers
And if applicable, we read in the message body:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-body
Finally we get to the actual ``Actor``:
.. includecode:: code/docs/io/HTTPServer.scala
:include: actor
And it's companion object:
.. includecode:: code/docs/io/HTTPServer.scala
:include: actor-companion
And the OKResponse:
.. includecode:: code/docs/io/HTTPServer.scala
:include: ok-response
A ``main`` method to start everything up:
.. includecode:: code/docs/io/HTTPServer.scala
:include: main

363
akka-docs/rst/scala/io.rst
View file

@ -1,21 +1,75 @@
.. _io-scala:
IO (Scala)
I/O (Scala)
==========
Introduction
------------
The ``akka.io`` package has been developed in collaboration between the Akka
and `spray.io`_ teams. Its design incorporates the experiences with the
``spray-io`` module along with improvements that were jointly developed for
more general consumption as an actor-based service.
This documentation is in progress and some sections may be incomplete. More will be coming.
Components
----------
.. note::
The old I/O implementation has been deprecated and its documentation has been moved: :ref:`io-scala-old`
Terminology, Concepts
---------------------
The I/O API is completely actor based, meaning that all operations are implemented as message passing instead of
direct method calls. Every I/O driver (TCP, UDP) has a special actor, called *manager* that serves
as the entry point for the API. The manager is accessible through an extension, for example the following code
looks up the TCP manager and returns its ``ActorRef``:
.. code-block:: scala
val tcpManager = IO(Tcp)
For various I/O commands the manager instantiates worker actors that will expose themselves to the user of the
API by replying to the command. For example after a ``Connect`` command sent to the TCP manager the manager creates
an actor representing the TCP connection. All operations related to the given TCP connections can be invoked by sending
messages to the connection actor which announces itself by sending a ``Connected`` message.
DeathWatch and Resource Management
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Worker actors usually need a user-side counterpart actor listening for events (such events could be inbound connections,
incoming bytes or acknowledgements for writes). These worker actors *watch* their listener counterparts, therefore the
resources assigned to them are automatically released when the listener stops. This design makes the API more robust
against resource leaks.
Thanks to the completely actor based approach of the I/O API the opposite direction works as well: a user actor
responsible for handling a connection might watch the connection actor to be notified if it unexpectedly terminates.
Write models (Ack, Nack)
^^^^^^^^^^^^^^^^^^^^^^^^
Basically all of the I/O devices have a maximum throughput which limits the frequency and size of writes. When an
application tries to push more data then a device can handle, the driver has to buffer all bytes that the device has
not yet been able to write. With this approach it is possible to handle short bursts of intensive writes --- but no buffer is infinite.
Therefore, the driver has to notify the writer (a user-side actor) either that no further writes are possible, or by
explicitly notifying it when the next chunk is possible to be written or buffered.
Both of these models are available in the TCP and UDP implementations of Akka I/O. Ack based flow control can be enabled
by providing an ack object in the write message (``Write`` in the case of TCP and ``Send`` for UDP) that will be used by
the worker to notify the writer about the success.
If a write (or any other command) fails, the driver notifies the commander with a special message (``CommandFailed`` in
the case of UDP and TCP). This message also serves as a means to notify the writer of a failed write. Please note, that
in a Nack based flow-control setting the writer has to buffer some of the writes as the failure notification for a
write ``W1`` might arrive after additional write commands ``W2`` ``W3`` has been sent.
.. warning::
An acknowledged write does not mean acknowledged delivery or storage. The Ack/Nack
protocol described here is a means of flow control not error handling: receiving an Ack for a write signals that the
I/O driver is ready to accept a new one.
ByteString
^^^^^^^^^^
A primary goal of Akka's IO support is to only communicate between actors with immutable objects. When dealing with network IO on the jvm ``Array[Byte]`` and ``ByteBuffer`` are commonly used to represent collections of ``Byte``\s, but they are mutable. Scala's collection library also lacks a suitably efficient immutable collection for ``Byte``\s. Being able to safely and efficiently move ``Byte``\s around is very important for this IO support, so ``ByteString`` was developed.
A primary goal of Akka's IO support is to only communicate between actors with immutable objects. When dealing with network I/O on the jvm ``Array[Byte]`` and ``ByteBuffer`` are commonly used to represent collections of ``Byte``\s, but they are mutable. Scala's collection library also lacks a suitably efficient immutable collection for ``Byte``\s. Being able to safely and efficiently move ``Byte``\s around is very important for this I/O support, so ``ByteString`` was developed.
``ByteString`` is a `Rope-like <http://en.wikipedia.org/wiki/Rope_(computer_science)>`_ data structure that is immutable and efficient. When 2 ``ByteString``\s are concatenated together they are both stored within the resulting ``ByteString`` instead of copying both to a new ``Array``. Operations such as ``drop`` and ``take`` return ``ByteString``\s that still reference the original ``Array``, but just change the offset and length that is visible. Great care has also been taken to make sure that the internal ``Array`` cannot be modified. Whenever a potentially unsafe ``Array`` is used to create a new ``ByteString`` a defensive copy is created. If you require a ``ByteString`` that only blocks a much memory as necessary for it's content, use the ``compact`` method to get a ``CompactByteString`` instance. If the ``ByteString`` represented only a slice of the original array, this will result in copying all bytes in that slice.
@ -57,7 +111,7 @@ After extracting data from a ``ByteIterator``, the remaining content can also be
.. includecode:: code/docs/io/BinaryCoding.scala
:include: rest-to-seq
with no copying from bytes to rest involved. In general, conversions from ByteString to ByteIterator and vice versa are O(1) for non-chunked ByteStrings and (at worst) O(nChunks) for chunked ByteStrings.
Encoding of data also is very natural, using ``ByteStringBuilder``
@ -65,180 +119,263 @@ Encoding of data also is very natural, using ``ByteStringBuilder``
.. includecode:: code/docs/io/BinaryCoding.scala
:include: encoding
The encoded data then can be sent over socket (see ``IOManager``):
.. includecode:: code/docs/io/BinaryCoding.scala
:include: sending
Using TCP
---------
IO.Handle
^^^^^^^^^
``IO.Handle`` is an immutable reference to a Java NIO ``Channel``. Passing mutable ``Channel``\s between ``Actor``\s could lead to unsafe behavior, so instead subclasses of the ``IO.Handle`` trait are used. Currently there are 2 concrete subclasses: ``IO.SocketHandle`` (representing a ``SocketChannel``) and ``IO.ServerHandle`` (representing a ``ServerSocketChannel``).
IOManager
^^^^^^^^^
The ``IOManager`` takes care of the low level IO details. Each ``ActorSystem`` has it's own ``IOManager``, which can be accessed calling ``IOManager(system: ActorSystem)``. ``Actor``\s communicate with the ``IOManager`` with specific messages. The messages sent from an ``Actor`` to the ``IOManager`` are handled automatically when using certain methods and the messages sent from an ``IOManager`` are handled within an ``Actor``\'s ``receive`` method.
Connecting to a remote host:
As with all of the Akka I/O APIs, everything starts with acquiring a reference to the appropriate manager:
.. code-block:: scala
val address = new InetSocketAddress("remotehost", 80)
val socket = IOManager(actorSystem).connect(address)
import akka.io.IO
import akka.io.Tcp
val tcpManager = IO(Tcp)
This is an actor that handles the underlying low level I/O resources (Selectors, channels) and instantiates workers for
specific tasks, like listening to incoming connections.
Connecting
^^^^^^^^^^
The first step of connecting to a remote address is sending a ``Connect`` message to the TCP manager:
.. code-block:: scala
val socket = IOManager(actorSystem).connect("remotehost", 80)
import akka.io.Tcp._
IO(Tcp) ! Connect(remoteSocketAddress)
// It is also possible to set various socket options or specify a local address:
IO(Tcp) ! Connect(remoteSocketAddress, Some(localSocketAddress), List(SO.KeepAlive(true)))
Creating a server:
After issuing the Connect command the TCP manager spawns a worker actor that will handle commands related to the
connection. This worker actor will reveal itself by replying with a ``Connected`` message to the actor who sent the
``Connect`` command.
.. code-block:: scala
val address = new InetSocketAddress("localhost", 80)
val serverSocket = IOManager(actorSystem).listen(address)
case Connected(remoteAddress, localAddress) =>
connectionActor = sender
At this point, there is still no listener associated with the connection. To finish the connection setup a ``Register``
has to be sent to the connection actor with the listener ``ActorRef`` as a parameter.
.. code-block:: scala
val serverSocket = IOManager(actorSystem).listen("localhost", 80)
connectionActor ! Register(listener)
Receiving messages from the ``IOManager``:
After registration, the listener actor provided in the ``listener`` parameter will be watched by the connection actor.
If the listener stops, the connection is closed, and all resources allocated for the connection released. During the
lifetime the listener may receive various event notifications:
.. code-block:: scala
def receive = {
case Received(dataByteString) => // handle incoming chunk of data
case CommandFailed(cmd) => // handle failure of command: cmd
case _: ConnectionClosed => // handle closed connections
case IO.Listening(server, address) =>
println("The server is listening on socket " + address)
The last line handles all connection close events in the same way. It is possible to listen for more fine-grained
connection events, see the appropriate section below.
case IO.Connected(socket, address) =>
println("Successfully connected to " + address)
case IO.NewClient(server) =>
println("New incoming connection on server")
val socket = server.accept()
println("Writing to new client socket")
socket.write(bytes)
println("Closing socket")
socket.close()
Accepting connections
^^^^^^^^^^^^^^^^^^^^^
case IO.Read(socket, bytes) =>
println("Received incoming data from socket")
To create a TCP server and listen for inbound connection, a ``Bind`` command has to be sent to the TCP manager:
case IO.Closed(socket: IO.SocketHandle, cause) =>
println("Socket has closed, cause: " + cause)
.. code-block:: scala
case IO.Closed(server: IO.ServerHandle, cause) =>
println("Server socket has closed, cause: " + cause)
import akka.io.IO
import akka.io.Tcp
IO(Tcp) ! Bind(handler, localAddress)
}
The actor sending the ``Bind`` message will receive a ``Bound`` message signalling that the server is ready to accept
incoming connections. Accepting connections is very similar to the last two steps of opening outbound connections: when
an incoming connection is established, the actor provided in ``handler`` will receive a ``Connected`` message whose
sender is the connection actor:
IO.Iteratee
^^^^^^^^^^^
.. code-block:: scala
Included with Akka's IO support is a basic implementation of ``Iteratee``\s. ``Iteratee``\s are an effective way of handling a stream of data without needing to wait for all the data to arrive. This is especially useful when dealing with non blocking IO since we will usually receive data in chunks which may not include enough information to process, or it may contain much more data than we currently need.
case Connected(remoteAddress, localAddress) =>
connectionActor = sender
This ``Iteratee`` implementation is much more basic than what is usually found. There is only support for ``ByteString`` input, and enumerators aren't used. The reason for this limited implementation is to reduce the amount of explicit type signatures needed and to keep things simple. It is important to note that Akka's ``Iteratee``\s are completely optional, incoming data can be handled in any way, including other ``Iteratee`` libraries.
At this point, there is still no listener associated with the connection. To finish the connection setup a ``Register``
has to be sent to the connection actor with the listener ``ActorRef`` as a parameter.
``Iteratee``\s work by processing the data that it is given and returning either the result (with any unused input) or a continuation if more input is needed. They are monadic, so methods like ``flatMap`` can be used to pass the result of an ``Iteratee`` to another.
.. code-block:: scala
The basic ``Iteratee``\s included in the IO support can all be found in the ScalaDoc under ``akka.actor.IO``, and some of them are covered in the example below.
connectionActor ! Register(listener)
Examples
--------
After registration, the listener actor provided in the ``listener`` parameter will be watched by the connection actor.
If the listener stops, the connection is closed, and all resources allocated for the connection released. During the
lifetime the listener will receive various event notifications in the same way as we has seen in the outbound
connection case.
Http Server
^^^^^^^^^^^
Closing connections
^^^^^^^^^^^^^^^^^^^
This example will create a simple high performance HTTP server. We begin with our imports:
A connection can be closed by sending one of the commands ``Close``, ``ConfirmedClose`` or ``Abort`` to the connection
actor.
.. includecode:: code/docs/io/HTTPServer.scala
:include: imports
``Close`` will close the connection by sending a ``FIN`` message, but without waiting for confirmation from
the remote endpoint. Pending writes will be flushed. If the close is successful, the listener will be notified with
``Closed``
Some commonly used constants:
``ConfirmedClose`` will close the sending direction of the connection by sending a ``FIN`` message, but receives
will continue until the remote endpoint closes the connection, too. Pending writes will be flushed. If the close is
successful, the listener will be notified with ``ConfirmedClosed``
.. includecode:: code/docs/io/HTTPServer.scala
:include: constants
``Abort`` will immediately terminate the connection by sending a ``RST`` message to the remote endpoint. Pending
writes will be not flushed. If the close is successful, the listener will be notified with ``Aborted``
And case classes to hold the resulting request:
``PeerClosed`` will be sent to the listener if the connection has been closed by the remote endpoint.
.. includecode:: code/docs/io/HTTPServer.scala
:include: request-class
``ErrorClosed`` will be sent to the listener whenever an error happened that forced the connection to be closed.
Now for our first ``Iteratee``. There are 3 main sections of a HTTP request: the request line, the headers, and an optional body. The main request ``Iteratee`` handles each section separately:
All close notifications are subclasses of ``ConnectionClosed`` so listeners who do not need fine-grained close events
may handle all close events in the same way.
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request
Throttling Reads and Writes
^^^^^^^^^^^^^^^^^^^^^^^^^^^
In the above code ``readRequest`` takes the results of 3 different ``Iteratees`` (``readRequestLine``, ``readHeaders``, ``readBody``) and combines them into a single ``Request`` object. ``readRequestLine`` actually returns a tuple, so we extract it's individual components. ``readBody`` depends on values contained within the header section, so we must pass those to the method.
*This section is not yet ready. More coming soon*
The request line has 3 parts to it: the HTTP method, the requested URI, and the HTTP version. The parts are separated by a single space, and the entire request line ends with a ``CRLF``.
Using UDP
---------
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request-line
UDP support comes in two flavors: connectionless, and connection based:
Reading the request method is simple as it is a single string ending in a space. The simple ``Iteratee`` that performs this is ``IO.takeUntil(delimiter: ByteString): Iteratee[ByteString]``. It keeps consuming input until the specified delimiter is found. Reading the HTTP version is also a simple string that ends with a ``CRLF``.
.. code-block:: scala
The ``ascii`` method is a helper that takes a ``ByteString`` and parses it as a ``US-ASCII`` ``String``.
import akka.io.IO
import akka.io.UdpFF
val connectionLessUdp = IO(UdpFF)
// ... or ...
import akka.io.UdpConn
val connectionBasedUdp = IO(UdpConn)
Reading the request URI is a bit more complicated because we want to parse the individual components of the URI instead of just returning a simple string:
UDP servers can be only implemented by the connectionless API, but clients can use both.
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-request-uri
Connectionless UDP
^^^^^^^^^^^^^^^^^^
For this example we are only interested in handling absolute paths. To detect if we the URI is an absolute path we use ``IO.peek(length: Int): Iteratee[ByteString]``, which returns a ``ByteString`` of the request length but doesn't actually consume the input. We peek at the next bit of input and see if it matches our ``PATH`` constant (defined above as ``ByteString("/")``). If it doesn't match we throw an error, but for a more robust solution we would want to handle other valid URIs.
Simple Send
............
Next we handle the path itself:
To simply send a UDP datagram without listening to an answer one needs to send the ``SimpleSender`` command to the
manager:
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-path
.. code-block:: scala
The ``step`` method is a recursive method that takes a ``List`` of the accumulated path segments. It first checks if the remaining input starts with the ``PATH`` constant, and if it does, it drops that input, and returns the ``readUriPart`` ``Iteratee`` which has it's result added to the path segment accumulator and the ``step`` method is run again.
IO(UdpFF) ! SimpleSender
// or with socket options:
import akka.io.Udp._
IO(UdpFF) ! SimpleSender(List(SO.Broadcast(true)))
If after reading in a path segment the next input does not start with a path, we reverse the accumulated segments and return it (dropping the last segment if it is blank).
The manager will create a worker for sending, and the worker will reply with a ``SimpleSendReady`` message:
Following the path we read in the query (if it exists):
.. code-block:: scala
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-query
case SimpleSendReady =>
simpleSender = sender
It is much simpler than reading the path since we aren't doing any parsing of the query since there is no standard format of the query string.
After saving the sender of the ``SimpleSendReady`` message it is possible to send out UDP datagrams with a simple
message send:
Both the path and query used the ``readUriPart`` ``Iteratee``, which is next:
.. code-block:: scala
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-uri-part
simpleSender ! Send(data, serverAddress)
Here we have several ``Set``\s that contain valid characters pulled from the URI spec. The ``readUriPart`` method takes a ``Set`` of valid characters (already mapped to ``Byte``\s) and will continue to match characters until it reaches on that is not part of the ``Set``. If it is a percent encoded character then that is handled as a valid character and processing continues, or else we are done collecting this part of the URI.
Headers are next:
Bind (and Send)
...............
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-headers
To listen for UDP datagrams arriving on a given port, the ``Bind`` command has to be sent to the connectionless UDP
manager
And if applicable, we read in the message body:
.. code-block:: scala
.. includecode:: code/docs/io/HTTPServer.scala
:include: read-body
IO(UdpFF) ! Bind(handler, localAddress)
Finally we get to the actual ``Actor``:
After the bind succeeds, the sender of the ``Bind`` command will be notified with a ``Bound`` message. The sender of
this message is the worker for the UDP channel bound to the local address.
.. includecode:: code/docs/io/HTTPServer.scala
:include: actor
.. code-block:: scala
And it's companion object:
case Bound =>
udpWorker = sender // Save the worker ref for later use
.. includecode:: code/docs/io/HTTPServer.scala
:include: actor-companion
The actor passed in the ``handler`` parameter will receive inbound UDP datagrams sent to the bound address:
And the OKResponse:
.. code-block:: scala
.. includecode:: code/docs/io/HTTPServer.scala
:include: ok-response
case Received(dataByteString, remoteAddress) => // Do something with the data
A ``main`` method to start everything up:
The ``Received`` message contains the payload of the datagram and the address of the sender.
.. includecode:: code/docs/io/HTTPServer.scala
:include: main
It is also possible to send UDP datagrams using the ``ActorRef`` of the worker saved in ``udpWorker``:
.. code-block:: scala
udpWorker ! Send(data, serverAddress)
.. note::
The difference between using a bound UDP worker to send instead of a simple-send worker is that in the former case
the sender field of the UDP datagram will be the bound local address, while in the latter it will be an undetermined
ephemeral port.
Connection based UDP
^^^^^^^^^^^^^^^^^^^^
The service provided by the connection based UDP API is similar to the bind-and-send service we have seen earlier, but
the main difference is that a connection is only able to send to the remoteAddress it was connected to, and will
receive datagrams only from that address.
Connecting is similar to what we have seen in the previous section:
.. code-block:: scala
IO(UdpConn) ! Connect(handler, remoteAddress)
// or, with more options:
IO(UdpConn) ! Connect(handler, Some(localAddress), remoteAddress, List(SO.Broadcast(true)))
After the connect succeeds, the sender of the ``Connect`` command will be notified with a ``Connected`` message. The sender of
this message is the worker for the UDP connection.
.. code-block:: scala
case Connected =>
udpConnectionActor = sender // Save the worker ref for later use
The actor passed in the ``handler`` parameter will receive inbound UDP datagrams sent to the bound address:
.. code-block:: scala
case Received(dataByteString) => // Do something with the data
The ``Received`` message contains the payload of the datagram but unlike in the connectionless case, no sender address
will be provided, as an UDP connection only receives messages from the endpoint it has been connected to.
It is also possible to send UDP datagrams using the ``ActorRef`` of the worker saved in ``udpWorker``:
.. code-block:: scala
udpConnectionActor ! Send(data)
Again, the send does not contain a remote address, as it is always the endpoint we have been connected to.
.. note::
There is a small performance benefit in using connection based UDP API over the connectionless one.
If there is a SecurityManager enabled on the system, every connectionless message send has to go through a security
check, while in the case of connection-based UDP the security check is cached after connect, thus writes does
not suffer an additional performance penalty.
Throttling Reads and Writes
^^^^^^^^^^^^^^^^^^^^^^^^^^^
*This section is not yet ready. More coming soon*
Architecture in-depth
---------------------
For further details on the design and internal architecture see :ref:`io-layer`.
.. _spray.io: http://spray.io