A TCP/IP Example
- Kevin Wu
A workflow containing a TCP/IP collection agent can be set up to send responses back to the source from which the incoming data was received. This requires an APL agent (Analysis or Aggregation) to be part of the workflow.
A TCP/IP workflow can be configured to send responses to the source
To illustrate how such a workflow is defined, an example is given where an incoming UDR is validated, resulting in either the field anum or a sequence number being sent back as a reply message to the source. Depending on if one or several TCP/IP connections are allowed, the format of the reply message sent from the Analysis agent differs:
Number of Connections | Description |
|---|---|
Single Connection | A bytearray is sent back as reply. |
Multiple Connections | A UDR, extended with the built-in TCPIPUDR format. The reply message must be inserted in the response field (a bytearray). |
Note!
To keep the example as simple as possible, the valid records are not processed. Usually, no reply is sent back until the UDRs are fully validated and processed. The example aims to focus on the response handling only.
Single TCP/IP Connection
Clear the Allow Multiple Connections checkbox in the TCP/IP Collection agent to allow only one TCP/IP session at a time. If another attempt to create a connection is made while a connection already exists, the new connection is rejected.
The TCP/IP Collection Agent
To be able to send reply messages, Send Response must be selected in the configuration window of the agent. Add an Analysis agent to the workflow and connect it to the TCP/IP agent. Drag and drop in the opposite direction to create a response route in the workflow.
Also, an Ultra format for decoding of incoming data must be defined. Note that no format has to be defined for the response - it will be sent as a bytearray from the Analysis agent.
TCP/IP agent configuration
The Analysis Agent
The Analysis agent validates the incoming records and sends the response. If the field duration is less than or equal to zero, the UDR is discarded and the field anum, in form of a bytearray, is sent back as a response. All other UDRs are routed to the next agent in turn, and instead a sequence number is sent as a response.
Note the use of the synchronized keyword. Updating a global variable within a real-time workflow must be done within a synchronized function to ensure consistency between threads. By default, a real-time workflow utilizes several threads.
int seqNum;
synchronized int createSeqNum() {
seqNum = seqNum + 1;
return seqNum;
}
consume {
bytearray reply;
if ( input.duration <= 0 ) {
strToBA( reply, input.anum );
udrRoute( reply, "response" );
} else {
strToBA( reply, (string)createSeqNum() );
udrRoute( reply, "response" );
udrRoute( input, "UDRs" );
}
}Multiple TCP/IP Connections
Select the Allow Multiple Connections checkbox in the TCP/IP Collection agent to allow several simultaneous TCP/IP sessions at a time. If an attempt to open a new connection is made when the maximum number of connections are already open, the new connection is refused.
The TCP/IP Collection Agent
To send reply messages, Send Response must be selected in the configuration window of the agent. An additional connection point will appear on the agent, to which you need to link an Analysis agent. Also, an Ultra format for the decoding of the incoming data must be defined. This format must contain the built-in TCPIP format. See the section below.
TCP/IP agent configuration
The Format Definition
The incoming external format must be extended with the TCPIPUDR format.Â
external asciiSEQ_ext sequential {
ascii callId: terminated_by(":");
int seqNum: terminated_by(",");
ascii anum: terminated_by(",");
ascii bnum: terminated_by(",");
ascii causeForOutput: terminated_by(",");
int duration: terminated_by(0xA);
};
internal TCP_Int :
extends_class( "com.digitalroute.wfc.tcpipcoll.TCPIPUDR" ) {
};
in_map TCP_InMap :
external( asciiSEQ_ext ),
internal( TCP_Int ),
target_internal( ascii_TCP_TI ) {
automatic;
};
out_map ascii_TCP_outMap : external( asciiSEQ_ext ),
internal( ascii_TCP_TI ) {
automatic;
};
decoder TCPData : in_map( TCP_InMap );
encoder TCPData : out_map( ascii_TCP_outMap );The Analysis Agent
The Analysis agent validates the incoming records and sends the response. If the field duration is less than or equal to zero, the UDR is discarded, the field anum is inserted into the response field, and the complete UDR is sent back as response. All other UDRs are routed to the next agent in turn and a sequence number is inserted into the response field before any routing takes place.
Note the use of the synchronized keyword. Updating a global variable within a real-time workflow must be done within a synchronized function to ensure consistency between threads. By default, a real-time workflow utilizes several threads.
int seqNum;
synchronized int createSeqNum() {
seqNum = seqNum + 1;
return seqNum;
}
consume {
bytearray reply;
if ( input.duration <= 0 ) {
strToBA( reply, input.anum );
input.response = reply;
udrRoute( input, "response" );
} else {
strToBA( reply, (string)createSeqNum() );
input.response = reply;
udrRoute( input, "response" );
udrRoute( input, "UDRs" );
}
}