SpringIntegration Splitter

Giriş

Açıklaması şöyle

A Splitter takes a single message and splits it into multiple messages based on specific criteria.

Örnek

Şöyle yaparız

@Configuration
public class SplitterConfig {

    @Bean
    public MessageChannel inputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel outputChannel() {
        return new DirectChannel();
    }

    @Bean
    @Splitter(inputChannel = "inputChannel", outputChannel = "outputChannel")
    public AbstractMessageSplitter splitter() {
        return new AbstractMessageSplitter() {
            @Override
            protected List<?> splitMessage(Message<?> message) {
                return Arrays.asList(message.getPayload().toString().split(","));
            }
        };
    }
}

Açıklaması şöyle

In this configuration, we define two DirectChannel beans: inputChannel and outputChannel. We also define a Splitter that splits the message payload into multiple messages based on the comma delimiter and sends them to the outputChannel.

SpringIntegration Aggregator

Giriş

Açıklaması şöyle

An Aggregator collects and combines messages that share a common correlation before sending them as a single message.

Örnek

Şöyle yaparız

@Configuration
public class AggregatorConfig {

    @Bean
    public MessageChannel inputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel outputChannel() {
        return new DirectChannel();
    }

    @Bean
    @Transformer(inputChannel = "inputChannel", outputChannel = "aggregatorChannel")
    public HeaderEnricher correlationHeaderEnricher() {
        Map<String, Expression> headersToAdd = new HashMap<>();
        headersToAdd.put("correlationId", new ValueExpression<>("aggregatedPayload"));
        return new HeaderEnricher(headersToAdd);
    }

    @Bean
    public PollableChannel aggregatorChannel() {
        return new QueueChannel();
    }

    @Bean
    @Aggregator(inputChannel = "aggregatorChannel", outputChannel = "outputChannel")
    public DefaultAggregatingMessageGroupProcessor aggregator() {
        return new DefaultAggregatingMessageGroupProcessor();
    }
}

Açıklaması şöyle

In this configuration, we define two DirectChannel beans: inputChannel and outputChannel. We use a Transformer to enrich the message headers with a correlationId. The Aggregator combines messages with the same correlationId and sends the aggregated message to the outputChannel.

Örn

SpringIntegration Content-Based Router

Örnek

Şöyle yaparız

@Configuration
public class ContentBasedRoutingConfig {

  @Bean
  public MessageChannel inputChannel() {
    return new DirectChannel();
  }

  @Bean
  public MessageChannel evenChannel() {
    return new DirectChannel();
  }

  @Bean
  public MessageChannel oddChannel() {
    return new DirectChannel();
  }

  @Bean
  Router(inputChannel = "inputChannel")
  public ExpressionEvaluatingRouter router() {
    SpelExpressionParser parser = new SpelExpressionParser();
    Expression expression = parser.parseExpression("payload % 2 == 0 ? 
      'evenChannel' : 'oddChannel'");
    return new ExpressionEvaluatingRouter(expression);
  }
}

SpringIntegration Message Channel (Point-to-Point)

Örnek

Şöyle yaparız

@Configuration
public class PointToPointConfig {

  @Bean
  public MessageChannel inputChannel() {
    return new DirectChannel();
  }

  @Bean
  public MessageChannel outputChannel() {
    return new DirectChannel();
  }

  @Bean
  @ServiceActivator(inputChannel = "inputChannel", outputChannel = "outputChannel")
  public Transformer uppercaseTransformer() {
    return new AbstractTransformer() {
      @Override
      protected Object doTransform(Message<?> message) {
        return message.getPayload().toString().toUpperCase();
      }
    };
  }
}

SpringIntegration Redis JdbcLockRegistry Sınıfı

Örnek

Şöyle yaparız

@Bean
public DefaultLockRepository DefaultLockRepository(DataSource dataSource){
  return new DefaultLockRepository(dataSource);
}

@Bean
public JdbcLockRegistry jdbcLockRegistry(LockRepository lockRepository){
  return new JdbcLockRegistry(lockRepository);
}

SpringIntegration Distributed Lock

Giriş

Distributed Lock iki şekilde gerçekleştirilebilir

1. Redis

2. JDBC

Redis

Maven

Şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>

 <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
 </dependency>
 <dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-redis</artifactId>
 </dependency>
 <dependency>
  <groupId>io.lettuce</groupId>
  <artifactId>lettuce-core</artifactId>
 </dependency>

RedisLockRegistry sınıfı kullanılır

JDBC

Maven

Açıklaması şöyle

The JDBC version of the distributed lock needs the database to have some tables and indexes set up in order to work. If you do not set these up the first time you attempt to obtain the lock, a JDBC Exception will be thrown. The current collection of SQL files for this can be found in the Spring Integration JDBC github repo.

In the following example, Flyway runs the SQL script automatically.

Şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>

<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-jdbc</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
  <groupId>org.postgresql</groupId>
  <artifactId>postgresql</artifactId>
</dependency>

<dependency>
  <groupId>org.flywaydb</groupId>
  <artifactId>flyway-core</artifactId>
</dependency>

JdbcLockRegistry sınıfı kullanılır

Genel Kullanım

Elimizde şöyle bir kod  olsun

public interface LockService {
  String lock();
  void failLock();
  String properLock();
}

@Service
public class RedisLockService implements LockService{
  private static final String MY_LOCK_KEY = "someLockKey";
  private final LockRegistry lockRegistry;

  public RedisLockService(LockRegistry redisLockRegistry) {
    this.lockRegistry = redisLockRegistry;
  }
}

@Service
public class JDBCLockService implements LockService{
  private static final String MY_LOCK_KEY = "someLockKey";
  private final LockRegistry lockRegistry;

  public JDBCLockService(JdbcLockRegistry jdbcLockRegistry) {
    this.lockRegistry = jdbcLockRegistry;
  }
}

Şöyle yaparız. obtain() ile bir Lock nesnesi elde edilir. tryLock() ile bu kilitlenir.

a@Override
public String lock() {
  Lock lock = null;
  String returnVal = null;
  try {
    lock = lockRegistry.obtain(MY_LOCK_KEY);
  
    if (lock.tryLock()) {
      returnVal =  "lock successful";
    }
    else {
      returnVal = "lock unsuccessful";
    }
  } catch (Exception e) {
    ...
  } finally {
    if (lock != null) {
      lock.unlock();
    }
  }
  return returnVal;
}

SpringIntegration Redis @RedisLockable Anotasyonu

Giriş

Kilidi kodla bırakmak için SpringIntegration Redis RedisLockRegistry Sınıfı yazısına bakabilirsiniz

Maven

Şu satırı dahil ederiz

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-redis</artifactId>
    <version>5.5.0</version>
</dependency>

Örnek

application.properties şöyle olsun

@Configuration
public class AppConfig {

  @Bean
  public RedisConnectionFactory redisConnectionFactory() {
    return new JedisConnectionFactory();
  }
}


@Service
public class MyService {

  @RedisLockable(key = "my-lock-key", leaseTime = 60_000, waitTime = 5_000)
  public void doSomethingWithLock() throws InterruptedException {
    // Do something with the lock
  }
}

Şöyle yaparız

@Configuration
public class AppConfig {

  @Bean
  public RedisConnectionFactory redisConnectionFactory() {
    return new JedisConnectionFactory();
  }
}


@Service
public class MyService {

  @RedisLockable(key = "my-lock-key", leaseTime = 60_000, waitTime = 5_000)
  public void doSomethingWithLock() throws InterruptedException {
    // Do something with the lock
  }
}

Açıklaması şöyle

In this example, we annotate the doSomethingWithLock method with the @RedisLockable annotation, which specifies the key to use for the lock, the lease time (in milliseconds), and the wait time (in milliseconds). The lease time determines how long the lock should be held before it is automatically released, while the wait time determines how long the method should wait to acquire the lock before giving up and throwing an exception.

SpringIntegration@MessagingGateway Anotasyonu

Giriş

Şu satırı dahil ederiz 

import org.springframework.integration.annotation.MessagingGateway;

Örnek

Şöyle yaparız

@MessagingGateway
public interface OrderSubmissionGateway {

  @Gateway(requestChannel = MessageChannels.ORDER_SUBMISSION_OUTPUT)
  void submitOrder(Order order);
}

@RestController
@RequestMapping("/orders")
public class OrderRestController {

  @Autowired
  private OrderSubmissionGateway orderSubmissionGateway;

  @PostMapping()
  public ResponseEntity<Order> submitOrder(@RequestBody @Valid Order order) {
    Order orderToBeSubmitted = ...
    orderSubmissionGateway.submitOrder(orderToBeSubmitted);
    return ResponseEntity.ok(orderToBeSubmitted);
  }
}

SpringIntegration Kullanımı

Maven

Şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>

Daha sonra gerekli bileşeni dahil ederiz. Örneğin ftp ile dosya çekeceksek şöyle yaparız

<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-ftp</artifactId>
</dependency>

Açıklaması şöyle

- If you throw in spring-integration-jmx, then a plethora of messaging metrics will automatically be collected across all your flows and published to JMX endpoints.

- spring-integration-jdbc provides the means to use an underlying JDBC data store as the messaging mechanism.

- spring-integration-amqp makes it possible to use RabbitMQ or any other AMQP-compliant broker as the primary mechanism to send messages between components

Message Arayüzü

Şu satırı dahil ederiz

import org.springframework.integration.Message;

getHeaders() ve getPayload() diye iki tane metod sunar. Böylece ortaya çok generic bir yapı çıkar.

Multiple Protocols

Çok çeşitli protokolleri destekler. Açıklaması şöyle

Spring Integration offers support for a wide array of communication protocols, including FTP, JMS, HTTP, STOMP, SOAP, REST, RSocket and even email.

Enterprise Integration Patterns

Açıklaması şöyle

Spring Integration is originated as an implementation of (most of) the patterns found in Enterprise Integration Patterns (by Hohpe and Woolf). This books contains a collection of common integration problems combined with their solutions. These patterns enable developers to tackle non-linear problems by breaking them down into smaller, more manageable pieces.

Some of the key patterns included in Spring Integration include:

- Channels

- Aggregators

- Filters

- Transformers

- Control Buses

- Gateways

- Endpoints

SpringIntegration IntegrationFlowBuilder

Örnek - Consumer
Şöyle yaparız.

@Bean
public IntegrationFlow lines(FileWritingMessageHandler fileOut) {
  return f -> f.handle(fileOut);
}

publishSubscribeChannel metodu
Açıklaması şöyle.

The publishSubscribeChannel() method broadcasts messages to all subscribing subflows.
...
In this way, the subflows are anonymous, meaning that they can't be independently addressed.

Örnek - Consumer + Producer
Şöyle yaparız.

@Bean
public IntegrationFlow markers() {
   return f -> f.<FileSplitter.FileMarker>filter(m ->...,...)
          .publishSubscribeChannel(s -> s
                  // first trigger file flushes
              .subscribe(...)
                   // send the first file
              .subscribe(...)
                   // send the second file
               .subscribe(...)
                   // send the third file
               .subscribe(...)
                   // send an email
               .subscribe(...));
}

SpringIntegration IntegrationFlows Sınıfı

Giriş
Açıklaması şöyle.

An integration scenario is represented using an “IntegrationFlow” interface in Spring Integration. One of the ways of creating this would be to use a builder pattern starting with the “IntegrationFlows” class. Here, you can use this to create an input channel adapter in order to get messages into the system and connect them to a message channel. Each message that is created and put through the message channels are represented using the “Message” interface. This will contain a domain-specific “payload” value and a set of headers.

route metodu - Producer Flow
Açıklaması şöyle. markers ve lines isimli iki tane IntegrationFlow tipinden bean olduğu anlatılıyor.

For each bean representing a message endpoint, you can reach its implicit input channel by simply appending “.input” to the bean name. This is how the “markers” and “lines” message endpoints’ input channels are named “markers.input” and “lines.input”, respectively. 

Örnek
Şöyle yaparız. Yapılan işlem sonucunu route() metodu ile başka kanala gönderir.

@Bean
public IntegrationFlow fromFile() {
  return IntegrationFlows.from(...)
    .handle(...)
    .<Object, Class<?>>route(...)
    .get();
}

SpringIntegration MessageChannels Sınıfı

publishSubscriber metodu
Örnek
Şöyle yaparız.

@Configuration
public class FTPIntegration {


  @Bean
  public MessageChannel outputIntegrationChannel(ErrorHandler errorHandler) {
    return MessageChannels
            .publishSubscribe()
            .errorHandler(errorHandler)
            .get();
  }

  @Bean
  public IntegrationFlow ftpIntegration(ErrorHandler errorHandler,
    MessageHandler messageHandler) {
    return IntegrationFlows
            .from(outputIntegrationChannel(errorHandler))
            .transform(Transformers.toJson())
            .handle(messageHandler)
            .get();
  }
}

Örnek
Şöyle yaparız.

@Bean(name = {"publishCha.input", "publishCha2.input"}) //2 subscribers
public MessageChannel publishAction() {
  PublishSubscribeChannel ps = MessageChannels.publishSubscribe().get();
  ps.setMaxSubscribers(8);
  return ps;
}

SpringIntegration FileWritingMessageHandler Sınıfı

constructor
Şöyle yaparız.

FileWritingMessageHandler handler = 
  new FileWritingMessageHandler(new File(processingDir));

setDeleteSourceFiles metodu
Şöyle yaparız.

handler.setDeleteSourceFiles(true);

setExpectReply metodu
Şöyle yaparız.

handler.setExpectReply(false);

setFileExistsMode metodu
Şöyle yaparız.

handler.setFileExistsMode(FileExistsMode.FAIL);

setFileNameGenerator metodu
Örnek
Elimizde şöyle bir kod olsun.

@Bean
public DefaultFileNameGenerator processingFileNameGenerator() {
  DefaultFileNameGenerator defaultFileNameGenerator = new DefaultFileNameGenerator();
  ...
  return defaultFileNameGenerator;
}

Şöyle yaparız.

handler.setFileNameGenerator(processingFileNameGenerator());