Spring Integration Routers LAB-12

-

Routers direct messages to an appropriate message channel typically based on what is in
the payload or header of a message. Routers do not alter the message like a transformer.
Routers don’t remove messages from the system like a filter can. They simply provide
forks in the road of message channels in a Spring Integration (SI) application. In this lab,
you explore a few commonly used SI routers.
Specifically, in this lab you will:

  • Configure and use an XPath router to route XML payload messages.
  • Define a recipient router to send the same message to a collection of channels.

Scenario – Route using XPath expression
In previous filter Lab, we used XPath expressions in filters to remove or discard some XML messages
from the system. In a similar fashion, as will be demonstrated in this part of the lab, an
XPath router uses an XPath expression to determine which message channel receives a
message containing XML in its payload.

In particular, note that the application already contains an inbound file adapter and
a file-to-string transformer to take XML files and get them to String form. Below is an EIP
model representing what is currently defined in the si-components.xml file. The adapter
gets XML files to the inboundChannel and then the transformer converts the file messages to
string messages for the xml-inboundChannel.
At this point, there are no outbound adapters.

Create an XPath router The ship order XML messages will again be used in the last to experiment with routers. The XML ship order messages contain a element nested in a element which in turn is nested in the root element. In this step, you build an XPath router to direct those XML messages containing “Norway” as the elements value to a norwayChannel and those containing “USA” to a usaChannel.

Add two messages channels – one for USA ship orders and one for Norway ship orders.

Add two outbound file adapters. Add two file adapters; one for USA ship orders and one for Norway ship orders. Each file adapter will place the messages it receives from the appropriate message channel in separate file system directories so that the routing can be seen to work.

  • Add a Norway outbound file adapter, as shown below, to place its messages in a directory called outboundNorway.
  • Add a USA outbound file adapter to place its messages in a directory called outboundUSA.

Add an XPath message router. The message router should use and XPath expression to get the value in the element and map it to the appropriate message channel as defined in step above.

Note the use of mapping elements in the router to map the value discovered in the
XPath expression to one of two message channels.

With the addition of this router component, you have provided a “fork” that
distributes the inbound XML messages to the appropriate directory based on the
content of the message.

The XPath router is a type of content router as it uses some part of the content of
the message – in this case an element from the XML – to route the message through
its intended path in the system.

Test the Router

In the si-components.xml, find the producer-file-adapter. Note the location of the
directory. It is set, by default, to file:c://inboundXML. This is the location where the XML
ship order messages will be taken into the application by the adapter. It is the same folder
used in the last lab (lab 4). If it does not already exist, create this message folder – changing
the location to suit your needs and your file system (change the producer-file-adapter to
reflect your location).

Test the application. Test the application to see the Files in the inbound file folder are routed to the correct outbound folder based on the element value

Using a Windows Explorer, open the two folders specified as the output adapter’s directories (these should be c://outboundNorway and c://outboundUsa per configuration in step 2.2 above). See that the messages have been routed to the correct folder based on the element in the message.

Stop the application. Recall the application is running in an infinite loop to allow
for the constant publishing and consuming of messages. Stop the application now

TO DO EXAMPLE AND GIT REPO details

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more