docs/tutorials/basic-5.rst

---
title: API - Read Records
top-nav-group: quickstart
top-nav-pos: 4
top-nav-title: API - Read Records
layout: default
---

.. contents:: Basic Tutorial - Tail reading records from a stream

Tail reading records from a stream
==================================

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

.. sectnum::

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

Open the stream to start read the records.

::

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


Read the records
----------------

Read the next available record from the stream. The future is satisified when the record is available.

::

        Future<LogRecordWithDLSN> readFuture = reader.readNext();


Register the read callback
---------------------------

Register a future listener on read completion. The reader will be notified once the record is ready for consuming.

::

        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());


Run the tutorial
~~~~~~~~~~~~~~~~

Run the example in the following steps:

Start the local bookkeeper cluster
----------------------------------

You can use follow command to start the distributedlog stack locally.
After the distributedlog 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
---------------------

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

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


Tail the stream
---------------

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


Write records
-------------

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


Check the results
-----------------

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