Round Robin Dequeuer Sample

The following discussion examines the output processing module, which is defined in systemtables.ssql. Bracketed numbers on the right are not part of the code, but instead link to lines of commentary below.

create input stream subscriptions(connectionid string, path string,  [1]
  logicalPath string, subscribe boolean, time timestamp);

create output memory table subscriptionstab (id long,                [2]
  connectionid string,   
  path string, logicalPath string, time timestamp,  
  primary key(connectionid, logicalPath) using btree);

create index subidindex on subscriptionstab using btree (id);        [3]

create memory table serialid(id int, serialnum long, 
  primary key(id) using hash);                                       [4]

create stream subscriptionsinsert as select * from subscriptions     [5]
  where subscribe == true;

create stream subscriptionswithid;                                   [6]

insert into serialid (id, serialnum) select 1 as id,                 [7]
  long(1) as serialnum        
  from subscriptionsinsert
    on duplicate key update serialnum = serialid.serialnum+1
    returning subscriptionsinsert.*, 
    serialid.serialnum as subscriptionid 
    INTO subscriptionswithid;

insert into subscriptionstab select subscriptionid,                  [8]
  connectionid, path, logicalPath,   
  time from subscriptionswithid;

create stream subscriptionsdelete as select * from subscriptions     [9]
  where subscribe == false;

delete from subscriptionstab using subscriptionsdelete              [10]
  where subscriptionstab.connectionid = subscriptionsdelete.connectionid 
  and subscriptionstab.logicalPath = subscriptionsdelete.logicalPath;

The following annotations describe points of interest in the systemtables module:

This section examines the StreamSQL version of dequeuer application itself, roundrobin.ssql, which contains the systemtables module.

create input stream subscriptions(connectionid string, path string,      [1]
  logicalPath string, subscribe boolean, time timestamp);

create table subscriptionstab;                                           [2]

APPLY MODULE "systemtables.ssql" FROM subscriptions=subscriptions        [3]
 INTO subscriptionstab=subscriptionstab;

create input stream in(a int);                                           [4]

create output stream out(a int, nextsubid long, nextconnectionid string) [5]
    with output predicate system.connectionid = nextconnectionid;

declare lastsubid long default 0                                         [6]
    update from (select nextsubid from out);

create output stream nextid AS                                           [7]
  select in.*, 
      subscriptionstab.id as nextsubid, 
      subscriptionstab.connectionid as nextconnectionid
  from in outer join subscriptionstab
  where subscriptionstab.id > lastsubid and path = getPath('out')
    order by id limit 1;
    
select nextid.a,                                                         [8]
  coalesce(nextid.nextsubid, subscriptionstab.id) as nextsubid, 
  coalesce(nextid.nextconnectionid, subscriptionstab.connectionid) 
    as nextconnectionid
    from nextid, subscriptionstab
    where path = getPath('out')
    order by id limit 1 into out;

The following annotations describe points of interest in the roundrobin.ssql application: