distributedlog-tutorials/distributedlog-basic/basic-5.md

Tail reading records from a stream

This tutorial shows how to tail read records from a stream.

Open a distributedlog manager

Create distributedlog URI.

    String dlUriStr = ...;
    URI uri = URI.create(dlUriStr);

Create distributedlog configuration.

    DistributedLogConfiguration conf = new DistributedLogConfiguration();

Build the distributedlog namespace.

    DistributedLogNamespace namespace = DistributedLogNamespaceBuilder.newBuilder()
            .conf(conf)
            .uri(uri)
            .build();

Open the distributedlog manager.

    DistributedLogManager dlm = namespace.openLog("basic-stream-9");

Get Last Record

Get the last record from the record. From the last record, we can use DLSN of last record to start tailing the stream.
```
    LogRecordWithDLSN record = dlm.getLastLogRecord();
    DLSN lastDLSN = record.getDlsn();
```

Read Records

Open the stream to start read the records.

    AsyncLogReader reader = FutureUtils.result(dlm.openAsyncLogReader(lastDLSN));

Read the next available record from the stream. The future is satisified when the record is available.
```
    Future<LogRecordWithDLSN> readFuture = reader.readNext();
```

    final FutureEventListener<LogRecordWithDLSN> readListener = new FutureEventListener<LogRecordWithDLSN>() {
        @Override
        public void onFailure(Throwable cause) {
            // executed when read failed.
        }

        @Override
        public void onSuccess(LogRecordWithDLSN record) {
            // process the record
            ...
            // issue read next
            reader.readNext().addEventListener(this);
        }
    };
    reader.readNext().addEventListener(readListener);

Close the reader

Close the reader after usage.

    FutureUtils.result(reader.asyncClose());

Usage

Run the example in the following steps:

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
```

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

Create the stream under the distributedlog uri.

    // Create Stream `basic-stream-9`
    // 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 9

Tailing the stream using TailReader to wait for new records.

    // Tailing Stream `basic-stream-9`
    // runner run org.apache.distributedlog.basic.TailReader ${distributedlog-uri} ${stream}
    ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.TailReader distributedlog://127.0.0.1:7000/messaging/distributedlog basic-stream-9

Run the example to write records to the stream in a console.

    // Write Records into Stream `basic-stream-9`
    // runner run org.apache.distributedlog.basic.ConsoleProxyWriter ${distributedlog-uri} ${stream}
    ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.ConsoleProxyWriter 'inet!127.0.0.1:8000' basic-stream-9

Example output from ConsoleProxyWriter and TailReader.

    // Output of `ConsoleProxyWriter`
    May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
    INFO: Resolver[inet] = com.twitter.finagle.InetResolver(com.twitter.finagle.InetResolver@756d7bba)
    May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
    INFO: Resolver[fixedinet] = com.twitter.finagle.FixedInetResolver(com.twitter.finagle.FixedInetResolver@1d2e91f5)
    May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
    INFO: Resolver[neg] = com.twitter.finagle.NegResolver$(com.twitter.finagle.NegResolver$@5c707aca)
    May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
    INFO: Resolver[nil] = com.twitter.finagle.NilResolver$(com.twitter.finagle.NilResolver$@5c8d932f)
    May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
    INFO: Resolver[fail] = com.twitter.finagle.FailResolver$(com.twitter.finagle.FailResolver$@52ba2221)
    May 08, 2016 10:27:41 AM com.twitter.finagle.Init$$anonfun$1 apply$mcV$sp
    [dlog] > test-proxy-writer
    [dlog] >


    // Output of `TailReader`
    Opening log stream basic-stream-9
    Log stream basic-stream-9 is empty.
    Wait for records starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
    Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
    """
    test-proxy-writer
    """