Spring Integration-1 :Messaging Channels

-

The main components

Essentially in Spring Integration System or for that matter any Enterprise Application integration system, There are 2 major components , those are Producers also known as Senders and Consumers also known as Receivers and these 2 high level components communicate with each other they do so using pipes also called as channels.

In Spring Integration Prouducers and consumers are known as Endpoints and pipes are known as Channels and communication occurs via Messages through Channels.

Message contains Data information or some sort of event details.

To understand more on this, lets discuss another use case

You have an application that provides customer data and perhaps customer data provided in XML format, and there is some sort of consuming application which want this customer data but this application like to have customers who have contacted in last year(filtered contacts) and also consumer want this data in json format, here we can achieve it using filter in payload because payload contains data which has to be exchanged b/w applications.

There are many types of Message Endpoints

  • Adapters (connect your channel to some other system)
  • Filter (remove some messages from channels based on header, content, etc.)
  • Transformer (convert a message content or structure)
  • Enricher (add content to the message header or payload)
  • Service activator (invoke service operations based on the arrival of a message)
  • Gateway (connect your channels without SI coupling)

We will discuss each one of them in details

Message Channels

Two general classifications of message channels

  • Pollable Channel
  • Subscribable Channel

While there are many subtypes, they all implement at least one of these SI channel interfaces. see below link for details

http://docs.spring.io/spring-integration/reference/html/messaging-channels-section.html

Pollable Channel

  1. May buffer its messages
  2. Requires a queue to hold the messages
  3. The queue has a designated capacity
  4. Waits for the consumer to get the messages
  5. Consumers actively poll to receive messages
  6. Typically a point-to-point channel
  7. Only one receiver of a message in the channel

Subscribable Channel

  1. Allows multiple subscribers (or consumers) to register for its messages.
  2. Messages are delivered to all registered subscribers on message arrival
  3. It has to manage a list or registry of subscribers.
  4. Doesn’t buffer its messages
  5. Usually used for “event” messages
  6. Notifying the subscribers that something happened and to take appropriate action.
  7. Usually used for sending information or “document” messages between endpoints

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