JMS Input and Output Adapters

By default, the JMS adapters only try to connect to a JNDI or JMS server one time before giving up. However, you can configure the adapter to retry the connection a certain number of times at set intervals before giving up. To do this, set the server-num-retries and server-reconnect-interval attributes of the relevant <jms-server> section in the adapter's configuration file. The server-num-retries attribute controls the number of times the adapter will try to connect before giving up. The default value of this attribute is 0, which specifies no retries; the maximum value is MAX_INT. The server-reconnect-interval attribute controls the number of seconds to wait between connection attempts. The default value is 30 seconds.

Note that these values apply to both the JNDI and JMS servers configured in this <jms-server> section.

This functionality, used in conjunction with the Command Input and Status Output ports described here, can be used to implement logic to fully control the adapter's connection logic to the JMS servers it is configured to use.

When the JMS Reader receives a message, it translates this message into a Tuple and when the JMS Writer receives a Tuple, it translates it into a JMS message. There are five JMS message types: BytesMessage, MapMessage, ObjectMessage, StreamMessage and TextMessage. Of these types, MapMessage is the one into which tuples translate most readily. Like a tuple, a MapMessage is a set of name-value pairs, making the translation of these two message types relatively straightforward. The JMS Reader includes a class that converts JMS MapMessages to StreamBase tuples, DefaultFromJMSMapMessageConverter. Similarly the JMS Writer adapter provides DefaultToJMSMapMessageConverter to translate from a Tuple to a JMS MapMessages.

StreamBase tuples and JMS MapMessages are essentially collections of name-value pairs. The JMS Reader pairs the fields held by a JMS MapMessage with fields in a StreamBase schema that have the same names and compatible types. If the name of a field in the MapMessage doesn't exist in the StreamBase schema then that field is dropped. Similarly, if the name of a field in the Schema doesn't exist in the MapMessage then there is no input value for that field and it is nulled.

In the case of the JMS Writer, all fields in the incoming Tuple are added to the MapMessage.

An interface is used to represent a class that can convert a JMS Message to a StreamBase tuple. This interface, com.streambase.sb.adapter.common.jms.enqueue.FromJMSMessageConverter, includes a method, Object convertFromJMSMessage(Message message) throws StreamBaseException that provides the implementation of the translation. A different translation class can be implemented, and the JMS Reader can then be configured to use this implementation. In this way, a converter to translate one of the other JMS message types can be specified, or an alternate implementation of a MapMessage converter can be specified.

As expected, the JMS Writer also provides a similar interface, com.streambase.sb.adapter.common.jms.dequeue.ToJMSMessageConverter which includes the Message convertToJMSMessage(Object object) throws StreamBaseException method to do the opposite conversion.

A JMS message consists of two primary pieces, a header and a payload (body). In addition, properties can be set on a JMS message.

By default, the JMS Embedded Adapters map the fields of a MapMessage to the fields of a StreamBase schema. The JMS Reader looks for entries in the payload of a MapMessage that have the same names as the fields of the StreamBase schema. The JMS Writer simply creates entries in the outgoing Message to match all the fields of the incoming StreamBase schema.

The JMS Embedded Adapters also provide facilities for handling other mappings, specified in the configuration file. For example, a name map can be used to map a field on a MapMessage to a field on a StreamBase schema that has a different name. Similarly, the fields in the header of a JMS message can be mapped to the fields of a StreamBase schema. In the same way, JMS properties can also be mapped to the fields of a StreamBase schema.

The JMS Reader consumes messages from JMS destinations. The manner in which these messages are acknowledged is configurable within the JMS Reader. The setting for the acknowledge mode can impact the performance of the reader.

The JMS Reader supports acknowledge modes of AUTO_ACKNOWLEDGE and DUPS_OK_ACKNOWLEDGE. The reader also supports the use of TIBCO's proprietary acknowledge mode, NO_ACKNOWLEDGE. Of course, use of this acknowledge mode implies that TIBCO's EMS message bus is being used with the reader. The acknowledge mode for a given JMS server is specified in the jms-server section of the configuration file, using the attribute acknowledge-mode.

Note also that the JMS Reader makes no effort to prevent or in any way handle the delivery of duplicate JMS messages. If a JMS server delivers the same message twice, the JMS Reader will simply consume it twice.

The JMS Writer will create JMS messages and deliver them to a JMS message bus. JMS messages can be delivered with either of two delivery modes, DeliveryMode.PERSISTENT or DeliveryMode.NON_PERSISTENT. The delivery mode for a JMS destination is specified in the destination section of the configuration file, using the delivery-mode attribute.

The semantics of JMS delivery modes is beyond the scope of this discussion. In general, a delivery mode of PERSISTENT implies a higher quality of service for the JMS destination. Some performance penalty is also implied, however. In particular, if a delivery mode of PERSISTENT is specified, and there are durable subscribers for the message, then the JMS message will need to be persisted before the JMS server can acknowledge the message producer.

The JMS Writer dequeues tuples from StreamBase streams, translates them into JMS messages, and routes them to JMS destinations. The adapter relies on two fields in the dequeued tuple to determine which destination on which JMS server to send the translated message. The two fields must be called __JMSServerName and __JMSDestinationName. These fields must contain the name of a configured JMS server and destination, respectively. These fields are optional if only one destination is specified in the configuration; they are required if the configuration specifies more than one destination (that is, more than one server or more than one destination in a single server).

When one of the JMS Embedded Adapters executes, it needs to connect to an executing instance of a JMS message bus. Typically, this bus will be from one of the JMS providers. In order to connect to the JMS provider's JMS bus, JAR files supplied by the JMS provider must be made available to the JMS Adapters. This is true whether the adapter is executing from within StreamBase Studio or at a command-line invocation of StreamBase Server:

Name: Use this field to specify or change the component's name, which must be unique in the application. The name must contain only alphabetic characters, numbers, and underscores, and no hyphens or other special characters. The first character must be alphabetic or an underscore.

Adapter Name: A read-only field that shows the formal name of the adapter.

Class: A field that shows the fully qualified class name that implements the functionality of this adapter. Use this class name when loading the adapter in StreamSQL programs with the APPLY JAVA statement. You can right-click this field and select Copy from the context menu to place the full class name in the system clipboard.

Start with application: If this check box is selected, an instance of this adapter starts as part of the containing StreamBase Server. If cleared, the adapter is loaded with the server, but does not start until you send an sbadmin resume command, or start the component with StreamBase Manager. With this option cleared, the adapter does not start even if the application as a whole is suspended and later resumed. The default and recommended setting is selected.

Enable Error Output Port: Select this check box to add an Error Port to this component. In the EventFlow canvas, the Error Port shows as a red output port, always the last port for the component. See Using Error Ports and Error Streams to learn about Error Ports.

Description: Optionally enter text to briefly describe the component's purpose and function. In the EventFlow canvas, you can see the description by pressing Ctrl while the component's tooltip is displayed.

There is only one required property for either of the JMS Embedded Adapters: its configuration file. Configuration of the adapter is fairly involved, and so is not accomplished through its properties, but rather through its configuration file. Keep in mind that just the name of the configuration file, rather than its path, should be specified in the property. The named file will be searched for using the path specified in the configuration for StreamBase itself.

There are also two optional properties to control whether the adapter should accept JMS connection/disconnection commands from the application and whether status tuples should be issued when JMS servers or destinations become available or unavailable.

The JMS Reader is similar to an input stream that gets its input from a JMS Message bus. As with an input stream, a Schema needs to be specified for the JMS Reader. The Schema used by the JMS Reader is specified in the Edit Schema tab of the Properties View in StreamBase Studio.

Here are some typical settings for connecting to JMS servers from TIBCO EMS, ActiveMQ, IBM WebSphere, and Solace Systems:

TIBCO EMS (using JNDI)

<jms-servers>
  <jms-server name="EMSServer" 
    provider-context-factory="com.naming.InitialContextFactory" 
    provider-url="tibjmsnaming://localhost:7222" 
    connection-factory-name="GenericConnectionFactory" 
    acknowledge-mode="NO_ACKNOWLEDGE"/>
</jms-servers>

ActiveMQ version 5.x (using JNDI)

<jms-servers>
  <jms-server name="ActiveMQServer" 
    provider-context-factory="org.apache.activemq.jndi.ActiveMQInitialContextFactory"
    provider-url="tcp://localhost:61616"
    connection-factory-name="org.apache.activemq.ActiveMQConnectionFactory"
    acknowledge-mode="NO_ACKNOWLEDGE"/>
</jms-servers>

IBM WebSphere Application Server version 6.1's Default JMS Implementation (using JNDI)

<jms-servers>
  <jms-server name="server1" 
    provider-context-factory="com.ibm.websphere.naming.WsnInitialContextFactory"
    provider-url="iiop://localhost:2809"
    connection-factory-name="jms/MyConnFactory"
    jndi-initial-context-builder="com.streambase.sb.adapter.common.jms.common.WASJNDIInitialContextBuilder"
    create-destinations="false"
    acknowledge-mode="AUTO_ACKNOWLEDGE"/>
</jms-servers>

Solace default JMS Implementation (using JNDI)

<jms-servers>
  <jms-server name="solaceJms" 
    provider-context-factory="com.solacesystems.jndi.SolJNDIInitialContextFactory"
    provider-url="smf://MyProviderHost:MyProviderPort"
    connection-factory-name="MyConnFactory"
    acknowledge-mode="AUTO_ACKNOWLEDGE"
    jndi-security-principal="MySecurityPrincipal"
    jndi-security-credentials="MySecurityCredentials"
    jndi-initial-context-builder="com.streambase.sb.adapter.common.jms.common.SolaceJNDIInitialContextBuilder"
    username="MyUsername"
    create-destinations="true"
    password="MyPassword" >
    <jndi-extra-properties>
      <extra-property name='Solace_JMS_OptimizeDirect' value='false'/>
      <extra-property name='Solace_JMS_VPN' value='MyJmsVpn'/>
    </jndi-extra-properties>
    <destinations>
      <destination name="topics.test1" name-map="MSFTConversion"/>
    </destinations>
  </jms-server>
</jms-servers>

The command input port may be used by the application to request connections or disconnections to a configured JMS server, or to pause or resume processing for a JMS destination.

If the adapter is configured to expose a command input port, its schema is expected to have the following fields:

Permitted values for the command field are the following (case is unimportant):

The status output port can be used by the application to receive tuples describing status events for configured JMS servers (connection and disconnection) and optionally for destinations (available or unavailable.)

If the adapter is configured to expose a status output port, its schema has the following fields:

Possible values for the status field are:

First, you must tell the JMS adapter what kind of destination you wish to create by adding a default-destination entry to your JMS server's destinations section in the adapter's configuration file. The syntax to this section is identical to that of regular destination entries, except that the name attribute is ignored. You must also make sure that the JMS server's definition includes a create-destinations="true" attribute:

<jms-servers>
  <jms-server name="myJMSServer" 
    provider-context-factory="..."
    provider-url="..."
    connection-factory-name="MyConnFactory"
    acknowledge-mode="AUTO_ACKNOWLEDGE"
    jndi-security-principal="MySecurityPrincipal"
    jndi-security-credentials="MySecurityCredentials"
    jndi-initial-context-builder="..."
    create-destinations="true"
    username="MyUsername"
    password="MyPassword" >
    <destinations>
      <default-destination name-map="MyNameMap" use-topics="true"/>
      <destination name="topics.test1" name-map="MyNameMap"/>
    </destinations>
  </jms-server>
</jms-servers>

The way endpoints are created dynamically at runtime depends on whether you wish to publish or subscribe to the new endpoint — in other words, it depends on whether you are using the JMS Reader or JMS Writer adapter.