Streams Swiftlet

Overview

The SwiftMQ Streams Swiftlet provides an embedded JavaScript-based stream processing engine within the SwiftMQ router. It allows users to deploy, configure, and execute stream processing scripts that interact with messages, queues, topics, and other resources in real-time. Streams are managed, monitored, and orchestrated directly inside the router, supporting advanced event processing patterns and integrations.

Features

JavaScript Stream Processing

Run JavaScript scripts inside the router to process messages in real-time, with access to a rich API for messaging, memory, timers, JDBC, mail, and more.

Declarative Stream Configuration

Streams are organized into domains, packages, and streams, with each stream configured via attributes and parameters in the router configuration.

Resource Factories

Streams can dynamically create and manage inputs, outputs, memories, timers, mail servers, JDBC lookups, and temporary queues using builder-style APIs.

Memory and MemoryGroup Abstractions

Support for in-memory, persistent (queue-based), and temporary queue-based message storage, with grouping, indexing, limits, and inactivity timeouts.

Management and Monitoring

Streams can consume management events, monitor router resources, and expose usage metrics for all active streams, inputs, outputs, timers, and more.

Hot-Deployable Libraries

External JARs can be hot-deployed per stream via the stream-lib-directory, supporting custom Java integrations.

Bundled System Streams

Includes ready-to-use system streams for last-value caching, mail output, message scheduling, route announcement, stream monitoring, stream registry, and script repository.

Stream API Reference

Stream Object

• stream.onStart(callback) — Registers a callback to be executed when the stream starts.
• stream.onStop(callback) — Registers a callback to be executed when the stream stops.
• stream.onMessage(callback) — Registers a callback to be executed for each incoming message.
• stream.onException(callback) — Registers a callback to be executed when an exception occurs in the stream.
• stream.create() — Returns a builder object to create resources such as inputs, outputs, memories, timers, mailservers, jdbcLookups, and tempQueues.
• stream.memory(name) — Returns the Memory instance with the given name, or null if not present.
• stream.memoryGroup(name) — Returns the MemoryGroup instance with the given name, or null if not present.
• stream.timer(name) — Returns the Timer instance with the given name, or null if not present.
• stream.input(name) — Returns the Input instance with the given name, or null if not present.
• stream.output(name) — Returns the Output instance with the given name, or null if not present.
• stream.mailserver(name) — Returns the MailServer instance with the given name, or null if not present.
• stream.jdbcLookup(name) — Returns the JDBCLookup instance with the given name, or null if not present.
• stream.tempQueue(name) — Returns the TempQueue instance with the given name, or null if not present.
• stream.purgeOutputs() — Closes all Outputs that were not used between the last and this call.
• stream.current() — Returns the currently processed Message.
• stream.log() — Returns the Log object for the stream.
• stream.cli() — Returns the CLI object for executing management commands.
• stream.routerName() — Returns the name of the local router.
• stream.fullyQualifiedName() — Returns the fully qualified stream name (domain.package.name).
• stream.restartCount() — Returns the number of times the stream has been restarted.
• stream.lastException() — Returns the last exception occurred on the stream.
• stream.lastStackTrace() — Returns the formatted stack trace of the last exception.
• stream.async(interfaceClassName, callback) — Wraps an async callback with a proxy that runs the callback on the stream's event queue.
• stream.executeCallback(callback, context) — Executes a function callback in the stream's event queue (not supported in GraalVM).

Examples

Queue Statistics Stream (management/queuestats.js)

Demonstrates monitoring queue usage via ManagementInput, MemoryGroup, and time-based tumbling windows.

// Create a ManagementInput on sys$queuemanager/usage and receive adds and changes.
stream.create().input("sys$queuemanager/usage")
    .management()
    .selector("name not like '%$%' and name not like 'qs\_%' and messagecount is not null")
    .onAdd(function (input) {
        stream.memoryGroup("queues").add(input.current());
    })
    .onChange(function (input) {
        stream.memoryGroup("queues").add(input.current());
    });

stream.create().memoryGroup("queues", "name").inactivityTimeout().minutes(2).onCreate(function (key) {
    var queueName = "qs_" + key;
    stream.cli()
        .execute("cc sys$queuemanager/queues")
        .exceptionOff()
        .execute("new " + queueName)
        .execute("save");
    stream.create().memory(queueName).queue(queueName).limit().time().tumbling().minutes(1).onRetire(function (retired) {
        print("Queue " + key + ", avg backlog last minute = " + retired.average("messagecount"));
    });
    return stream.memory(queueName);
}).onRemove(function (key) {
    stream.cli()
        .execute("cc sys$queuemanager/queues")
        .execute("delete qs_" + key)
        .execute("save");
});

stream.create().timer("queues").interval().minutes(1).onTimer(function (timer) {
    stream.memoryGroup("queues").checkLimit();
});

Nuclear Powerplant Temperature Monitor (nuclear/tempmonitor.js)

Shows use of multiple memories, sliding windows, time windows, and sending critical alerts via email.

// Checking parameters
var inputQueue = parameters.require("input-queue");
var mailHost = parameters.optional("mail-host", "localhost");
var mailUser = parameters.require("mail-user");
var mailPassword = parameters.require("mail-password");
var mailFrom = parameters.require("mail-from");
var mailTo = parameters.require("mail-to");
var mailBcc = parameters.require("mail-bcc");

stream.create().memory("all").heap().limit().time().tumbling().seconds(10).onRetire(function (retired) {
    stream.log().info("AVERAGE TEMP LAST 10 SECS=" + retired.average("temp") +
        ", MIN=" + retired.min("temp").property("temp").value().toObject() +
        ", MAX=" + retired.max("temp").property("temp").value().toObject());
    stream.log().info(retired.values("temp"));
});
stream.create().memory("warning").heap().limit().count(2).sliding();
stream.create().memory("critical").heap().limit().count(4).sliding();

stream.create().mailserver(mailHost).username(mailUser).password(mailPassword).connect();

stream.create().input(inputQueue).queue().onInput(function (input) {
    input.current().property("temp").set(input.current().property("TEMP").value().toInteger());
});

stream.create().timer("monitor").interval().seconds(10).onTimer(function (timer) {
    stream.memory("all").checkLimit();
});

stream.onMessage(function () {
    stream.memory("all").add(stream.current());
    stream.memory("warning").add(stream.current());
    stream.memory("critical").add(stream.current());
    if (stream.memory("warning").select("temp > 100").size() == 2)
        stream.log().warning("LAST 2 TEMPS > 100!");
    if (stream.memory("critical").size() == 4 &&
        stream.memory("critical").first().property("temp").value().toDouble() > 400 &&
        stream.memory("critical").ascendingSeries("temp") &&
        stream.memory("critical").last().property("temp").value().toDouble() >
        stream.memory("critical").first().property("temp").value().toDouble() * 1.5) {
        stream.log().error("WE HAVE A SPIKE!!!");
        stream.mailserver(mailHost)
            .email()
            .from(mailFrom)
            .to(mailTo)
            .bcc(mailBcc)
            .subject("Nuclear Powerplant Monitor - CRITICAL!")
            .body("Temp spike detected, last temps=" + stream.memory("critical").values("temp"))
            .send();
    }
});

Order Aggregator (ordercollect/ordercollector.js)

Implements the Enterprise Integration Pattern 'Aggregator' by joining order head and order position messages and outputting completed orders as XML.

// 3 queues are required: orderhead, orderpos, and the output
var orderHeadQueue = parameters.require("orderhead-queue");
var orderPosQueue = parameters.require("orderpos-queue");
var outputQueue = parameters.require("output-queue");

stream.create().memory("orderhead").queue("stream_ordercollector_memory_orderhead").createIndex("ORDERHEADID");
stream.create().memory("orderpos").queue("stream_ordercollector_memory_orderpos").createIndex("ORDERHEADID");
stream.memory("orderhead").clear();
stream.memory("orderpos").clear();

stream.create().input(orderHeadQueue).queue().onInput(function (input) {
    stream.memory("orderhead").add(input.current());
});

stream.create().input(orderPosQueue).queue().onInput(function (input) {
    input.current().property("ITEMNO").set(input.current().body().get("ITEMNO").toInteger());
    input.current().property("QTY").set(input.current().body().get("QTY").toInteger());
    stream.memory("orderpos").add(input.current());
});

stream.create().timer("purger").interval().hours(1).onTimer(function (timer) {
  stream.cli()
      .exceptionOff()
      .execute("cc sys$queuemanager/usage")
      .execute("remove " + outputQueue + " *");
});

stream.create().output(outputQueue).queue();

stream.onMessage(function () {
    var orderHeadId = stream.current().property("ORDERHEADID").value().toInteger();
    var orderHeadMem = stream.memory("orderhead").index("ORDERHEADID").get(orderHeadId);
    if (orderHeadMem.size() == 1) {
        var orderPosMem = orderHeadMem.join(stream.memory("orderpos"), "ORDERHEADID");
        if (orderHeadMem.first().property("NPOSITIONS").value().toInteger() == orderPosMem.size()) {
            var orderHead = orderHeadMem.first();
            var msg = "<order orderheadid=\"" + orderHead.body().get("ORDERHEADID").toStringValue() + "\"" +
                " accountno=\"" + orderHead.body().get("ACCOUNTNO").toStringValue() + "\">\n";
            orderPosMem.forEach(function (message) {
                msg += "  <position itemno=\"" + message.property("ITEMNO").value().toString() + "\"" +
                    " quantity=\"" + message.property("QTY").value().toString() + "\"/>\n";
            });
            msg += "</order>";
            stream.log().info(msg);
            stream.output(outputQueue).send(stream.create().message().textMessage().persistent().body(msg));
            stream.memory("orderhead").index("ORDERHEADID").remove(orderHeadId);
            stream.memory("orderpos").index("ORDERHEADID").remove(orderHeadId);
        }
    }
});

Simple Request/Reply Service (replier/replier.js)

Implements a temp queue-based echo service, registering the temp queue in JNDI for JMS clients.

// A simple echo service that creates an input on a temp queue where it receives its requests
// The temp queue is registered in JNDI so that JMS clients can look it up and send requests to it
stream.create().input(stream.create().tempQueue("requestQueue").registerJNDI()).queue();

stream.onMessage(function () {
    stream.create().output(null).forAddress(stream.current().replyTo())
        .send(stream.create().message().textMessage().correlationId(stream.current().messageId())
            .body("RE: " + stream.current().body())).close();
});

Bundled System Streams

lastvalue.js

Implements a last-value cache for a topic, storing and distributing the most recent value per key to new subscribers.

Parameters: last-value-property, last-value-value, expiration, persistent, topic-name, verbose, store-queue

mailout.js

Receives messages from a queue and sends them as emails using a configured SMTP server.

Parameters: input-queue, servername, username, password, default-from, default-to, default-cc, default-bcc, default-subject

messagescheduler.js

Schedules messages for delayed delivery to queues or topics, supporting both new and legacy scheduling properties.

Parameters: input-queue, store-queue, delay-format, interval, purge-interval, max-batch-size

routeannouncer.js

Announces active routes (routers) on a topic and responds to initialization requests from clients.

Parameters: topic

streammonitor.js

Monitors all streams registered at the stream registry and logs all messages received from them.

Parameters: registry-topic

streamregistry.js

Maintains a registry of available streams, handling registration, unregistration, and initialization requests.

Parameters: topic

streamrepository.js

Implements a repository for storing, listing, and removing JavaScript stream scripts, backed by persistent memory.

Parameters: input-queue, store-queue

Sample Applications

Queue Statistics Monitoring

Monitors queue statistics using management input.

queuestats.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2017, All Rights Reserved
 */

// Create a ManagementInput on sys$queuemanager/usage and receive adds and changes.
// No need to set a onRemove callback as these Memories time out by the inactivity timeout.
// The selector filters those queue which are not system queues and not our own queue (qs_)
// and the message must contain the messagecount property (other property changes are not of interest).
stream.create().input("sys$queuemanager/usage")
    .management()
    .selector("name not like '%$%' and name not like 'qs\_%' and messagecount is not null")
    .onAdd(function (input) {
        // A new queue has been created.
        // Add it to the memory group.
        stream.memoryGroup("queues").add(input.current());
    })
    .onChange(function (input) {
        // Property "messagecount" has changed.
        // Add it to the MemoryGroup
        stream.memoryGroup("queues").add(input.current());
    });

// Create a MemoryGroup for all queues and register a callback for new groups
stream.create().memoryGroup("queues", "name").inactivityTimeout().minutes(2).onCreate(function (key) {
    // This is an example on how to use these callbacks. You can create whatever type of Memories.
    // Here we create a QueueMemory
    var queueName = "qs_" + key;

    // We need to create a regular queue as a backstore for the QueueMemory
    stream.cli()
        .execute("cc sys$queuemanager/queues")
        .exceptionOff()  // Queue might be already defined
        .execute("new " + queueName)
        .execute("save");

    // A new value of the group is received and we create a Memory for it
    // with a 1 min tumbling time window
    stream.create().memory(queueName).queue(queueName).limit().time().tumbling().minutes(1).onRetire(function (retired) {
        // Window is closed, we print the average message count over the last minute
        print("Queue " + key + ", avg backlog last minute = " + retired.average("messagecount"));
    });

    // Return the new memory
    return stream.memory(queueName);
}).onRemove(function (key) {
    // Memory has been removed. Delete the corresponding persistent queue
    stream.cli()
        .execute("cc sys$queuemanager/queues")
        .execute("delete qs_" + key)
        .execute("save");
});

// Create a timer that calls checkLimit on the memory group each minute
stream.create().timer("queues").interval().minutes(1).onTimer(function (timer) {
    stream.memoryGroup("queues").checkLimit();
});

Temperature Producer/Monitor (CEP)

Complex event processing with temperature monitoring.

tempmonitor.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 */

// Checking parameters
var inputQueue = parameters.require("input-queue");
var mailHost = parameters.optional("mail-host", "localhost");
var mailUser = parameters.require("mail-user");
var mailPassword = parameters.require("mail-password");
var mailFrom = parameters.require("mail-from");
var mailTo = parameters.require("mail-to");
var mailBcc = parameters.require("mail-bcc");

// Create 3 Memories, one for monitoring (all), one to detect warning, one to detect critical conditions
stream.create().memory("all").heap().limit().time().tumbling().seconds(10).onRetire(function (retired) {
    stream.log().info("AVERAGE TEMP LAST 10 SECS=" + retired.average("temp") +
        ", MIN=" + retired.min("temp").property("temp").value().toObject() +
        ", MAX=" + retired.max("temp").property("temp").value().toObject());
    stream.log().info(retired.values("temp"));
});
stream.create().memory("warning").heap().limit().count(2).sliding();
stream.create().memory("critical").heap().limit().count(4).sliding();

// Create a mail server to send critical mails
stream.create().mailserver(mailHost).username(mailUser).password(mailPassword).connect();

// Create the Inputs
stream.create().input(inputQueue).queue().onInput(function (input) {
    // We need a property "temp" instead of "TEMP"
    input.current().property("temp").set(input.current().property("TEMP").value().toInteger());
});

// Create a timer to trigger the retirement of the "all" memory every 10 secs
stream.create().timer("monitor").interval().seconds(10).onTimer(function (timer) {
    stream.memory("all").checkLimit();
});

// Set the onMessage callback
stream.onMessage(function () {
    // Add current message to all memories
    stream.memory("all").add(stream.current());
    stream.memory("warning").add(stream.current());
    stream.memory("critical").add(stream.current());

    // Check if we have a warning condition (last 2 temps > 100)
    if (stream.memory("warning").select("temp > 100").size() == 2)
        stream.log().warning("LAST 2 TEMPS > 100!");

    // Check if we have a critical condition (spike):
    // At least 4 messages received,
    // First temp > 400,
    // All temps are greater than the temp before (ascendingSeries),
    // Last temp is 1.5 times greater than the first temp.
    if (stream.memory("critical").size() == 4 &&
        stream.memory("critical").first().property("temp").value().toDouble() > 400 &&
        stream.memory("critical").ascendingSeries("temp") &&
        stream.memory("critical").last().property("temp").value().toDouble() >
        stream.memory("critical").first().property("temp").value().toDouble() * 1.5) {

        stream.log().error("WE HAVE A SPIKE!!!");
        stream.mailserver(mailHost)
            .email()
            .from(mailFrom)
            .to(mailTo)
            .bcc(mailBcc)
            .subject("Nuclear Powerplant Monitor - CRITICAL!")
            .body("Temp spike detected, last temps=" + stream.memory("critical").values("temp"))
            .send();
    }

});

tempproducer.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 */

// Check parameters
var outputQueue = parameters.require("output-queue");

// Create the Output to send the temp messages
stream.create().output(outputQueue).queue();

// Static temp data that will generate a warning and critical condition
var temps = [90, 97, 101, 80, 90, 110, 120, 200, 250, 500, 450, 320, 401, 402, 450, 800];
var idx = 0;

// Timer to send temps in a 2 secs interval
stream.create().timer("ticker").interval().seconds(2).onTimer(function (timer) {
    stream.log().info("Sending temp: " + temps[idx]);
    var msg = stream.create().message().message().nonpersistent().property("TEMP").set(temps[idx++]);
    if (idx == temps.length)
        idx = 0;
    stream.output(outputQueue).send(msg);
});

Order Collector

Aggregator pattern collecting orders into batches.

itemstats.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 *
 * This example demonstrates: WireTap, Management Input, Timer reconfiguration.
 * It creates a wiretap into the orderpos queue and generates a statistic.
 */

// Parameters
var orderPosQueue = parameters.require("orderpos-queue");
var statIntervalSec = parameters.require("statistic-interval-sec");

// Callback function for onRetire to print the statistics
function printStatistic(retired) {

    // Generate statistics in the streams's log file
    stream.log().info("Item Statistics:");

    // Get and log the statistic data (items with summarized quantity in descending order)
    retired.group("ITEMNO").sum("QTY").sort("QTY").reverse().forEach(function (message) {
        stream.log().info(message.property("ITEMNO").value().toString() + " = " + message.property("QTY").value().toString());
    });
}

// Create the itemstats memories for the item statistics
stream.create().memory("itemstats").heap().limit().time().tumbling().seconds(statIntervalSec).onRetire(printStatistic);

// Create a timer that triggers the item statistics every n secs
stream.create().timer("stats").interval().seconds(statIntervalSec).onTimer(function (timer) {
    // Reduce the memory according to the limit
    stream.memory("itemstats").checkLimit();
});


// Create a wiretap input on the orderpos queue.
stream.create().input(orderPosQueue).wiretap("w1").onInput(function (input) {
    // We need ITEMNO and QTY as a property
    input.current().property("ITEMNO").set(input.current().body().get("ITEMNO").toInteger());
    input.current().property("QTY").set(input.current().body().get("QTY").toInteger());

    stream.memory("itemstats").add(input.current());
});

// Create a Management Input that receives changes on my own statistic-interval-sec parameter
// and reconfigures the Timer
stream.create().input("sys$streams/domains/swiftmq/packages/samples/streams/" + stream.name() + "/parameters/statistic-interval-sec").management().onChange(function (input) {

    // Get the new value
    var secs = input.current().property("value").value().toInteger();

    // Reset and reconfigure the Timer with the new value
    stream.timer("stats").reset().seconds(secs).reconfigure();

    // Recreate the itemstats Memory
    stream.memory("itemstats").close();
    stream.create().memory("itemstats").heap().limit().time().tumbling().seconds(secs).onRetire(printStatistic);

    // Log it into the Logfile
    stream.log().info("Statistic interval reconfigured to " + secs + " seconds");

});

ordercollector.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 *
 * This example implements the EI-Pattern "Aggregator". It receives OrderHead
 * and OrderPos messages from different queues and generates a XML messages
 * of the full order once completely received.
 */

// 3 queues are required: orderhead, orderpos, and the output
var orderHeadQueue = parameters.require("orderhead-queue");
var orderPosQueue = parameters.require("orderpos-queue");
var outputQueue = parameters.require("output-queue");


// Create the memories: orderhead, orderpos
stream.create().memory("orderhead").queue("stream_ordercollector_memory_orderhead").createIndex("ORDERHEADID");
stream.create().memory("orderpos").queue("stream_ordercollector_memory_orderpos").createIndex("ORDERHEADID");
stream.memory("orderhead").clear();
stream.memory("orderpos").clear();

// Create the inputs
stream.create().input(orderHeadQueue).queue().onInput(function (input) {
    // Add message to orderhead memory
    stream.memory("orderhead").add(input.current());
});

stream.create().input(orderPosQueue).queue().onInput(function (input) {
    // We need ITEMNO and QTY as a property
    input.current().property("ITEMNO").set(input.current().body().get("ITEMNO").toInteger());
    input.current().property("QTY").set(input.current().body().get("QTY").toInteger());

    // Add message to orderpos memory
    stream.memory("orderpos").add(input.current());
});

// Purge output queue every hour
stream.create().timer("purger").interval().hours(1).onTimer(function (timer) {
  stream.cli()
      .exceptionOff()
      .execute("cc sys$queuemanager/usage")
      .execute("remove " + outputQueue + " *");
});

// Create an Output to send the collected Orders
stream.create().output(outputQueue).queue();

// Create the onMessage callback
stream.onMessage(function () {

    // Get the orderhead id from the current message (may be a orderhead or orderpos message)
    var orderHeadId = stream.current().property("ORDERHEADID").value().toInteger();

    // Get all orderheads we currently have for this orderhead id
    var orderHeadMem = stream.memory("orderhead").index("ORDERHEADID").get(orderHeadId);

    // Check if we have one orderhead for this orderhead
    if (orderHeadMem.size() == 1) {

        // Join it with the orderpos memory
        var orderPosMem = orderHeadMem.join(stream.memory("orderpos"), "ORDERHEADID");

        // Check if we have all orderpositions mentioned in the orderhead
        if (orderHeadMem.first().property("NPOSITIONS").value().toInteger() == orderPosMem.size()) {

            // Generate XML message with the completed order
            var orderHead = orderHeadMem.first();
            var msg = "<order orderheadid=\"" + orderHead.body().get("ORDERHEADID").toStringValue() + "\"" +
                " accountno=\"" + orderHead.body().get("ACCOUNTNO").toStringValue() + "\">\n";

            orderPosMem.forEach(function (message) {
                msg += "  <position itemno=\"" + message.property("ITEMNO").value().toString() + "\"" +
                    " quantity=\"" + message.property("QTY").value().toString() + "\"/>\n";

            });

            msg += "</order>";

            // Log it into the stream's log file
            stream.log().info(msg);

            // Send it to the output queue as persistent text message
            stream.output(outputQueue).send(stream.create().message().textMessage().persistent().body(msg));

            // Remove the messages with this orderhead id from orderhead and orderpos memories
            stream.memory("orderhead").index("ORDERHEADID").remove(orderHeadId);
            stream.memory("orderpos").index("ORDERHEADID").remove(orderHeadId);
        }
    }
});

orderheadproducer.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 */

// We need 1 output queue
var outputQueue = parameters.require("output-queue");

// Create the Output to send Orderhead messages
stream.create().output(outputQueue).queue();

// Static data for the orderhead messages
var accounts = [1513, 271, 3300, 8800, 17777, 55167, 99220, 1513, 8800, 1900, 3344, 19220];
var orderheadids = [1011, 998, 1567, 3318, 90001, 44526, 33900, 1444, 9344, 1788, 9900, 14144];
var npositions = [3, 5, 1, 5, 6, 1, 1, 4, 8, 7, 3, 5];
var idx = 0;

// Create a timer that sends an orderhead map message every 5 secs
stream.create().timer("ticker").interval().seconds(5).onTimer(function (timer) {
    if (idx < accounts.length) {
        stream.output(outputQueue).send(stream.create().message().mapMessage().persistent()
            .property("ACCOUNTNO").set(accounts[idx])
            .property("ORDERHEADID").set(orderheadids[idx])
            .property("NPOSITIONS").set(npositions[idx])
            .body().set("ACCOUNTNO", accounts[idx])
            .set("ORDERHEADID", orderheadids[idx])
            .set("NPOSITIONS", npositions[idx]).message());
        idx++;

    } else {
        stream.create().timer("restarter").next().beginOfHour().onTimer(function (timer) {
            idx = 0;
        }).start();
    }
});

orderposproducer.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 */

// We need 1 output queue
var outputQueue = parameters.require("output-queue");

// Create the Output to send Orderpos messages
stream.create().output(outputQueue).queue();

// Static orderpos data
var orderpos = [[14144, 4711, 2],
    [14144, 3318, 3],
    [14144, 1715, 1],
    [9900, 1522, 1],
    [9900, 3318, 2],
    [9900, 4711, 1],
    [1788, 4355, 1],
    [1788, 6722, 1],
    [1011, 1677, 3],
    [998, 1544, 1],
    [998, 4711, 1],
    [1788, 2011, 4],
    [1788, 1988, 1],
    [1788, 3789, 1],
    [9344, 1988, 1],
    [9344, 2011, 1],
    [9344, 9211, 1],
    [998, 9211, 1],
    [998, 3318, 1],
    [9344, 1031, 2],
    [9344, 1544, 1],
    [9344, 1677, 1],
    [1444, 1988, 1],
    [1444, 6722, 4],
    [90001, 1677, 3],
    [14144, 2011, 2],
    [14144, 1988, 5],
    [90001, 3318, 2],
    [90001, 2011, 5],
    [1788, 1566, 3],
    [1788, 9913, 1],
    [90001, 1677, 1],
    [90001, 4355, 2],
    [3318, 1988, 2],
    [3318, 9913, 2],
    [9344, 3318, 1],
    [9344, 4417, 4],
    [33900, 6722, 1],
    [44526, 4711, 2],
    [90001, 9211, 2],
    [3318, 4355, 2],
    [1444, 3318, 2],
    [1444, 4355, 3],
    [3318, 1544, 2],
    [3318, 1677, 2],
    [1567, 4711, 1],
    [998, 2011, 10],
    [1011, 1031, 2],
    [1011, 4444, 11]];
var idx = 0;

// Create a timer that sends one orderpos per second
stream.create().timer("ticker").interval().seconds(5).onTimer(function (timer) {
    if (idx < orderpos.length) {
        stream.output(outputQueue).send(stream.create().message().mapMessage().persistent()
            .property("ORDERHEADID").set(orderpos[idx][0])
            .property("ITEMNO").set(orderpos[idx][1])
            .property("QTY").set(orderpos[idx][2])
            .body().set("ORDERHEADID", orderpos[idx][0])
            .set("ITEMNO", orderpos[idx][1])
            .set("QTY", orderpos[idx][2]).message());
        idx++;
    }
    else {
        stream.create().timer("restarter").next().beginOfHour().onTimer(function (timer) {
            idx = 0;
        }).start();
    }
});

Request-Reply Echo Service

Echo service demonstrating request-reply pattern.

replier.js:

/**
 * Author: IIT Software GmbH, Muenster/Germany, (c) 2016, All Rights Reserved
 */

// A simple echo service that creates an input on a temp queue where it receives its requests
// The temp queue is registered in JNDI so that JMS clients can look it up and send requests to it
stream.create().input(stream.create().tempQueue("requestQueue").registerJNDI()).queue();

stream.onMessage(function () {

    stream.create().output(null).forAddress(stream.current().replyTo())
        .send(stream.create().message().textMessage().correlationId(stream.current().messageId())
            .body("RE: " + stream.current().body())).close();
});

Configuration Reference

The top-level entity in routerconfig.xml is <swiftlet name="sys$streams">.

<swiftlet name="sys$streams"> Properties

These properties are attributes of the <swiftlet name="sys$streams"> entity.

Parameter	Type	Default	Mandatory	Reboot Required	Description
collect-interval	Long	1000	No	No	Interval for collecting Stream usage information
stream-grant-predicates	String	stream\_%	No	No	SQL-Like Predicates to grant Stream Topic access
stream-lib-directory	String	../data/streamlib	No	Yes	Hot deploy directory for external jar files used from a stream

<swiftlet name="sys$streams" collect-interval="1000" stream-grant-predicates="stream\_%" stream-lib-directory="../data/streamlib"/>

<domains> in <swiftlet name="sys$streams">

Stream Domain Names

Each <domain> entry is identified by its name attribute (the Domain).

<swiftlet name="sys$streams">
  <domains>
    <domain name="..."/>
  </domains>
</swiftlet>

<packages> in <domains>

Stream Packages

Each <package> entry is identified by its name attribute (the Package).

<swiftlet name="sys$streams">
  <domains>
    <domain name="...">
      <packages>
        <package name="..."/>
      </packages>
    </domain>
  </domains>
</swiftlet>

Changelog

13.2.0 (2025-11-03)

• Modified Stream
• StreamController: added timeout handling
• StreamController: added timeout handling, added validation, added null check, added error handling, fixed boundary condition; StreamProcessor.dispatch(): added null check, added timeout handling; StreamProcessor: added safeAccept()
• StreamController: added timeout handling

13.1.0 (2025-01-16)

• StreamProcessor: removed process()
• Modified StreamContext; StreamController: fixed boundary condition, added error handling; StreamController: added loadClass(); StreamController: removed findClass(); Modified GraalSetup
• Modified StreamsSwiftlet; Modified StreamProcessor
• Modified StreamController