Memory Queue Connector

The Memory Queue (MemQueue) Connector provides a connector-like functionality to read and write to the memory queue feature (aka. MemBufferQ). This is an alternative to writing script to access a memory queue and is an extension of the Memory Queue Function Component (function component).

The objects used to communicate between components are not persistent and are not capable of handling large return sets. For example, large data returned by an ldapsearch operation. In order to solve this problem, an internal threadsafe memory queue can be used as a communications data structure between AL components. It can contain embedded logic that would trigger whenever buffer is x% full/empty/data available.

There can be multiple readers and writers for the same queue. Every writer has to obtain a lock before adding data. The writer has to release lock before a reader can access it. Connectors in Iterator mode have a parameter that determines when the read lock is released - After single read, on AL cycle end or Connector close.

This Connector supports AddOnly and Iterator modes only.

This Connector can also be used in connection with MemQueue pipes set up from JavaScript, although it is important to note that a MemQueue pipe created by the MemQueue Connector will be terminated when the Connector closes.

The Memory queue buffer is a FIFO type of data structure, where adding and reading can occur simultaneously. It works as a pipe where additions happen at one end and reading happens at the other end and reading removes the data from queue.

The Memory queue buffer provides overflow storage using the System Store when a threshold value is reached, which is a function of the runtime memory available.

Memory queue components

Paged memory buffer queue

A queue type buffer having the following functions:

1. Read and write
2. Finding the size
3. Registering and unregistering callback triggers
4. Generating triggers by calling callback methods
5. Option for setting the system store to use for paging

Watermark

This is the maximum size of the queue; refer to the configuration page for operational details.

Pages

The connector is able to buffer a set of objects before writing them to the System Store. The buffer it uses is called a page. Users can specify the number of the objects each page may contain. This is done using the pagesize parameter. Using pages helps the connector more efficiently read/write objects from/to the System Store. This option is used only used when paging is used; when paging support is switched off, the queue is not divided into pages, but is purely a sequence of elements.

Threshold

This is the number of pages that can exist in the memory and beyond which pages must be flushed to the system store. This is calculated depending on the page size entered by the user and runtime available memory.

System Store

Database that stores the pages when the threshold is reached.

Global Lookup Table

A global table that can store memory buffer objects. This is to support a named pipe mechanism of sharing between threads. A thread can lookup a memory buffer queue using a name in the table, if it exists a reference to the queue object is returned to the user else a new queue object is created with that name and added to the GLT.

High level workflow

The Memory queue Buffer is a queue of pages containing objects. When a particular threshold (the "watermark") is reached, a new thread is created that starts writing to another buffer of pages; when a page is full, it transfers the page to either to the main queue or to the system store. When a page is read from the main queue, one page is transferred from system store to the main queue; in doing it also deletes that page from the system store.

Configuration

Instance

Name of the Config instance. Current instance is assumed if it is null.

Queue

Name of the queue or pipe which is to be created.

Read timeout

The interval in milliseconds to wait for, before control returns, if no entries were found in the queue.

Iterator Read Lock Release

When the connector is in iterator mode, this determines when the read lock on the specified memory queue will be released. You can select one of these values: On Single Read (default), AL cycle end and Connector close.

Percent memory to use

This determines what percentage of memory can be utilized by the memory queue. The default value is 50.

Watermark

This is the threshold at which objects are persisted to the System Store. Note that the Page Size determines when pages are actually written, so the Watermark should be a multiple of the Page Size.

Page Size

Number of entries in one page. The default value is 100.

Database name

System Store database name: a JDBC URL to a System Store database (or blank for the default System Store).

Username

Username to connect to the System Store database.

Password

Password to use when signing on to the database.

Table name

Name of System Store table to use for paging.

Detailed Log

Check for detailed log messages.

Accessing the Memory Queue programmatically

The Memory Queue can be accessed directly from JavaScript, not only through the Connector.

1. To create new pipe - There are two methods for this.
   1. Paging disabled - newPipe(String instName,String pipeName,int watermark) // Does not require any DB related entries
   2. Paging enabled - newPipe(String instName,String pipeName,int watermark,int pagesize) // Requires DB initialization
   An example script with paging enabled:

   var memQ=system.newPipe( "inst","Q1",1000,10) ; 
   memq.initDB(dbName, jdbcLogin, jdbcPassword, tableName);  // Required to Initialize DB 
   memQ.write(conn);

2. getPipe(String instName,String pipeName)
3. purgeQueue()

   An example script would look something like this:

   var q =system.getPipe("Inst1","Q1") ;
   q.purgeQueue();

4. deletePipe(String pipeName)

   Example:

   var q =system.getPipe("Inst1","Q1");
   q.deletePipe();

The following is an example script to read from the Memory Queue using API calls:

var memQ=system.getPipe( "inst","Q1") ; 
var size=memQ.size();

for(var count=0;i<=size;count++){
	main.logmsg(memQ.read());
}