Support guaranteed delivery from ForwardOutput to ForwardInput

Protocols between ForwardOutput and ForwardInput are extended, adding `option` object
to the end of the array, which is used to identify which chunks are written into the buffer.
Additional behaviors of ForwardOutput and ForwardInput are as follows:

ForwardOutput sends data including `option` object which has object id of the chunk
with a value of `seq` key. After sending data, ForwardOutput waits for a response
from InputForward in a certain time.
* If there is a response:
  * and if `ack` value of the response is same as `seq` value of sent data, it succeeded.
  * but if `ack` value and `seq` value are different, ForwardOutput sends the chunk again.
* If there is no response, ForwardOutput regards the target node as unavailable
  but does not send the chunk again because of compatibility.

InputForward checks if received data includes `option` object. If it does, InputForward
responds to OutputForward after the chunk is written into the buffer (Engine.emit).
The response has the `seq` value of the received data as a value of `ack` key, and is
serialized in the same way as received data.
Fix an undefined variable name

n    ->
node
Avoid deserializing an empty string which leads to EOFError

When connection is closed by remote host, Socket#recv returns an empty
string. So deserialize received data only when it is not empty.
Add test cases with DummyEngineDriver

In test cases, DummyEngineDriver creates a TCP server in another thread and runs ForwardInput on it.
To prevent ForwardInput from accessing Fluent::Engine by calling Engine.emit_streams,
DummyEngineDriver creates a dummy engine and sets Fluent::ForwardInput::Engine to it.
Add config parameters for requiring in_forward to respond with ack

There are three parameters for ack (two are newly added, one is renamed):

* extend_internal_protocol
  Whether to add an optional hash to the end of the sent data from
  out_forward to in_forward. This is used for requiring ack responses or
  some future functions.

* require_ack_response
  Whether to require in_forward to respond with ack. If true, the
  optional hash contains a chunk id.

* ack_response_timeout
  How many seconds out_forward waits for an ack response in.

To enable ack responses, set both extend_internal_protocol and
require_ack_response to true, and set ack_response_timeout to greater
than zero.
Rename test methods which do not correspond with the situations
Set when OutputTestDrier#run ends

`d.expected_emits_length` is set only to wait for Driver to finish
running on timeout. The length of emits is not suitable for a post
condition because ForwardOutput does not actually call
Engine.emit_streams and @emit_streams is always empty.
fix to set dummy engine into @router, to follow master
Fix to use chunk.unique_id to check ack
 * unique_id is safe for restarts
 * base64 encoding is escape msgpack miss-decoding
  * by Msgpack encode-decode: \u0005 -> \x05
  * this behavior breaks chunk id check
Fix to assume extend_internal_protocol from other options which uses extended protocols
Fix default ack timeout w/ linux kernel default configuration
Fix to raise error for ACK
 * retry send_data for ack timeout
   * by this behavior, fluentd can try other node for failed chunk
   * by this behavior, user can find badly-configured situation
pre-encoded forward header bytes, for performance
log typo
Merge pull request #428 from tagomoris/support-guaranteed-delivery

Implementation of guaranteed delivery from ForwardOutput to ForwardInput