EventFlow Applications and LiveView: Publishers

First, if you have not already done so, load the Hello LiveView sample. Follow the steps described here. The embedded publisher sample is based on the Hello LiveView sample, so it is useful to have the Hello LiveView sample available for review.

Now, load and run the "Shows an embedded publisher application" sample. Follow these steps:

Adding an embedded publisher to LiveView requires a LiveView configuration file describing the publisher. In this sample, the lvconf file called ItemsPublisher.lvconf configures the publisher. The root element of this configuration file is <liveview-configuration>, as is true for all LiveView configuration files.

The <publisher> element contains attributes that specify the embedded publisher name, description, and the location of the StreamBase application writes to the LiveView tables. The publisher element of the ItemsPublisher.lvconf file is shown here:

<!-- This file configures an embedded publisher for the Items and Categories tables. -->
    <publisher 
        id="ItemsPublisher"
        description="Publishes to the Items table"
        short-description="Items Table Publisher"
        filename="ItemsPublisher.sbapp"
        sub-folder="ItemsPublisher">
        
        <!-- The LiveView tables this publisher publishes to, in this case Items and Categories. -->
        <tables>
            <table table-ref="Items"></table>
            <table table-ref="Categories"></table>
        </tables>
        
    </publisher>

These attributes of the <publisher> element are:

Th <tables> element is a child of the <publisher> element. This element contains one or more <table> child elements . The <table> attribute table-ref is the name of the project table to which the publisher sends data. The <tables> element of the ItemsPublisher.lvconf file is shown here. It will publish to the Items table and the Categories table.

<tables>
   <table table-ref="Items"></table>
   <table table-ref="Categories"></table>
</tables>

The publisher configures the input and output streams for publication in the project tables. When LiveView Server starts it establishes the connections between the project containers so that data is published in the selected tables. You can see how this works by looking at how the project containers are connected. To do this, open a StreamBase command-line prompt. The sbc list -C ItemsPublisher container-connections command shows all container connections to the ItemsPublisher container:

>sbc list -C ItemsPublisher container-connections
container-connections   Categories.DataIn=ItemsPublisher.PublishToCategoriesOut
container-connections   Categories.QueryTheLastGoodRecord=ItemsPublisher.QueryTheLastGoodRecordOut
container-connections   Items.DataIn=ItemsPublisher.PublishToItemsOut
container-connections   Items.QueryTheLastGoodRecord=ItemsPublisher.QueryTheLastGoodRecordOut
container-connections   ItemsPublisher.ServerStatusIn=default.ServerStatus
container-connections   ItemsPublisher.TheLastGoodRecordIn=Categories.TheLastGoodRecordOut
container-connections   ItemsPublisher.TheLastGoodRecordIn=Items.TheLastGoodRecordOut

The container-connections connect destination streams to source streams. So, for example, the statement:

Items.DataIn=ItemsPublisher.PublishToItemsOut

assigns the output stream PublishToItemsOut to the Items table stream DataIn. Every LiveView table receives data from a DataIn stream.

An embedded publisher can publish data to tables without those tables being configured to call the publisher. This means that the lvconf files for individual tables do not need to have data-sources individually configured. Compare Items.lvconf from the Hello LiveView sample to Items.lvconf in the publisher sample. In the publisher sample, the Items table gets data from the configured publisher. The table's lvconf file does not have a data-source configured:

<data-table table-space-ref="PersistentTableSpace" id="Items"
            description="Live view of all items, quantity in stock and the last sold price"
            short-description="Quantity and last-sold prices">
  <fields>
    <!-- The publisher sets these fields -->
    <field name="sku" type="string"></field>
    <field name="category" type="string"></field>
    <field name="color" type="string"></field>
    <field name="quantityRemaining" type="int"></field>
    <field name="lastSoldPrice" type="double"></field>
  </fields>
  <primary-key>
    <field ref="sku" />
  </primary-key>
</data-table>

In contrast, the Items table in the Hello LiveView sample receives data from a data-source, in this case, an application called from the project folder ItemsDataSource. The Hello LiveView Items.lvconf file looks like this:

<data-table table-space-ref="DefaultTableSpace" id="Items"
            description="Live view of all items, quantity in stock and the last sold price"
            short-description="Quantity and last-sold prices">
  <fields>
    <field name="sku" type="string"></field> <!-- The application in the data-source tag sets this field -->
    <field name="category" type="string"></field>
    <field name="color" type="string"></field>
    <field name="quantityRemaining" type="int"></field> <!-- see the comment for 'sku' -->
    <field name="lastSoldPrice" type="double"></field> <!-- see the comment for 'sku' -->
  </fields>
  <primary-key>
    <field ref="sku" />
  </primary-key>
<!-- This declares that an output stream from our data source application (defined
     in ItemsDataSource.lvconf) feeds this table. This application is started as
     soon as all LiveView tables have loaded. -->
  <data-sources>
    <data-source>
      <application output-stream="Data" application-ref="ItemsDataSource" />
    </data-source>
  </data-sources>
</data-table>

LiveView tables can be configured for persistence such that their contents are saved to disk in log-based files and recovered from these files after a crash or server restart. Part of the recovered data is the table’s publisher sequence number. During initialization, an embedded publisher typically retrieves the sequence for each persistent table and resumes publishing at that point, to avoid overwriting previously written, recovered records.

The embedded publisher interface provides two streams for retrieving the publisher sequence number:

The publisher, in requesting last good record information, has the option of specifying:

Specifying a null table name in the request retrieves information for all tables being published to, while specifying a null publisher ID requests the last good record for all previous publishers to the table(s).

In response to a “query the last good record” request, LiveView Server returns one or more tuples for each table specified in the request. If the table name is null in the request, the server returns responses for all tables being published to; otherwise it returns responses for just the specified table.

The last response tuple returned by LiveView Server for each table is a punctuation tuple, which does not carry last good record information (all sequence number fields are null) and is identified by a null in the PublisherID field. All tuples returned by the server, including punctuation tuples, have a non-null Tablename field.

Thus, in response to a “query the last good record” request, the publisher should expect either one punctuation tuple, if the Tablename field was non-null in the request, or one punctuation tuple per table the publisher is configured to publish to, if the Tablename field was null in the request. The number of tables the publisher is configured to publish to is equal to the number of top-level fields present in the PublishSchemasIn input stream's schema.

The following example presents a publish scenario followed by the recovery activity for each of the four “query the last good record” request combinations.

Published tuples:

(Publisher A, 1) -> Table-X
(Publisher A, 2) -> Table-Y 
(Publisher B, 3) -> Table-X 
(Publisher B, 4) -> Table-Y 
(Publisher C, 5) -> Table-X 
(Publisher D, 6) -> Table-Y

If publisher sends request with PublisherID == null, Tablename == null, server returns:

(Publisher A, Table-X, 1) 
(Publisher B, Table-X, 3) 
(Publisher C, Table-X, 5) 
(null, Table-X, null) <- punctuation table for Table-X 

(Publisher A, Table-Y, 2) 
(Publisher B, Table-Y, 4) 
(Publisher D, Table-Y, 6) 
(null, Table-Y, null) <- punctuation table for Table-Y

If publisher sends request with PublisherID == Publisher A, Tablename == null, server returns:

(Publisher A, Table-X, 1) 
(null, Table-X, null)

(Publisher A, Table-Y, 2) 
(null, Table-Y, null)

If publisher sends request with PublisherID == null, Tablename == Table-X, server returns:

(Publisher A, Table-X, 1) 
(Publisher B, Table-X, 3) 
(Publisher C, Table-X, 5) 
(null, Table-X, null) <- punctuation table for Table-X

If publisher sends request with PublisherID == Publisher A, Tablename == Table-X, server returns:

(Publisher A, Table-X, 1) 
(null, Table-X, null)

The LiveView embedded publisher sample consists of a StreamBase application, ItemsPublisher.sbapp, whose key component is an embedded input adapter, ItemsPublisher.java. The adapter source code is provided with the sample and is intended to be used as a starting point for LiveView customers creating their own embedded publishers.

Double-click the application name in Package Explorer view to open this application in the EventFlow editor.

The adapter has three input streams and three output streams corresponding to streams of the LiveView-created interface for the publisher. The first input stream, PublishSchemasIn, has two fields of type tuple that convey the schemas of the adapter’s first two output ports. No tuples flow through this stream; it is used strictly to convey schema information, allowing the adapter to set the schemas of its first two output ports, which are used to publish to the Items and Categories tables.

The second input stream, ServerStatusIn, allows the adapter to determine when the LiveView server is ready to accept published tuples. Its single input field, IsReady, contains true when the LiveView server is ready.

The third input stream and third output stream are used to retrieve sequence numbers during recovery, as described above.

The adapter emulates the data source used by the Hello LiveView sample. It creates a fixed set of 33 SKUs for which price and volume information is published to the Items table. The adapter’s Tuples per Second property determines the rate at which tuples are published. To illustrate multi-table publishing, the adapter publishes to a second table, Categories, a mapping of category names to descriptions. A second property, Log Level, controls the adapter’s verbosity of logging.

During initialization, the adapter’s typecheck() method validates its property values and the schemas of its input ports, sets the schemas of its output ports, and creates cached tuples for publishing to the Items and Categories tables. The adapter’s init() method then generates the 33 SKUs used for publishing to the Items table and registers its run() method, which executes once the adapter completes initialization.

The adapter’s processTuple() method is invoked when a tuple is received on one of its input ports. When a server status tuple arrives indicating the LiveView server is ready, the adapter sets a flag that signals its run() method to begin publishing.

The adapter’s run() method executes once initialization is complete. It polls the LiveView server ready flag used to commence publishing, sleeping in between to avoid a CPU spin loop. Once LiveView is ready, the adapter emits a tuple on its QueryTheLastGoodRecordOut port to retrieve the Items table’s sequence number at which to resume publishing. The Items table, being persistent, might contain previously-written, recovered records that shouldn’t be overwritten. The response to the “query the last good request” comes in via a tuple to the adapter’s processTuple() method on its TheLastGoodRecordIn port. The lowest persisted sequence number from this table is saved and subsequently used by the run() method in publishing to the Items table. The adapter’s run() method publishes once to the (static) Categories table a mapping of names to descriptions.

The adapter then begins publishing to the Items table. The adapter publishes at the rate specified by its Tuples per Second property. During each one-second interval, it attempts to spread the publishing of tuples evenly across the remaining time in this interval. For example, if the adapter is configured to publish 10 tuples per second, it sleeps approximately 100 ms between publishing each tuple. The sleep time is adjusted during the interval to account for the time spent doing the actual publishing.

Creating an embedded publisher involves the following three steps:

This section shows you how to modify the embedded publisher sample application by having it publish to a third LiveView table. It assumes you have loaded the sample in Studio, as described above.

You can now open the Items Publisher input adapter source module, ItemsPublisher.java, and add code to the run method to publish to the MyTable LiveView table by emitting tuples on the new output port.