Container Connections

StreamBase supports stream-to-stream container connections and CSV file container connections.

Stream-to-Stream

In this configuration, you set up an EventFlow module's input stream in one container to receive tuples from an output stream of another module in another container. Both containers must be running in the same JVM engine.

Stream-to-stream connections can be of two types:

• Asynchronous. In the default form of stream-to-stream container connection, the flow of tuples is set to run asynchronously, with a queue automatically set up by StreamBase to manage the flow through the connection.

• Synchronous. You can optionally specify low latency, direct connections between container streams without the benefit of queue management. This option offers a possible latency improvement for certain circumstances. In general, asynchronous connections improve throughput, and synchronous connections improve latency. See Synchronous Container Connections.

Stream-to-stream connections can also be made from an output stream to an input stream within the same container, where the destination and source container name is the same. Intra-container connections must be Asynchronous, and the same access restrictions apply: that is, streams you wish to connect to or from in sub-modules must be marked as Always exposed in EventFlow modules.

In the following diagram, we add two reader modules (B and C) to a running EventFlow engine that was already hosting module A:

In the next diagram, we add two writer modules (B and C) to a running engine that was already hosting module A.

Notes

• In the diagrams above, the EventFlow modules are shown for simplicity as having the same components. This is for illustration only; modules in containers are not required to have the same set of components. However, the streams used in container connections between modules must have the same schema.

• When removing containers and modules from a running server process, try to remove them in the reverse order that they were added.

StreamBase provides several methods of specifying a connection between containers. Some methods are interactive (for testing and management of containers on running servers), while some methods specify that the connection is made immediately when the enclosing server starts (or when the enclosing container on that server starts). However, all methods are equally effective. Use the method that works best for your application and its stage of development or deployment, but only specify one method per connection for a given run of StreamBase Server. For example, do not specify a connection between containers A and B in both the Run Configuration dialog and in the Advanced tab of a stream.

Each container connection must be specified only once for each connected pair of containers. Do not specify a container connection in the output stream of the sending container and again in the input stream of the receiving container. You can specify the connection in either the sending or receiving container to the same effect. Do not specify the container connection in both places.

The methods of specifying container connections are:

Studio: Connections in Stream Property Advanced Tabs

In StreamBase Studio, in the Advanced tab of the Properties view for either an input stream or output stream, use the Container connection field to specify a StreamBase expression that resolves to the qualified name of a stream in another container. A qualified stream name follows this syntax:

container-name.stream-name

When using the Advanced tab for an output stream, you are already in the source stream, so you only need to specify the qualified name of the destination stream. Similarly, in the Advanced tab for an input stream, you are in the destination stream, and only need to specify the qualified name of the source stream.

You must express the container-qualified stream name as a StreamBase expression. To express a known container and stream name, enclose the qualified name in single or double quotes to render the name as an expression.

The container connection expression can optionally include variables, constants, and expression language functions that resolve to a qualified container and stream name. However, remember that the container connection is established at container start time, so your container connection expression cannot reference tuple field names or contents, since no tuples have yet passed through at container start time. For the same reason, the container connection expression cannot resolve any value of dynamic variables, including the default value.

Specify each container connection only once, in either the sending or the receiving stream. Do not specify the container connection in both places.

When you use the Advanced tab to specify a container connection, the connection is stored as part of the EventFlow XML definition of the module, and thus travels with the module. As long as the container names are the same, the same connection will be made whether the module is run in Studio or at the command prompt with the epadmin command.

The icons for input and output streams show an overlay decoration on their upper back corner to indicate that this stream initiates a container connection to another stream by means of the Advanced tab's Container connection field.

Streams on the other side of the connection in the receiving container are not decorated. Decorations are not applied for connections specified to start at runtime in a deployment file, in a Studio launch configuration, in StreamBase Manager, or from the command line with the epadmin command.

There are restrictions on the order in which you start connected containers, both at the command line and in Studio. See Container Start Order for details.

epadmin [sn|ad] add container --module=mymodule.sbapp --name=myContainer 
--connections="myContainer.anotherModule.inputStream=default.myModule.outputStream...

epadmin [sn|ad] add container --module=preprocess.sbapp --name=A
epadmin [sn|ad] add container --module=mainprocess.sbapp --name=B  
--connections="B.InputMain=A.OutputPre"
epadmin [sn|ad] add container --module=postprocess.sbapp --name=C  
--connections="C.InputPost=B.OutputMain"

dest-container.instream=source-container.outstream

To specify a container connection, you must specify both source and destination stream names, and you must qualify the stream names with their container names. The following syntax is explicitly required when using the epadmin command, and is expressed in other ways with the other container connection methods:

inputStream=outputStream

If the incoming or outgoing stream is in a module called by a top-level application in a container, then you must specify the module name as well:

dest-container.moduleA.inputStream=source-container.outputStream
Property Data Type Determines Generated Control
dest-container.inputStream=source-container.moduleB.outputStream
dest-container.moduleA.inputStream=source-container.moduleB.outputStream

If the source container is running on a remote StreamBase Server (or if the server is running on a different port on the same host), specify a full StreamBase URI enclosed in parentheses:

dest-container.inputStream=(sb://remotesbhost:9955/app3.outputStream)
dest-container.inputStream=(sb://localhost:10001/default.outputStream6)

You can use a StreamBase URI for the receiving container, as well. For example:

(sb://remotehost:8855/app2.inputStream)=source-container.outputStream
(sb://localhost:10001/default.inputStream4)=source-containter.outputStream

You can use the same syntax for remote hosts in both the destination and source attributes in the HOCON file:

    containerA = {
      moduleName = "primary.sbapp" 
      }
    containerB = {
      moduleName = "secondary.sbapp"
      }
    containerConnections = [
      { 
       destination ="B.InputStream1" 
       source="(sb://remotesbhost:9955/app3.outstream)" 
      }
      { 
        destination ="B.InputStream2" 
        source="A.OutputStream2" 
      }
    ]

The remote host syntax also works in the Advanced tab of input and output streams, but not when using the Run Configuration dialog in Studio or StreamBase Manager.

In three of the ways to define a container connection, you can specify a predicate expression filter to limit the tuples that cross to the destination container. The expression to enter must resolve to true or false, and should match against one or more fields in the schema of the connection's stream. You can specify a container connection predicate filter in:

There is no mechanism to specify a predicate filter for container connections specified in Studio's Launch Configuration dialog, in StreamBase Manager, or in the server configuration file.

For example, the HOCON file could have a containerConnections element like the following:

  containerConnections = [ 
    {
    destination ="holder2.input1"
    source="default.output3" --where="tradeID % 2 == 0"
    }
    ]

When using epadmin:

epadmin [sn|ad] modify container holder2.input=default.output3 
   --where "tradeID % 2 == 0"

When using the filter expression field on the Advanced tab in the Properties view of an Input Stream or Output Stream, enter only the expression: tradeID % 2 == 0

With a valid container connection specified, when the module containing the connection starts, the hosting server locates the specified container and stream, and makes the connection. As the application runs, tuples exiting one container's output stream are consumed by the specified input stream in the other container.

The schema of both sides of a container connection must match, as described in Schema Matching for Container Connections.

If the hosting server cannot make the container connection for any reason, it stops and writes an error message to the console. The likely reason for a failure to make a container connection is a failure to locate either the specified container or the specified input stream in the module running in that container. In this case, check the spelling of your container connection specification.

When a container is removed, the input or output to any dependent container disappears. The dependent container continues to function, but its input or output is removed.

Stream-to-stream container connections are not limited to one container on each side of the connection. That is, you can configure:

When input and output streams are shared, the streams are still owned by the original container. This means that any input or output to the shared stream ultimately needs to go through the original container. When tuples are passed between asynchronous container connections, they are passed using the same queuing mechanism used between parallel modules. This means that the tuple is queued for the other containers. When the tuple is queued, the original container must wait for the tuple to be pushed onto the queue, but will not be blocked by the processing of the tuple (which is done in another thread). Because of the queuing architecture, there is no guarantee of the order of tuples when they are shared between multiple containers.

In the following example, containers A and B share the stream A.InputStream1. When a tuple is enqueued into A.InputStream1 the tuple is first passed to all containers that share that stream (in this case, container B) and then it is processed to completion by itself.

epadmin [sn|ad] add container --name=A --module=app.sbapp
epadmin [sn|ad] add container --name=B --module=app.sbapp 
--connections="B.InputStream1=A.InputStream1"

Nothing prevents you from enqueuing directly to an input stream in container B while that stream also receives input from an upstream connection from container A. Because upstream tuples are queued, any input directly into container B is randomly interleaved with input from container A.

The use of container connections imposes some restrictions on the startup order of containers, especially when making connections with epadmin commands. The general rule is: the container that holds the receiving end of a container connection must be running when attempting to connect the containers.

When you run a deployment file, it starts all containers first, then tries to make any specified container connections. Because all containers are running, all container connections succeed. Thus, container start order issues do not arise when using deployment files.

However, it is possible to specify a container connection as part of an epadmin add container command. In this case, the receiving end of the container connection must be started first.

The schemas of the outgoing and incoming streams in a container connection must exactly match, or be coercible to match. That is, the connected streams must share exactly the same field names, field data types, and field positions between the two streams. If the two streams share the same field types, sizes, and names, but are in different order, StreamBase cannot map the streams together.

Use epadmin modify container to dynamically add or remove a container connection from an existing container while StreamBase Server is running:

epadmin [sn|ad] modify container --name=container-name [addConnection | removeConnection] 
    connection-expression1 connection-expression2 ...

where each container-expression has the syntax destination=source as in this example: containerA.inputStream1=containerB.module1.outputStream1

By default, connections between containers are set up to run asynchronously, with a queue to manage the flow of tuples through the connection.

StreamBase also supports synchronous container connections, which are low latency, direct connections between containers, and do not have queue management.

Specify a synchronous container connection with the epadmin command by using := (colon-equals) instead of = (equals) in your container connection assignment statement. For example:

epadmin [sn|ad] add container --module=app1.sbapp --name=A
epadmin [sn|ad] add container --module=app2.sbapp --name=B
epadmin [sn|ad] modify container addConnection B.incoming:=A.outgoing

You can also specify a synchronous container connection in a deployment file by adding the synchronicity attribute to the containerConnections element that specifies your connection:

...
      containerConnections [
    {
      destination ="B.InputStream1" 
      source="A.OutputStream1"
      synchronicity="SYNCHRONOUS" 
    }
   ]
...

To determine whether a synchronous container connection will improve your application, you must test your application with and without the synchronous setting, using a test input stream that emulates the actual conditions your application will face in production. Use the Rules of StreamBase Execution Order in the Execution Order and Concurrency page as general guidelines to help decide whether to use synchronous container connections.

StreamBase detects and prevents any attempt to make a synchronous container connection to:

When using a StreamBase URI as part of a stream-to-stream container connection string to specify that one side of the container connection is on a remote StreamBase Server, you can optionally specify URI parameters as part of the remote URI.

The URI parameters you can specify for connections to remote servers depend on which side of the connection is remote. There are two cases:

These parameters have the same meanings for container connections as the similarly named property of the StreamBase to StreamBase adapters described in StreamBase to StreamBase Output Adapter and StreamBase to StreamBase Input Adapter.

Append one or more container connection parameters to a StreamBase URI, each preceded by a semicolon. For example, to specify non-default buffer sizes for a destination connection, use a container connection string like the following:

("sb://remotehost:9900/boxC.instr2;enqueue-buffer-size=300;max-enqueue-buffer-size=2000")=boxD.outstr6

To filter data as it is enqueued, you can use a --where predicate, as illustrated below for a field named Price:

("sb://remotehost:9900/boxC.instr2;enqueue-buffer-size=300;max-enqueue-buffer-size=2000")=boxD.outstr6 --where Price > 100

See StreamBase to StreamBase Output Adapter for a discussion of these property settings.