### Write Records to Multiple Streams

This tutorial shows how to write records using write proxy multi stream writer. The `DistributedLogMultiStreamWriter`
is a wrapper over `DistributedLogClient` on writing records to a set of streams in a `round-robin` way and ensure low write latency even on single stream ownership failover.

#### Open a write proxy client

-   Create write proxy client builder.
    ```
        DistributedLogClientBuilder builder = DistributedLogClientBuilder.newBuilder()
                .clientId(ClientId.apply("console-proxy-writer"))
                .name("console-proxy-writer");
    ```

-   Enable thrift mux.
    ```
        builder = builder.thriftmux(true);
    ```

-   Point the client to write proxy using finagle name.
    ```
        String finagleNameStr = "inet!127.0.0.1:8000";
        builder = builder.finagleNameStr(finagleNameStr);
    ```

-   Build the write proxy client.
    ```
        DistributedLogClient client = builder.build();
    ```

#### Create a `MultiStreamWriter`

-   Create multi stream writer builder.
    ```
        DistributedLogMultiStreamWriterBuilder builder = DistributedLogMultiStreamWriter.newBuilder();
    ```

-   Build the writer to write a set of streams.
    ```
        List<String> streams = ...;
        builder = builder.streams(streams);
    ```

-   Point the multi stream writer to use write proxy client.
    ```
        builder = builder.client(client);
    ```

-   Configure the flush policy for the multi stream writer.
    ```
        // transmit immediately after a record is written.
        builder = builder.bufferSize(0);
        builder = builder.flushIntervalMs(0);
    ```

-   Configure the request timeouts and retry policy for the multi stream writer.
    ```
        // Configure the speculative timeouts - if writing to a stream cannot
        // complete within the speculative timeout, it would try writing to
        // another streams.
        builder = builder.firstSpeculativeTimeoutMs(10000)
        builder = builder.maxSpeculativeTimeoutMs(20000)
        // Configure the request timeout.
        builder = builder.requestTimeoutMs(50000);
    ```

-   Build the multi writer.
    ```
        DistributedLogMultiStreamWriter writer = builder.build();
    ```

#### Write Records

-   Write records to multi streams. 
    ```
        byte[] data = ...;
        Future<DLSN> writeFuture = writer.write(ByteBuffer.wrap(data));
    ```

-   Register a future listener on write completion.
    ```
        writeFuture.addEventListener(new FutureEventListener<DLSN>() {
            @Override
            public void onFailure(Throwable cause) {
                // executed when write failed.
            }

            @Override
            public void onSuccess(DLSN value) {
                // executed when write completed.
            }
        });
    ```

#### Usage

Run the example in the following steps:

1.  Start the local bookkeeper cluster. After the bookkeeper cluster is started, you could access
    it using distributedlog uri *distributedlog://127.0.0.1:7000/messaging/distributedlog*.

    ```
        // dlog local ${zk-port}
        ./distributedlog-core/bin/dlog local 7000
    ```

2.  Start the write proxy, listening on port 8000.
    ```
        // DistributedLogServerApp -p ${service-port} --shard-id ${shard-id} -sp ${stats-port} -u {distributedlog-uri} -mx -c ${conf-file}
        ./distributedlog-proxy-server/bin/dlog org.apache.distributedlog.service.DistributedLogServerApp -p 8000 --shard-id 1 -sp 8001 -u distributedlog://127.0.0.1:7000/messaging/distributedlog -mx -c ${distributedlog-repo}/distributedlog-proxy-server/conf/distributedlog_proxy.conf
    ```

3.  Create multi streams under the distributedlog uri.

    ```
        // Create Stream `basic-stream-{3-7}`
        // dlog tool create -u ${distributedlog-uri} -r ${stream-prefix} -e ${stream-regex}
        ./distributedlog-core/bin/dlog tool create -u distributedlog://127.0.0.1:7000/messaging/distributedlog -r basic-stream- -e 3-7
    ```

4.  Tailing multi streams using `MultiReader` to wait for new records.
    ```
        // Tailing Stream `basic-stream-{3-7}`
        // runner run org.apache.distributedlog.basic.MultiReader ${distributedlog-uri} ${stream}[,${stream}]
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.MultiReader distributedlog://127.0.0.1:7000/messaging/distributedlog basic-stream-3,basic-stream-4,basic-stream-5,basic-stream-6,basic-stream-7
    ```

6.  Run the example to write records to the multi streams in a console.
    ```
        // Write Records into Stream `basic-stream-{3-7}`
        // runner run org.apache.distributedlog.basic.ConsoleProxyMultiWriter ${distributedlog-uri} ${stream}[,${stream}]
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.ConsoleProxyMultiWriter 'inet!127.0.0.1:8000' basic-stream-3,basic-stream-4,basic-stream-5,basic-stream-6,basic-stream-7
    ```

7.  Example output from `ConsoleProxyMultiWriter` and `MultiReader`.
    ```
        // Output of `ConsoleProxyWriter`
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[inet] = com.twitter.finagle.InetResolver(com.twitter.finagle.InetResolver@fbb628c)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[fixedinet] = com.twitter.finagle.FixedInetResolver(com.twitter.finagle.FixedInetResolver@5a25adb1)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[neg] = com.twitter.finagle.NegResolver$(com.twitter.finagle.NegResolver$@5fae6db3)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[nil] = com.twitter.finagle.NilResolver$(com.twitter.finagle.NilResolver$@34a433d8)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[fail] = com.twitter.finagle.FailResolver$(com.twitter.finagle.FailResolver$@847c4e8)
        May 08, 2016 11:09:22 AM com.twitter.finagle.Init$$anonfun$1 apply$mcV$sp
        [dlog] > message-1
        [dlog] > message-2
        [dlog] > message-3
        [dlog] > message-4
        [dlog] > message-5
        [dlog] >


        // Output of `MultiReader`
        Opening log stream basic-stream-3
        Opening log stream basic-stream-4
        Opening log stream basic-stream-5
        Opening log stream basic-stream-6
        Opening log stream basic-stream-7
        Log stream basic-stream-4 is empty.
        Wait for records from basic-stream-4 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-4
        Log stream basic-stream-5 is empty.
        Wait for records from basic-stream-5 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-5
        Log stream basic-stream-6 is empty.
        Wait for records from basic-stream-6 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-6
        Log stream basic-stream-3 is empty.
        Wait for records from basic-stream-3 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-3
        Log stream basic-stream-7 is empty.
        Wait for records from basic-stream-7 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-7
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-4
        """
        message-1
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-6
        """
        message-2
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-3
        """
        message-3
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-7
        """
        message-4
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-5
        """
        message-5
        """
    ```