Spring Integration Channels-Direct channel :Lab-3

-

The only difference b/w Pollable and Direct channel is , There will not be any queue size limit in Direct channel

<int:channel id=”messageChannel” />


Everything else will remain same as with pollable channel.

spring-integration-channels.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: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">	<!-- message producer / a Spring Integration wrapped Java Standard input 		stream -->	<int-stream:stdin-channel-adapter id="producer"		channel="messageChannel" />	<!-- a pair of message consumers / a pair of Spring Integration wrapped 		Java Standard output streams -->	<int-stream:stdout-channel-adapter		id="consumer1" channel="messageChannel" append-newline="true" />	<int-stream:stdout-channel-adapter		id="consumer2" channel="messageChannel" append-newline="true" />	<int:poller id="defaultPoller" default="true"		max-messages-per-poll="5" fixed-rate="200" />	<!-- a pub/sub message channel -->	<!--<int:publish-subscribe-channel id="messageChannel" />-->        <int:channel id="messageChannel" >          <!--<int:queue capacity="2"/>-->        </int:channel></beans>

Main container loader class which will load all the beans

Startup.java

import org.springframework.context.support.ClassPathXmlApplicationContext;public class Startup {	@SuppressWarnings({ "resource", "unused" })	// @SuppressWarnings({ "resource" })	public static void main(String[] args) {		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(				"/META-INF/spring/si-components.xml");		while (true) {		}			}}

Retest the application. With the pollable channel in place of the subscribable
channel, rerun the application.

The application is now running awaiting your text input. In the Console view, again
enter some text and then hit the Enter key. A text message created from the text you enter
into the Standard Input will immediately be marshalled into the direct channel. Notice how
the message is only delivered to a single (the first) consumer of the messages (like a pollable
channel but without a queue!). Note that the text is only echoed one time.

log4j:WARN No appenders could be found for logger (org.springframework.core.env.StandardEnvironment).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Contanier started…..
Deepak
Deepak
Garv
Garv

Source code can be found in below GIT repository

https://github.com/techstudioonline/spring-integration/blob/master/src/main/java/com/dpq/containers/DirectChannelStartup.java

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