== Streams

There are several objects in Vert.x that allow items to be read from and written.

In previous versions the streams.adoc package was manipulating `link:../../scaladocs/io/vertx/scala/core/buffer/Buffer.html[Buffer]`
objects exclusively. From now, streams are not coupled to buffers anymore and they work with any kind of objects.

In Vert.x, write calls return immediately, and writes are queued internally.

It's not hard to see that if you write to an object faster than it can actually write the data to
its underlying resource, then the write queue can grow unbounded - eventually resulting in
memory exhaustion.

To solve this problem a simple flow control (_back-pressure_) capability is provided by some objects in the Vert.x API.

Any flow control aware object that can be _written-to_ implements `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html[WriteStream]`,
while any flow control object that can be _read-from_ is said to implement `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html[ReadStream]`.

Let's take an example where we want to read from a `ReadStream` then write the data to a `WriteStream`.

A very simple example would be reading from a `link:../../scaladocs/io/vertx/scala/core/net/NetSocket.html[NetSocket]` then writing back to the
same `NetSocket` - since `NetSocket` implements both `ReadStream` and `WriteStream`. Note that this works
between any `ReadStream` and `WriteStream` compliant object, including HTTP requests, HTTP responses,
async files I/O, WebSockets, etc.

A naive way to do this would be to directly take the data that has been read and immediately write it
to the `NetSocket`:

[source,scala]
----
var server = vertx.createNetServer(NetServerOptions()
  .setPort(1234)
  .setHost("localhost")
)
server.connectHandler((sock: io.vertx.scala.core.net.NetSocket) => {
  sock.handler((buffer: io.vertx.scala.core.buffer.Buffer) => {
    // Write the data straight back
    sock.write(buffer)
  })
}).listen()

----

There is a problem with the example above: if data is read from the socket faster than it can be
written back to the socket, it will build up in the write queue of the `NetSocket`, eventually
running out of RAM. This might happen, for example if the client at the other end of the socket
wasn't reading fast enough, effectively putting back-pressure on the connection.

Since `NetSocket` implements `WriteStream`, we can check if the `WriteStream` is full before
writing to it:

[source,scala]
----
var server = vertx.createNetServer(NetServerOptions()
  .setPort(1234)
  .setHost("localhost")
)
server.connectHandler((sock: io.vertx.scala.core.net.NetSocket) => {
  sock.handler((buffer: io.vertx.scala.core.buffer.Buffer) => {
    if (!sock.writeQueueFull()) {
      sock.write(buffer)
    }
  })

}).listen()

----

This example won't run out of RAM but we'll end up losing data if the write queue gets full. What we
really want to do is pause the `NetSocket` when the write queue is full:

[source,scala]
----
var server = vertx.createNetServer(NetServerOptions()
  .setPort(1234)
  .setHost("localhost")
)
server.connectHandler((sock: io.vertx.scala.core.net.NetSocket) => {
  sock.handler((buffer: io.vertx.scala.core.buffer.Buffer) => {
    sock.write(buffer)
    if (sock.writeQueueFull()) {
      sock.pause()
    }
  })
}).listen()

----

We're almost there, but not quite. The `NetSocket` now gets paused when the file is full, but we also need to unpause
it when the write queue has processed its backlog:

[source,scala]
----
var server = vertx.createNetServer(NetServerOptions()
  .setPort(1234)
  .setHost("localhost")
)
server.connectHandler((sock: io.vertx.scala.core.net.NetSocket) => {
  sock.handler((buffer: io.vertx.scala.core.buffer.Buffer) => {
    sock.write(buffer)
    if (sock.writeQueueFull()) {
      sock.pause()
      sock.drainHandler((done: java.lang.Void) => {
        sock.resume()
      })
    }
  })
}).listen()

----

And there we have it. The `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#drainHandler(io.vertx.core.Handler))[drainHandler]` event handler will
get called when the write queue is ready to accept more data, this resumes the `NetSocket` that
allows more data to be read.

Wanting to do this is quite common while writing Vert.x applications, so we provide a helper class
called `link:../../scaladocs/io/vertx/scala/core/streams/Pump.html[Pump]` that does all of this hard work for you.
You just feed it the `ReadStream` plus the `WriteStream` then start it:

[source,scala]
----
var server = vertx.createNetServer(NetServerOptions()
  .setPort(1234)
  .setHost("localhost")
)
server.connectHandler((sock: io.vertx.scala.core.net.NetSocket) => {
  Pump.pump(sock, sock).start()
}).listen()

----

This does exactly the same thing as the more verbose example.

Let's now look at the methods on `ReadStream` and `WriteStream` in more detail:

=== ReadStream

`ReadStream` is implemented by `link:../../scaladocs/io/vertx/scala/core/http/HttpClientResponse.html[HttpClientResponse]`, `link:../../scaladocs/io/vertx/scala/core/datagram/DatagramSocket.html[DatagramSocket]`,
`link:../../scaladocs/io/vertx/scala/core/http/HttpClientRequest.html[HttpClientRequest]`, `link:../../scaladocs/io/vertx/scala/core/http/HttpServerFileUpload.html[HttpServerFileUpload]`,
`link:../../scaladocs/io/vertx/scala/core/http/HttpServerRequest.html[HttpServerRequest]`, `link:../../scaladocs/io/vertx/scala/core/eventbus/MessageConsumer.html[MessageConsumer]`,
`link:../../scaladocs/io/vertx/scala/core/net/NetSocket.html[NetSocket]`, `link:../../scaladocs/io/vertx/scala/core/http/WebSocket.html[WebSocket]`, `link:../../scaladocs/io/vertx/scala/core/TimeoutStream.html[TimeoutStream]`,
`link:../../scaladocs/io/vertx/scala/core/file/AsyncFile.html[AsyncFile]`.

Functions:

- `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html#handler(io.vertx.core.Handler))[handler]`:
set a handler which will receive items from the ReadStream.
- `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html#pause()[pause]`:
pause the handler. When paused no items will be received in the handler.
- `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html#resume()[resume]`:
resume the handler. The handler will be called if any item arrives.
- `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html#exceptionHandler(io.vertx.core.Handler)[exceptionHandler]`:
Will be called if an exception occurs on the ReadStream.
- `link:../../scaladocs/io/vertx/scala/core/streams/ReadStream.html#endHandler(io.vertx.core.Handler))[endHandler]`:
Will be called when end of stream is reached. This might be when EOF is reached if the ReadStream represents a file,
or when end of request is reached if it's an HTTP request, or when the connection is closed if it's a TCP socket.

=== WriteStream

`WriteStream` is implemented by `link:../../scaladocs/io/vertx/scala/core/http/HttpClientRequest.html[HttpClientRequest]`, `link:../../scaladocs/io/vertx/scala/core/http/HttpServerResponse.html[HttpServerResponse]`
`link:../../scaladocs/io/vertx/scala/core/http/WebSocket.html[WebSocket]`, `link:../../scaladocs/io/vertx/scala/core/net/NetSocket.html[NetSocket]`, `link:../../scaladocs/io/vertx/scala/core/file/AsyncFile.html[AsyncFile]`,
and `link:../../scaladocs/io/vertx/scala/core/eventbus/MessageProducer.html[MessageProducer]`

Functions:

- `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#write(java.lang.Object)[write]`:
write an object to the WriteStream. This method will never block. Writes are queued internally and asynchronously
written to the underlying resource.
- `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#setWriteQueueMaxSize(int)[setWriteQueueMaxSize]`:
set the number of object at which the write queue is considered _full_, and the method `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#writeQueueFull()[writeQueueFull]`
returns `true`. Note that, when the write queue is considered full, if write is called the data will still be accepted
and queued. The actual number depends on the stream implementation, for `link:../../scaladocs/io/vertx/scala/core/buffer/Buffer.html[Buffer]` the size
represents the actual number of bytes written and not the number of buffers.
- `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#writeQueueFull()[writeQueueFull]`:
returns `true` if the write queue is considered full.
- `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#exceptionHandler(io.vertx.core.Handler)[exceptionHandler]`:
Will be called if an exception occurs on the `WriteStream`.
- `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#drainHandler(io.vertx.core.Handler))[drainHandler]`:
The handler will be called if the `WriteStream` is considered no longer full.

=== Pump

Instances of Pump have the following methods:

- `link:../../scaladocs/io/vertx/scala/core/streams/Pump.html#start()[start]`:
Start the pump.
- `link:../../scaladocs/io/vertx/scala/core/streams/Pump.html#stop()[stop]`:
Stops the pump. When the pump starts it is in stopped mode.
- `link:../../scaladocs/io/vertx/scala/core/streams/Pump.html#setWriteQueueMaxSize(int)[setWriteQueueMaxSize]`:
This has the same meaning as `link:../../scaladocs/io/vertx/scala/core/streams/WriteStream.html#setWriteQueueMaxSize(int)[setWriteQueueMaxSize]` on the `WriteStream`.

A pump can be started and stopped multiple times.

When a pump is first created it is _not_ started. You need to call the `start()` method to start it.