Spring Integration Filters (Message Selectors) :Lab-6

-

Filters allow, on the basis of a message’s content or metadata (in the message header), a message to pass from one channel to the next or reject and discard the message from the system – that is, not allowing the rejected message into the next channel. 

Not all messages are of interest. That is, there is often a need to filter out some messages
that enter a channel. Applications may only want to get messages formatted in a certain
way (XML or JSON). Applications may only be interested in certain types of data (new sales
messages but not messages about shipments). Other application may be interested in
messages with certain message headers (for example when a message was created – only
wanting to look at messages created after 5pm).
In this lab, you explore Spring Integration (SI) filters to examine messages in a channel and
accept those of interest and discard the others.
Specifically, in this lab you will:

  • Implement the SI MessageSelector interface and configure an SI filter.
  • Explore and configure a built-in SI XPath filter to sort XML messages.
  • Work with a built-in SI Validation filter to week out non-validating XML messages

Above 3 points will be covered in 3 labs and in this lab we will focus on MessageSelector and SI Filter

Here we will filter File messages (messages with File payload) looking for
those File names that begin with a specified string. You provide the filter’s logic by
implementing the MessageSelector interface and configuring the filter in XML

spring-integration-message-selector-filter.xml.xml

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"	xmlns:int="http://www.springframework.org/schema/integration"	xmlns:int-file="http://www.springframework.org/schema/integration/file"	xmlns:int-mail="http://www.springframework.org/schema/integration/mail"	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int-stream="http://www.springframework.org/schema/integration/stream"	xsi:schemaLocation="	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd	http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd	http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd	http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">	<!-- Adapter for reading files -->	<int-file:inbound-channel-adapter id="producer-file-adapter"		channel="inboundChannel" directory="/Users/dpq/Desktop/techWorrk/fileInbound"		prevent-duplicates="true">		<int:poller fixed-rate="5000" />	</int-file:inbound-channel-adapter>	<int:channel id="inboundChannel" />	<int:filter input-channel="inboundChannel" output-channel="outboundChannel"		ref="selector" />	<bean id="selector" class="com.dpq.spring.integration.helpers.FileSelector" />	<!-- a direct channel -->	<int:channel id="outboundChannel" />	<!-- Adapter for writing files -->	<int-file:outbound-channel-adapter		channel="outboundChannel" id="consumer-file-adapter" directory="/Users/dpq/Desktop/techWorrk/fileOutbound" />	<int:poller id="defaultPoller" default="true"		max-messages-per-poll="5" fixed-rate="200" /></beans>

This Filter class will implement Message selector and it will check condition which is inside accept method

FileSelector.java

package com.dpq.spring.integration.helpers;import java.io.File;import org.springframework.integration.core.MessageSelector;import org.springframework.messaging.Message;public class FileSelector implements MessageSelector {	@Override	public boolean accept(Message<?> message) {		 if (message.getPayload() instanceof File 				 && ((File) message.getPayload()).getName().startsWith("msg")) {		 return true;		 }		 return false;	}}

Startup.java

package com.dpq.filter.containers;import org.springframework.context.support.ClassPathXmlApplicationContext;import org.springframework.integration.support.MessageBuilder;import org.springframework.messaging.Message;import org.springframework.messaging.MessageChannel;public class MessageSelectorFilterStartup {	public static void main(String[] args) {		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(				"spring-integration-message-selector-filter.xml");		System.out.println("Contanier started.....");		while(true) {					}			}}

Here messages starting with ‘msg’ will be considered and same will be processed, so if we push xyz.txt,msg.txt and msg123.xml inside file inbound directory ‘/Users/dpq/Desktop/techWorrk/fileInbound‘ then only msg.txt and msg123.xml will be filtered and pushed to outbound directory “/Users/dpq/Desktop/techWorrk/fileOutbound“

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