FILENAME Statement: ACTIVEMQ Access Method

Assigns a fileref that enables you to access an ActiveMQ broker by using the ACTIVEMQ access method.

Valid in:	Anywhere
Category:	Data Access
Requirement:	The ActiveMQ broker that you access must be release 5.7.0 or later.

Table of Contents

Syntax

FILENAME fileref ACTIVEMQ <"destination-name" >< access-method-options>

Summary of Optional Arguments

"destination-name"

specifies the destination name for the ActiveMQ message broker.

AUTHDOMAIN= "domain"

specifies a metadata authentication domain that is used to retrieve user credentials.

CLIENTID= "client-name" | GENERATE | NONE

specifies an ID string that is used to identify the client.

CONTENTTYPE= "type"

specifies the format of messages or records that are being written.

CORRELATIONID=correlId-variable

specifies the correlation ID for outgoing messages, or stores the correlation ID for incoming messages to a variable.

DELIVERYMODE=PERSISTENT | NONPERSISTENT

specifies the message delivery mode.

DESTINATION= "destination-name"

specifies the name of the message destination on the ActiveMQ broker.

DESTTYPE= "type"

specifies whether the destination is a queue or a topic.

ENCODING= "encoding-type"

transcodes messages or records by using the specified encoding type.

EOFCHECK= "check-string"

specifies a string that indicates the end of the file.

MATCHOPTIONS= CORRELATIONID | MESSAGEID | BOTHIDS

adds a value to an existing message selector string, or creates a message selector string.

MESSAGEID= variable

assigns message ID values to the DATA step variable that you specify.

MSGBODYPREFIX= "prefix-string"

specifies a string that is appended to the start of each message that you write.

MSGSELECTOR= "filter-string"

specifies a filter string that restricts the messages that are received by the access method.

PASSWORD= "password" | _PROMPT_

specifies the password that is used to connect to the ActiveMQ broker.

PRIORITY=n

specifies the message priority for outgoing messages, where a larger priority number indicates a higher priority.

PROXYPASSWORD= "password"

specifies the password that is used to connect to the proxy server.

PROXYURL= "URL-string"

specifies the URL for a proxy server that is used to connect to the ActiveMQ broker.

PROXYUSER= "user-ID"

specifies the user ID that is used to connect to the proxy server.

TERMSTR= LF | CRLF | CR | NONE | NULL

specifies the character sequence that indicates the end of a line in messages that are being read.

TIMEOUT=n

specifies a time limit in milliseconds to wait for a message to arrive.

TIMETOLIVE=n

specifies the default length of time in milliseconds that a message is retained by the message system.

URL= "URL-string"

specifies the URL string that is used to connect to the ActiveMQ broker.

USER= "user-ID" | _PROMPT_

specifies the user ID that is used to connect to the ActiveMQ broker.

Required Arguments

fileref

is a valid file reference.

ACTIVEMQ

specifies the access method that enables you to access an ActiveMQ broker.

Optional Argument

"destination-name"

specifies the destination name for the ActiveMQ message broker.

Interaction	You can specify the destination name by using either the destination–name argument, the DESTINATION= option, the URL= option, or aggregate syntax.

Access Method Options

AUTHDOMAIN= "domain"

specifies a metadata authentication domain that is used to retrieve user credentials. The credentials that are associated with the are used to connect to the ActiveMQ broker.

Requirement	To use the AUTHDOMAIN= option, your SAS session must be able to connect to the SAS Metadata Server. For information about specifying a metadata server connection, see SAS Language Interfaces to Metadata.
Tip	The value does not require quotation marks if it is a valid SAS name.

CLIENTID= "client-name" | GENERATE | NONE

specifies an ID string that is used to identify the client. The ID is used by the ActiveMQ broker to clean up resources from a client session. You can specify any string, or you can specify GENERATE to automatically generate a universally unique ID for the client. You can also specify NONE to specify that no client ID is used.

Default	GENERATE

CONTENTTYPE= "type"

specifies the format of messages or records that are being written. You can specify either the text/xml or the application/x-www-form-urlencoded content types.

Note: The CONTENTTYPE= option is valid in the FILE and FILENAME statements only.

Alias	CTYPE=
Default	text/xml

CORRELATIONID=correlId-variable

specifies the correlation ID for outgoing messages, or stores the correlation ID for incoming messages to a variable. For the FILE statement, specifies a DATA step variable whose value is used as the correlation ID of the message. The correlation ID is added to outgoing messages in the JMSCorrelationID and CorrelationID parameters.

For the INFILE statement:

the correlation ID of the incoming message is stored in the variable that you specify.
the value of the variable that you specify is used as the correlation ID for the message selector when you specify MATCHOPTS=CORRELID or MATCHOPTS=BOTHIDS.

Note: The CORRELATIONID= option is valid in the FILE and INFILE statements only.

Alias	CORRELID=

DELIVERYMODE=PERSISTENT | NONPERSISTENT

specifies the message delivery mode. Specify one of the following:

PERSISTENT

specifies that messages are stored in a stable storage location. Messages can be recovered if the ActiveMQ broker fails to deliver the message.

NONPERSISTENT

specifies that messages are not stored in a stable storage location. The NONPERSISTENT mode offers better performance than the PERSISTENT mode, but messages might be lost if the ActiveMQ broker fails.

Note: The DELIVERYMODE= option is valid in the FILE and FILENAME statements only.

Default	PERSISTENT

DESTINATION= "destination-name"

specifies the name of the message destination on the ActiveMQ broker. The destination can also be specified by using the destination-name argument, aggregate syntax, or as a parameter on the = option.

Alias	DEST=
Interaction	You can specify a destination by using either the destination-name argument, the DESTINATION= option, or aggregate syntax.
Tip	The value does not require quotation marks if it is a valid SAS name.

DESTTYPE= "type"

specifies whether the destination is a queue or a topic. Specify "Q" or "QUEUE" for a , or specify "TOPIC" for a topic.

Default	"QUEUE"
Interaction	If the destination is specified on the URL= option, then the DESTTYPE= option is ignored.

ENCODING= "encoding-type"

transcodes messages or records by using the specified encoding type. When writing, the encoding that you specify is used to transcode the messages that are written to the message broker. When reading, the encoding that you specify is used to transcode the messages to your SAS session encoding.

For a list of encoding types, see Encoding Values in SAS Language Elements in SAS National Language Support (NLS): Reference Guide.

Default	When writing, the default encoding is the SAS session encoding. When reading, the default encoding is UTF8.

EOFCHECK= "check-string"

specifies a string that indicates the end of the file. The ACTIVEMQ access method returns end of file when it encounters a message that matches the EOFCHECK= value or the EOFCHECK= value prefixed with the string "OBJMQ_STOP:"

For example, if you specify EOFCHECK="STOPMSG", then a message value of either STOPMSG or OBJMQ_STOP:STOPMSG causes the ACTIVEMQ access method to return end of file.

LRECL=n | nk

specifies the record length in bytes. Add k to the number to specify a multiple of 1024.

Default	32767

MATCHOPTIONS= CORRELATIONID | MESSAGEID | BOTHIDS

adds a value to an existing message selector string, or creates a message selector string. Specify one of the following values:

CORRELATIONID

adds the correlation ID to the selector string.

Alias	CORRELID

MESSAGEID

adds the message ID to the selector string.

Alias	MSGID

BOTHIDS

adds both the correlation ID and the message ID to the selector string.

Alias	MATCHOPTS=

MESSAGEID= variable

assigns message ID values to the DATA step variable that you specify. You can use the MESSAGEID= option to obtain the message ID of an incoming or outgoing message. For outgoing messages, each Write operation writes a new message ID value to the variable.

Note: The MESSAGEID= option is valid in the FILE and INFILE statements only.

Alias	MSGID=

MSGBODYPREFIX= "prefix-string"

specifies a string that is appended to the start of each message that you write. Adding a prefix is useful when you are writing multiple messages at once, such as if you are using the FCOPY function to write messages.

Note: The MSGBODYPREFIX= option is valid in the FILE and FILENAME statements only.

MSGSELECTOR= "filter-string"

specifies a filter string that restricts the messages that are received by the access method. The string uses SQL92 conditional expression syntax. A message selector matches a message if the selector expression evaluates to true for that message.

For example, if you specify msgselector="JMSPriority = 9", then the access method only receives messages where the priority is 9.

Note: The MSGSELECTOR= option is valid in the FILENAME and INFILE statements only.

PASSWORD= "password" | _PROMPT_

specifies the password that is used to connect to the ActiveMQ broker. Specify a password or specify _PROMPT_ to be prompted to enter your password interactively.

Note: You can submit a password as clear text or as an encoded string from the PWENCODE procedure. For more information, see Encryption in SAS.

Aliases	PASS=
	PW=
	PWD=
Tips	You can specify the AUTHDOMAIN= option instead of the USER= and PASSWORD= options.
Tips	The value does not require quotation marks if it is a valid SAS name.

PRIORITY=n

specifies the message priority for outgoing messages, where a larger priority number indicates a higher priority. The value must be between 0 and 9.

PRIORITY= has no effect for incoming messages. You can filter incoming messages based on priority by using the value of JMSPriority in a message selector. For example, the following message selector specifies a priority of 5.

msgselector="JMSPriority=5"

The default value is determined by the ActiveMQ broker. If the ActiveMQ broker uses a JDBC message store, then the default value is 4. For a KahaDB message store, the default value is 5.

PROXYPASSWORD= "password"

specifies the password that is used to connect to the proxy server. Specify a password or specify _PROMPT_ to be prompted to enter your password interactively.

Note: You can submit a password as clear text or as an encoded string from the PWENCODE procedure. For more information, see Encryption in SAS.

Aliases	PROXYPASS=
Aliases	PROXYPASSWD=
Interaction	This option has no effect if you do not specify the PROXYURL= option.
Tip	The value does not require quotation marks if it is a valid SAS name.

PROXYURL= "URL-string"

specifies the URL for a proxy server that is used to connect to the ActiveMQ broker.

PROXYUSER= "user-ID"

specifies the user ID that is used to connect to the proxy server. Specify a user ID or specify _PROMPT_ to be prompted to enter your user ID interactively.

Alias	PROXYUSERID=
Interaction	This option has no effect if you do not specify the PROXYURL= option.
Tip	The value does not require quotation marks if it is a valid SAS name.

TERMSTR= LF | CRLF | CR | NONE | NULL

specifies the character sequence that indicates the end of a line in messages that are being read. Specify one of the following values:

LF

specifies a line feed character.

CRLF

specifies a carriage return character and a line feed character.

CR

specifies a carriage return character.

NONE

specifies that the entire message is read, including any line-ending characters.

Note: The TERMSTR= option is valid in the FILENAME and INFILE statements only.

Default	LF

TIMEOUT=n

specifies a time limit in milliseconds to wait for a message to arrive. If the time expires before a message arrives, then the access method returns end of file. If you do not specify the TIMEOUT= option, then the access method returns end of file after one millisecond if there are no messages. A value of 0 indicates that the access method waits an unlimited amount of time for a message to arrive.

Default	1

TIMETOLIVE=n

specifies the default length of time in milliseconds that a message is retained by the message system.

Note: The TIMETOLIVE= option is valid in the FILE and FILENAME statements only.

Default	0 (no limit)

URL= "URL-string"

specifies the URL string that is used to connect to the ActiveMQ broker. The format of the URL string is "scheme://hostname:port/webcontext/serverlet/destination?parameter1&parameter2"

The /destination part of the URL string is optional if you specify the destination by using another option or parameter (for example, the DESTINATION= option.)

If you specify the destination as part of the URL string, then you must also specify the “&TYPE=” parameter as part of your URL string. Specify &type=queue to specify that the destination is a queue, or specify &type=topic to specify that the destination is a topic.

Note: The web for the ActiveMQ broker URL is specified in the jetty.xml file in the /conf subdirectory of your ActiveMQ installation. The name of your context path is specified by the value of the org.eclipse.jetty.webapp.WebAppContext property. For ActiveMQ 5.7.0 and earlier, the default context is demo. For ActiveMQ 5.8.0 and later, the default context is api.

Default	http://localhost:8161/demo/message
Interaction	If you specify the destination by using the DESTINATION= option, the destination-name argument, or aggregate file syntax, then the destination value from those parameters overrides the destination that is specified on the URL=.

USER= "user-ID" | _PROMPT_

specifies the user ID that is used to connect to the ActiveMQ broker. Specify a user ID or specify _PROMPT_ to be prompted to enter your user ID interactively.

Alias	USERID=
Tips	You can specify the AUTHDOMAIN= option instead of the USER= and PASSWORD= options.
Tips	The value does not require quotation marks if it is a valid SAS name.

Details

Overview

The ActiveMQ file access method enables SAS programs to read and write records to and from an ActiveMQ broker directly through the protocol. No message-oriented middleware is required on the machine where SAS is running.

Filerefs assigned with the ActiveMQ file access method can be used in the same ways as other external file access methods can be used, including with the following statements: FILENAME, FILE, INFILE, and %INCLUDE. If you specify a fileref on both the FILENAME statement and the FILE or INFILE statement, then the option values from the FILE or INFILE statement will override the values from the FILENAME statement.

Communication with the ActiveMQ Broker

The ACTIVEMQ access method uses the protocol to interact with an ActiveMQ broker by using the RESTful web API. The ActiveMQ MessageServlet class implements the integration between HTTP and the ActiveMQ dispatcher. The SAS ActiveMQ access method uses HTTP POST for sending, and HTTP GET for receiving messages.

For details about the MessageServlet class, see https://activemq.apache.org/maven/apidocs/org/apache/activemq/web/MessageServlet.html at the Apache website.

The REST API can connect to secured brokers. The API uses basic header format to get user ID and password information.

Web Application Contexts

The web for the ActiveMQ broker is specified in the jetty.xml file in the /conf subdirectory of your ActiveMQ installation. The name of your context path is specified by the value of the org.eclipse.jetty.webapp.WebAppContext property.

For ActiveMQ 5.7.0 and earlier, the default context is demo. For ActiveMQ 5.8.0 and later, the default context is api.

Examples

Example 1: Installing and Testing ActiveMQ from a SAS Program

This example downloads and installs ActiveMQ and then performs a simple send and receive test using the local ActiveMQ installation.

Alternative Download Method

You can also download ActiveMQ manually from activemq.apache.org.

%let tempZipFile=c:\\activemq.zip;
%let installDir=c:\\;
filename fromweb URL
'http://apache.mirrors.tds.net/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip'
    lrecl=8192 recfm=s;

data _null_;
    infile fromweb;
    file "&tempZipFile" recfm=n lrecl=8192;
    input;
    put _infile_;
run;
proc groovy;
    add sasjar="ANT";
    submit "&tempZipFile" "&installDir";
        def ant = new AntBuilder();
        ant.unzip( src:args[0], dest:args[1], overwrite:"false" )
        ant.delete( file:args[0] )
    endsubmit;
quit;
options noxwait;
x "start &installDir.apache-activemq-5.8.0\bin\activemq";

data _null_; call sleep( 8, 1 ); run;   * Give broker time to initialize.;
data _null_;
    file 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin;
    format a datetime21.1;
    a = datetime(); put a ' Testing: 1 2';
    a = datetime(); put a +1 'This is a test. 3 4';
    a = datetime();
    put a @22 'This is only a test. 5 6';
run;
data _null_;
    infile 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin length=len timeout=1;
    input a datetime21.2 @;
    len = len - 25;
    input b $varying60. len c d;
    put a= datetime19. b= c= d=;
run;

%let tempZipFile=c:\\activemq.zip;
%let installDir=c:\\;
filename fromweb URL
'http://apache.mirrors.tds.net/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip'
    lrecl=8192 recfm=s;

data _null_;
    infile fromweb;
    file "&tempZipFile" recfm=n lrecl=8192;
    input;
    put _infile_;
run;
proc groovy;
    add sasjar="ANT";
    submit "&tempZipFile" "&installDir";
        def ant = new AntBuilder();
        ant.unzip( src:args[0], dest:args[1], overwrite:"false" )
        ant.delete( file:args[0] )
    endsubmit;
quit;
options noxwait;
x "start &installDir.apache-activemq-5.8.0\bin\activemq";

data _null_; call sleep( 8, 1 ); run;   * Give broker time to initialize.;
data _null_;
    file 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin;
    format a datetime21.1;
    a = datetime(); put a ' Testing: 1 2';
    a = datetime(); put a +1 'This is a test. 3 4';
    a = datetime();
    put a @22 'This is only a test. 5 6';
run;
data _null_;
    infile 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin length=len timeout=1;
    input a datetime21.2 @;
    len = len - 25;
    input b $varying60. len c d;
    put a= datetime19. b= c= d=;
run;

Program Description

Specify installation parameters.The tempZipFile= macro variable specifies the filename for the downloaded installation file. The installDir= macro variable specifies the directory where ActiveMQ will be installed (ActiveMQ will be installed in a subdirectory named apache-activemq-release-number.)

%let tempZipFile=c:\\activemq.zip;
%let installDir=c:\\;

Download the ActiveMQ binary distribution from the web.If necessary, specify the proxy for your internet connection by using the PROXY= option in the FILENAME statement. If you encounter a connection error, obtain a current recommended URL from ActiveMQ download site:https://activemq.apache.org/components/classic/download.

filename fromweb URL
'http://apache.mirrors.tds.net/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip'
    lrecl=8192 recfm=s;

data _null_;
    infile fromweb;
    file "&tempZipFile" recfm=n lrecl=8192;
    input;
    put _infile_;
run;

Extract the ActiveMQ ZIP file into the destination directory.

proc groovy;
    add sasjar="ANT";
    submit "&tempZipFile" "&installDir";
        def ant = new AntBuilder();
        ant.unzip( src:args[0], dest:args[1], overwrite:"false" )
        ant.delete( file:args[0] )
    endsubmit;
quit;

Start an instance of the ActiveMQ broker.

options noxwait;
x "start &installDir.apache-activemq-5.8.0\bin\activemq";

data _null_; call sleep( 8, 1 ); run;   * Give broker time to initialize.;

Put some test messages on the queue.

data _null_;
    file 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin;
    format a datetime21.1;
    a = datetime(); put a ' Testing: 1 2';
    a = datetime(); put a +1 'This is a test. 3 4';
    a = datetime();
    put a @22 'This is only a test. 5 6';
run;

Get the messages back off of the queue and print them in the log.

data _null_;
    infile 'myqueue' activemq url='http://localhost:8161/api/message'
        user=admin pass=admin length=len timeout=1;
    input a datetime21.2 @;
    len = len - 25;
    input b $varying60. len c d;
    put a= datetime19. b= c= d=;
run;

Log Output

Each test message appears in the :

a=21FEB2013:13:48:48 b=Testing: c=1 d=2
a=21FEB2013:13:48:48 b=This is a test. c=3 d=4
a=21FEB2013:13:48:48 b=This is only a test. c=5 d=6

Example 2: Using a Message Selector String

This example uses a message selector string to categorize data and to prioritize the order in which the messages for those data values are read.

%let url='http://localhost:8161/api/message';
%let moreOpts=user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;
data _null_;
    length name $10 rating $5;
    file requestq;
    do k=0 to 9;
        name='hotel' || put(k,1.); r=floor(mod(10,k+1)); rating=repeat('A',r);
        put name rating; end;
run;
data _null_;
    length msg $50 name $6 rating $5;
    infile requestq;
    file response cType='application/x-www-form-urlencoded';
    input name $ rating $; p=length(rating);
    msg='&JMSPriority=' || put(p,1.) || '&body=';
    msg=trim(msg) || trim(name) || ' ' || trim(rating);
    put msg;
run;
data _null_;
    infile response msgSelector="JMSPriority > 3";
    input name $ rating $;
    put name rating=;
run;
data _null_;
    infile response msgSelector="JMSPriority > 2";
    input name $ rating $;
    put name rating=;
run;
data _null_; 
    infile response;
    input;
    put _infile_;
run;

%let url='http://localhost:8161/api/message';
%let moreOpts=user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;
data _null_;
    length name $10 rating $5;
    file requestq;
    do k=0 to 9;
        name='hotel' || put(k,1.); r=floor(mod(10,k+1)); rating=repeat('A',r);
        put name rating; end;
run;
data _null_;
    length msg $50 name $6 rating $5;
    infile requestq;
    file response cType='application/x-www-form-urlencoded';
    input name $ rating $; p=length(rating);
    msg='&JMSPriority=' || put(p,1.) || '&body=';
    msg=trim(msg) || trim(name) || ' ' || trim(rating);
    put msg;
run;
data _null_;
    infile response msgSelector="JMSPriority > 3";
    input name $ rating $;
    put name rating=;
run;
data _null_;
    infile response msgSelector="JMSPriority > 2";
    input name $ rating $;
    put name rating=;
run;
data _null_; 
    infile response;
    input;
    put _infile_;
run;

Program Description

Initialize the filerefs for the sample.If necessary, change the URL=, USER=, and PASSWORD= options to match your ActiveMQ deployment.

%let url='http://localhost:8161/api/message';
%let moreOpts=user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;

Put some sample hotel rating messages on a request queue.

data _null_;
    length name $10 rating $5;
    file requestq;
    do k=0 to 9;
        name='hotel' || put(k,1.); r=floor(mod(10,k+1)); rating=repeat('A',r);
        put name rating; end;
run;

Read messages from the request queue and then put them onto the response queue with a priority property.The CTYPE= option specifies the content type application/x-www-form-urlencoded in order to write the JMSPriority message header parameter as part of the message. This enables the ActiveMQ access method to write a different priority value for each message according to the number of As in the rating.

data _null_;
    length msg $50 name $6 rating $5;
    infile requestq;
    file response cType='application/x-www-form-urlencoded';
    input name $ rating $; p=length(rating);
    msg='&JMSPriority=' || put(p,1.) || '&body=';
    msg=trim(msg) || trim(name) || ' ' || trim(rating);
    put msg;
run;

Read the messages with a priority greater than 3 from the response queue.The MSGSELECTOR= option specifies a message selector string.

data _null_;
    infile response msgSelector="JMSPriority > 3";
    input name $ rating $;
    put name rating=;
run;

Read the messages with a priority greater than 2 from the response queue.The MSGSELECTOR= option specifies a message selector string.

data _null_;
    infile response msgSelector="JMSPriority > 2";
    input name $ rating $;
    put name rating=;
run;

Read the remaining messages from the queue.

data _null_; 
    infile response;
    input;
    put _infile_;
run;

Log Output

For each message on the queue, the values are written to the log:

hotel5 rating=AAAAA
hotel6 rating=AAAA

hotel3 rating=AAA
hotel7 rating=AAA

hotel0 A
hotel2 AA
hotel8 AA
hotel1 A
hotel4 A
hotel9 A

Example 3: Using Request and Response Queues

This example uses separate request and response queues to send messages and then retrieve them by using their message ID.

%let moreOpts=url='http://localhost:8161/api/message'
              user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;
data requests;
    length name $40 mId $45;
    file requestq messageId=mId deliveryMode=nonPersistent;
    loanAmount=1000; creditScore=675; name='Smith, C.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=2000; creditScore=500; name='Brown, A.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=3000; creditScore=750; name='White, D.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=50000; creditScore=700; name='Jones, B.';
    put loanAmount @10 creditScore @15 name;
    output;
run;
data _null_; set requests; put name= @30 mId=; run;
data _null_;
    length name $40 mId $45;
    infile requestq messageId=mId length=len;
    file response correlId=mId deliveryMode=nonPersistent;
    input loanAmount creditScore @;
    len = len - 14;
    input @15 name $varying40. len;
    if ( creditScore > 600 and loanAmount < 10000 ) then
        decision = 'approved';
    else
        decision = 'declined';
    put decision name;
run;
proc sort data=requests; by name; run;
%macro getAnswers;
    %local i nobs;
    data _null_;
        if 0 then set requests nobs=nobs;
        call symput( 'nobs', put(nobs,1.) );
        stop;
    run;
    %do i=1 %to &nobs;
        data _null_;
            length msgName $40 cId $51;
            obsnum=&i;
            set requests point=obsnum;
            cId=mId;
            infile response 
                matchopts=correlid correlid=cId length=len;
            input @1 answer $8. @;
            len = len - 9;
            input +1 msgName $varying40. len;
            put '------------------';
            put 'Request message ID was ' mId;
            put name 'with a credit score of ' creditScore 'was ' answer
            'for a loan of' loanAmount dollar8.;
            put '------------------';
            stop;
        run;
    %end;
%mend;
%getAnswers;

%let moreOpts=url='http://localhost:8161/api/message'
              user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;
data requests;
    length name $40 mId $45;
    file requestq messageId=mId deliveryMode=nonPersistent;
    loanAmount=1000; creditScore=675; name='Smith, C.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=2000; creditScore=500; name='Brown, A.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=3000; creditScore=750; name='White, D.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=50000; creditScore=700; name='Jones, B.';
    put loanAmount @10 creditScore @15 name;
    output;
run;
data _null_; set requests; put name= @30 mId=; run;
data _null_;
    length name $40 mId $45;
    infile requestq messageId=mId length=len;
    file response correlId=mId deliveryMode=nonPersistent;
    input loanAmount creditScore @;
    len = len - 14;
    input @15 name $varying40. len;
    if ( creditScore > 600 and loanAmount < 10000 ) then
        decision = 'approved';
    else
        decision = 'declined';
    put decision name;
run;
proc sort data=requests; by name; run;
%macro getAnswers;
    %local i nobs;
    data _null_;
        if 0 then set requests nobs=nobs;
        call symput( 'nobs', put(nobs,1.) );
        stop;
    run;
    %do i=1 %to &nobs;
        data _null_;
            length msgName $40 cId $51;
            obsnum=&i;
            set requests point=obsnum;
            cId=mId;
            infile response 
                matchopts=correlid correlid=cId length=len;
            input @1 answer $8. @;
            len = len - 9;
            input +1 msgName $varying40. len;
            put '------------------';
            put 'Request message ID was ' mId;
            put name 'with a credit score of ' creditScore 'was ' answer
            'for a loan of' loanAmount dollar8.;
            put '------------------';
            stop;
        run;
    %end;
%mend;
%getAnswers;

Program Description

Initialize the filerefs for the sample.If necessary, change the URL=, USER=, and PASSWORD= options to match your ActiveMQ deployment.

%let moreOpts=url='http://localhost:8161/api/message'
              user=admin password=admin;
filename requestq activemq url=&url &moreOpts destination=requestq;
filename response activemq url=&url &moreOpts destination=response;

Put four loan requests on a request queue and save the message IDs.The MESSAGEID= option specifies a variable to store the message ID values.

data requests;
    length name $40 mId $45;
    file requestq messageId=mId deliveryMode=nonPersistent;
    loanAmount=1000; creditScore=675; name='Smith, C.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=2000; creditScore=500; name='Brown, A.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=3000; creditScore=750; name='White, D.';
    put loanAmount @10 creditScore @15 name;
    output;
    loanAmount=50000; creditScore=700; name='Jones, B.';
    put loanAmount @10 creditScore @15 name;
    output;
run;

Write the request message IDs to the log.

data _null_; set requests; put name= @30 mId=; run;

Read the messages from the request queue and then write answers to the response queue by using the message ID as the correlation ID.The MESSAGEID= option in the INFILE statement stores the message IDs from the request queue in the MID variable. The CORRELID= option in the FILE statement specifies that values from the MID variable are used as the correlation IDs for the messages on the response queue.

data _null_;
    length name $40 mId $45;
    infile requestq messageId=mId length=len;
    file response correlId=mId deliveryMode=nonPersistent;
    input loanAmount creditScore @;
    len = len - 14;
    input @15 name $varying40. len;
    if ( creditScore > 600 and loanAmount < 10000 ) then
        decision = 'approved';
    else
        decision = 'declined';
    put decision name;
run;

Sort the requests to demonstrate that the messages are not simply read in order.

proc sort data=requests; by name; run;

Read the answers from the response queue and write them to the log.In the INFILE statement, the MATCHOPTS= option specifies that the correlation ID is used to select messages. The CORRELID= option specifies that the CID variable is used for the correlation ID.

%macro getAnswers;
    %local i nobs;
    data _null_;
        if 0 then set requests nobs=nobs;
        call symput( 'nobs', put(nobs,1.) );
        stop;
    run;
    %do i=1 %to &nobs;
        data _null_;
            length msgName $40 cId $51;
            obsnum=&i;
            set requests point=obsnum;
            cId=mId;
            infile response 
                matchopts=correlid correlid=cId length=len;
            input @1 answer $8. @;
            len = len - 9;
            input +1 msgName $varying40. len;
            put '------------------';
            put 'Request message ID was ' mId;
            put name 'with a credit score of ' creditScore 'was ' answer
            'for a loan of' loanAmount dollar8.;
            put '------------------';
            stop;
        run;
    %end;
%mend;
%getAnswers;

Log Output

For each observation in the data set, a message appears in the log:

------------------
Request message ID was ID:d73274-57938-1360081257920-3:1:1:1:2
Brown, A.  with credit score of 500 was declined  for loan of  $2,000
------------------

------------------
Request message ID was ID:d73274-61889-1360699056153-3:1:1:1:4
Jones, B.  with credit score of 700 was declined  for loan of $50,000
------------------

------------------
Request message ID was ID:d73274-61889-1360699056153-3:1:1:1:1
Smith, C.  with credit score of 675 was approved  for loan of  $1,000
------------------

------------------
Request message ID was ID:d73274-61889-1360699056153-3:1:1:1:3
White, D.  with credit score of 750 was approved  for loan of  $3,000
------------------

Example 4: Using the FCOPY Function

This example demonstrates using the FCOPY function to copy messages from one fileref to another. It also demonstrates the use of message selector strings and the MSGBODYPREFIX= option.

%let url='http://localhost:8161/api/message/testq?type=queue';
%let moreOpts=user=admin password=admin;
filename myq activemq url=&url &moreOpts destination=testq;

data _null_;
    file myq correlId=corId;
    j=0;
    do i=1 to 3;
        d=datetime(); j=j+1; corId='Unpaid';
        put j @5 'Went unpaid >30 days.' +1 d datetime21.2 +1 'Xyz';
        d=datetime(); j=j+1; corId='Paid';
        put j @5 'Paid on time.' +1 d datetime21.2 +1 'Xyz';
    end;
run;
filename myq activemq url=&url &moreOpts
    msgSelector="JMSCorrelationID='Unpaid'";
filename sameq activemq url=&url &moreOpts
    ctype='application/x-www-form-urlencoded'
    msgBodyPrefix='&JMSCorrelationID=Paid&CorrelationID=Paid&body=';
data _null_;
    rc = fcopy( 'myq', 'sameq' );
    if rc ^= 0 then do;
        msg = sysmsg(); put rc= msg=; 
    end;
run;
data _null_;
    length msg $50;
    file log;
    infile 'testq' activemq url=&url &moreOpts
        length=len correlid=corid;
    corid='';
    input j @@;
    len = len - 30;
    input @5 msg $varying50. len +1 d datetime21.2 + 1 x $3.;
    put corid= j= d= datetime21.2 msg= x=;
run;

%let url='http://localhost:8161/api/message/testq?type=queue';
%let moreOpts=user=admin password=admin;
filename myq activemq url=&url &moreOpts destination=testq;

data _null_;
    file myq correlId=corId;
    j=0;
    do i=1 to 3;
        d=datetime(); j=j+1; corId='Unpaid';
        put j @5 'Went unpaid >30 days.' +1 d datetime21.2 +1 'Xyz';
        d=datetime(); j=j+1; corId='Paid';
        put j @5 'Paid on time.' +1 d datetime21.2 +1 'Xyz';
    end;
run;
filename myq activemq url=&url &moreOpts
    msgSelector="JMSCorrelationID='Unpaid'";
filename sameq activemq url=&url &moreOpts
    ctype='application/x-www-form-urlencoded'
    msgBodyPrefix='&JMSCorrelationID=Paid&CorrelationID=Paid&body=';
data _null_;
    rc = fcopy( 'myq', 'sameq' );
    if rc ^= 0 then do;
        msg = sysmsg(); put rc= msg=; 
    end;
run;
data _null_;
    length msg $50;
    file log;
    infile 'testq' activemq url=&url &moreOpts
        length=len correlid=corid;
    corid='';
    input j @@;
    len = len - 30;
    input @5 msg $varying50. len +1 d datetime21.2 + 1 x $3.;
    put corid= j= d= datetime21.2 msg= x=;
run;

Program Description

Initialize the filerefs for the sample.If necessary, change the URL=, USER=, and PASSWORD= options to match your ActiveMQ deployment.

%let url='http://localhost:8161/api/message/testq?type=queue';
%let moreOpts=user=admin password=admin;
filename myq activemq url=&url &moreOpts destination=testq;

Put six messages on a queue named TESTQ.The CORRELID= option specifies the variable that is used for the correlation ID.

data _null_;
    file myq correlId=corId;
    j=0;
    do i=1 to 3;
        d=datetime(); j=j+1; corId='Unpaid';
        put j @5 'Went unpaid >30 days.' +1 d datetime21.2 +1 'Xyz';
        d=datetime(); j=j+1; corId='Paid';
        put j @5 'Paid on time.' +1 d datetime21.2 +1 'Xyz';
    end;
run;

Reassign the MYQ fileref to add a selector.The MSGSELECTOR= option specifies a message selector string.

filename myq activemq url=&url &moreOpts
    msgSelector="JMSCorrelationID='Unpaid'";

Create a new fileref to change the correlation IDs of the messages on the queue.The MSGBODYPREFIX= option specifies a prefix string to each message. The CTYPE= option specifies the content type application/x-www-form-urlencoded in order to write the message header parameters (through the MSGBODYPREFIX= option) as part of the message.

filename sameq activemq url=&url &moreOpts
    ctype='application/x-www-form-urlencoded'
    msgBodyPrefix='&JMSCorrelationID=Paid&CorrelationID=Paid&body=';

Use the FCOPY function to copy the contents of the MYQ fileref to the SAMEQ fileref.The FCOPY function reads each message through the MYQ fileref and then writes it back to the same queue by using the SAMEQ fileref.

data _null_;
    rc = fcopy( 'myq', 'sameq' );
    if rc ^= 0 then do;
        msg = sysmsg(); put rc= msg=; 
    end;
run;

Read the messages from the queue and write them to the log.The CORRELID= option stores the correlation ID of each message in the CORID variable.

data _null_;
    length msg $50;
    file log;
    infile 'testq' activemq url=&url &moreOpts
        length=len correlid=corid;
    corid='';
    input j @@;
    len = len - 30;
    input @5 msg $varying50. len +1 d datetime21.2 + 1 x $3.;
    put corid= j= d= datetime21.2 msg= x=;
run;

Log Output

For each message on the queue, the values are written to the log:

CORID=Paid j=2 d=21FEB2013:15:22:10.22 msg=Paid on time. x=Xyz
CORID=Paid j=4 d=21FEB2013:15:22:10.24 msg=Paid on time. x=Xyz
CORID=Paid j=6 d=21FEB2013:15:22:10.24 msg=Paid on time. x=Xyz
CORID=Paid j=1 d=21FEB2013:15:22:10.22 msg=Went unpaid >30 days. x=Xyz
CORID=Paid j=3 d=21FEB2013:15:22:10.24 msg=Went unpaid >30 days. x=Xyz
CORID=Paid j=5 d=21FEB2013:15:22:10.24 msg=Went unpaid >30 days. x=Xyz

Last updated: October 20, 2022