SpringCloud Stream MessageRoutingCallback Sınıfı

Giriş

Açıklaması şöyle

CloudEvents is an open standard that provides a common format for describing event data and metadata, making it easier to share events between different systems. 

Örnek

Elimizde bir NewsEvent ve AlertEvent sınıf hiyerarşisi olsun. Producer CloudEventMessageBuilder sınıfını kullanarak şöyle gönderir. 

@Component
public class NewsEventProducer {

  private final StreamBridge streamBridge;

  public NewsEventProducer(StreamBridge streamBridge) {
    this.streamBridge = streamBridge;
  }
  ...
  public Message<NewsEvent> send(String key, NewsEvent newsEvent) {
     Message<NewsEvent> message = CloudEventMessageBuilder
       .withData(newsEvent)
       .setHeader("partitionKey", key)
       .build();

     streamBridge.send("news-out-0", message);
     ...
     return message;
  }
}
@Component
public class AlertEventProducer {

  private final StreamBridge streamBridge;

  public AlertEventProducer(StreamBridge streamBridge) {
    this.streamBridge = streamBridge;
  }
  ...
  public Message<AlertEvent> send(String key, AlertEvent alertEvent) {
    Message<AlertEvent> message = CloudEventMessageBuilder
      .withData(alertEvent)
      .setHeader("partitionKey", key)
      .build();

    streamBridge.send("alert-out-0", message);
    return message;
  }
}

Producer için application.properties şöyledir. alert-out-0 ve news-out-0 kanallarına yazılan mesajlar Kafka'daki news.event ve alerts.events topiclerine yazılır

spring.cloud.stream.output-bindings=news;alert
...
spring.cloud.stream.bindings.news-out-0.destination=news.events
spring.cloud.stream.bindings.alert-out-0.destination=alert.events
...

Consumer tarafında  application.properties şöyledir. Hangi topic'leri dinlemek istediğimizi belirtiriz. Kafka'daki news.event ve alerts.events topiclerini dinleriz

spring.cloud.function.definition=functionRouter
spring.cloud.stream.bindings.functionRouter-in-0.destination=news.events,alert.events

Consumer MessageRoutingCallback ile event'leri dağıtır. Şöyle yaparız. Tek yapmamız gereken appConfigurationProperties map nesnesine sınıfın fully qualified ismine karşılık gelen kanalın ismini yazmak

@Configuration
public class MessageRoutingConfig {

  private AppConfigurationProperties appConfigurationProperties;

  public MessageRoutingConfig(AppConfigurationProperties appConfigurationProperties) {
    this.appConfigurationProperties = appConfigurationProperties;
  }

  @Bean
  public MessageRoutingCallback messageRoutingCallback() {
    return new MessageRoutingCallback() {
      @Override
      public String routingResult(Message<?> message) {
        return appConfigurationProperties.getRoutingMap()
          .getOrDefault(CloudEventMessageUtils.getType(message), "unknownEvent");
      }
    };
  }

  @Bean
  public Consumer<Message<?>> unknownEvent() {
    return message -> log.warn("...", message.getHeaders(), message.getPayload());
  }
}

Gerçek consumer kodları ise şöyle

@Component
public class NewsEventConsumer {

  @Bean
  public Consumer<Message<CNNNewsCreated>> cnnNewsCreated() {
    return message -> ...;
  }

  @Bean
  public Consumer<Message<DWNewsCreated>> dwNewsCreated() {
    return message -> ...;
  }

  @Bean
  public Consumer<Message<RAINewsCreated>> raiNewsCreated() {
    return message -> ...;
  }
}

@Component
public class AlertEventConsumer {

  @Bean
  public Consumer<Message<EarthquakeAlert>> earthquakeAlert() {
    return message -> ...;
  }

  @Bean
  public Consumer<Message<WeatherAlert>> weatherAlert() {
    return message -> ...;
  }
}

SpringCloud Stream @StreamListener Anotasyonu - Kullanmayın

Örnek

application.yaml şöyle olsun

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        orderSubmissionInput:
          destination: orderSubmitted.exchange
          group: inventory        
        movementEventInput:
          destination: movementEvent.exchange
          group: inventory

Şöyle yaparız

public interface MessageChannels {
  static final String ORDER_SUBMISSION_INPUT = "orderSubmissionInput";
  static final String MOVEMENT_EVENT_INPUT = "movementEventInput";

  @Input(ORDER_SUBMISSION_INPUT)
  SubscribableChannel orderSubmissionChannel();

  @Input(MOVEMENT_EVENT_INPUT)
  SubscribableChannel movementEventChannel();
}

@Slf4j
@Component
public class OrderListener {
  @StreamListener(MessageChannels.ORDER_SUBMISSION_INPUT)
  public void handleSubmittedOrder(Order order) {
    log.info("Order received - {}", order.toString());
  }
}


@Slf4j
@Component
public class MovementEventListener {
  @StreamListener(MessageChannels.MOVEMENT_EVENT_INPUT)
  public void handleMovementEvent(MovementEvent event) {
    log.info("Movement event received - {}", event.toString());
  }
}

SpringCloud Stream @Output Anotasyonu - Kullanmayın

Giriş

@Output anotasyonu yerine StreamBridge sınıfı da kullanılabilir.

Örnek

Şeklen şöyle

application.yaml şöyle olsun

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        orderSubmissionOutput:
          destination: orderSubmitted.exchange

Şöyle yaparız. Böylece topic'e mesaj yazılır

public interface MessageChannels {

  static final String ORDER_SUBMISSION_OUTPUT = "orderSubmissionOutput";

  @Output (ORDER_SUBMISSION_OUTPUT)
  MessageChannel orderSubmissionChannel();
}


@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 = order.withSubmissionDate(Instant.now());
    orderSubmissionGateway.submitOrder(orderToBeSubmitted);
    return ResponseEntity.ok(orderToBeSubmitted);
  }
}

Örnek - Producer

Şöyle yaparız. Burada @Output ile Source arayüzünün output isimli topic'e bir şey yazacağını belirtiriz.

public interface Source {

  String OUTPUT = "output";

  @Output(Source.OUTPUT)
  MessageChannel output();

}

application.properties şöyledir

cloud.stream:
 bindings.input:
   destination: payment-approval-topic
   group: payment-service-consumer
 bindings.output:
   destination: payment-notification-topic
   contentType: application/json

Kullanmak için şöyle yaparız

@EnableBinding({Source.class})
public class SubscriptionRequestsProducer {
  private final Source source;

  public SubscriptionRequestsProducer(Source source) {
    this.source = source;
  }

  public void requestApproval(Map<String, Object> subscriptionRequest) {
    source.output().send(MessageBuilder.withPayload(subscriptionRequest).build());
  }
}

Örnek - Processor

Şöyle yaparız

public interface MyProcessor {
   String INPUT = "myInput";

   @Input
   SubscribableChannel myInput();

   @Output("myOutput")
   MessageChannel anOutput();

   @Output
   MessageChannel anotherOutput();
}

SpringCloud Stream Consumer Bean

Giriş

1. spring.cloud.stream.function.definition ile bean isimleri tanımlanır

2. Bu bean'ler spring.cloud.stream.bindings ile bir topic'e bağlanır. Binding name için açıklama şöyle. Yani tek girdi varsa 0 kullanmak yeterli.

... the binding name is determined by the framework based on this naming convention: <function name>-in-<index> where <index> is always 0 for most cases unless functions with multiple inputs and outputs.

Örnek - RabbitMq

Elimizde şöyle bir application.yaml olsun

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        updateInventory-in-0:
          destination: orderSubmitted.exchange
          group: inventory
        updateMovement-in-0:
          destination: movementEvent.exchange
          group: inventory
      function:
        definition: updateInventory;updateMovement

Şöyle yaparız

@Slf4j
@Configuration
public class MessagingFunctionConfig {
    
  @Bean
  public Consumer<Order> updateInventory() {
    return order -> log.info("Update inventory for newly submitted order - {}", 
      order.toString());
   }

   @Bean
   public Consumer<MovementEvent> updateMovement() {
     return movementEvent -> log.info("Movement event received - {}", 
       movementEvent.toString());
  }
}

Örnek - Kafka Topic

Şöyle yaparız. Burada producer, processor ve consumer bean'ler tanımlanıyor

spring:
  cloud:
    stream:
      function:
        definition: fizzBuzzProducer;fizzBuzzProcessor;fizzBuzzConsumer

      bindings:
        fizzBuzzProducer-out-0:
          destination: numbers
        fizzBuzzProcessor-in-0:
          destination: numbers
        fizzBuzzProcessor-out-0:
          destination: fizz-buzz
        fizzBuzzConsumer-in-0:
          destination: fizz-buzz
      kafka:
        binder:
          brokers: localhost:9092
          auto-create-topics: true

Tüm bean'leri tanımlamak için şöyle yaparız

@Configuration
@Slf4j
public class KafkaConfiguration {

  @Bean
  public Supplier<Flux<Integer>> fizzBuzzProducer(){
    return () -> Flux.interval(Duration.ofSeconds(5))
      .map(value -> random.nextInt(1000 - 1) + 1)
      .log();
  }

  @Bean
  public Function<Flux<Integer>, Flux<String>> fizzBuzzProcessor(){
    return longFlux -> longFlux
      .map(i -> evaluateFizzBuzz(i))
      .log();
  }

  @Bean
  public Consumer<String> fizzBuzzConsumer(){
    return (value) -> log.info("Consumer Received : " + value);
  }

  private String evaluateFizzBuzz(Integer value) {
    ...
  }
}

Örnek - Kafka Topic + Kafka Group

Önce bean'ler tanımlanır. Şöyle yaparız. consumeMessage isimli bean, Kafka üzerindeki product-topic isimli kuyruğu tüketecek.

spring:
  cloud:
    stream:
      function:
        definition: consumeMessage;produceMessage

Daha sonra binding name tanımlanır. Şöyle yaparız. 

spring:
  cloud:
    stream:
      bindings:
        consumeMessage-in-0:
          destination: product-topic
          binder: kafka
          group: product-consumer-group

Açıklaması şöyle

The properties are similar to the producer properties but here we have to use the in keyword to indicate that we want to create an incoming channel. We set the consumer group with the help of group property. This is used by Kafka to determine the offset from where it has to continue reading after restart.

Örnek - Kafka Topic + Kafka Group

application.yml şöyledir

spring:
  cloud:
    stream:
      bindings:
        onReceive-in-0:
          destination: uppercase-values-topic
          group: consumer

Şöyle yaparız

@Slf4j
@Component
public class ValuesConsumer {

  @Bean
  public Consumer<String> onReceive() {
    return (message) -> {
      log.info("Received the value {} in Consumer", message);
    };
  }
}

Örnek - batch

Consumer için batch işlemler artık destekleniyor. Şöyle yaparız. Burada batch-mode=true yapılıyor

spring.cloud.stream.bindings.input-in-0.destination=TOPIC-NAME
spring.cloud.stream.bindings.input-in-0.group=grp
spring.cloud.stream.bindings.input-in-0.content-type=application/json

spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true
spring.cloud.stream.bindings.input-in-0.consumer-properties.max.poll.records=500

Burada artık consumer List alır. Şöyle yaparız.

@Bean
public Consumer<List<String>> input() {
  return list -> {
    System.out.println(list);
    ...
  };
}

Kullanılabilecek bazı alanların açıklaması şöyle

max.poll.records

The maximum number of records returned in a single call to poll(). Note, that max.poll.records does not impact the underlying fetching behaviour.

SpringCloud Stream StreamBridge Sınıfı - Controller'dan Topic'e Erişim İçindir

Giriş

Şu satırı dahil ederiz

import org.springframework.cloud.stream.function.StreamBridge;

send metodu

Gönderilecek topic ismi ve mesaj nesnesini alır

Örnek

Şeklen şöyle

Şöyle yaparız

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
  cloud:
    stream:
      bindings:
        orderSubmissionOutput:
          destination: orderSubmitted.exchange

Açıklaması şöyle

The framework will automatically create the topic exchange “orderSubmitted.exchange” on RabbitMQ upon application initialization.

Şöyle yaparız

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

  static final String ORDER_SUBMISSION_OUTPUT = "orderSubmissionOutput";
    
  @Autowired
  private StreamBridge streamBridge;

  @PostMapping
  public ResponseEntity<Order> submitOrder(@RequestBody @Valid Order order) {
    Order orderToBeSubmitted = order.withSubmissionDate(Instant.now());
    streamBridge.send(ORDER_SUBMISSION_OUTPUT, orderToBeSubmitted);
    return ResponseEntity.ok(orderToBeSubmitted);
  }
}

Örnek

Şöyle yaparız. values-topic isimli topic'e string yazar.

@Slf4j
@RestController
public class ValueController {

  private StreamBridge streamBridge;

  public ValueController(StreamBridge streamBridge) {
    this.streamBridge = streamBridge;
  }

  @GetMapping("values/{value}")
  public ResponseEntity<String> values(@PathVariable String value) {
    log.info("Sending value {} to topic", value);
    streamBridge.send("values-topic", value);
    return ResponseEntity.ok("ok");
  }
}

Açıklaması şöyle

As you can see, there is no code or configuration in the Producer microservice that links it to RabbitMQ. The addition of the RabbitMQ binder in the dependency did all the bindings for us. This makes it very easy to switch the underlying messaging provider.

Örnek

Şöyle yaparız

@Component
public class KafkaProducer {

  @Autowired
  private StreamBridge streamBridge;

  @Scheduled(cron = "*/2 * * * * *")
  public void sendMessage(){
    streamBridge.send("producer-out-0",new Message("jack from Stream bridge"));
  }
}

SpringCloud Stream Processor Bean

Giriş

Açıklaması şöyle

Processors can forward the incoming messages towards output channels after processing it.

Açıklaması şöyle

The processor’s return type is the Function interface which has two generic parameters. The first one is the input data (reactive stream in our case) and the second one is the output. 

Spring otomatik olarak processor method ismi-in + index ve method ismi-out + index şeklinde bir topic oluşturur ve processor bu "in" topic'i dinleyip çıktısını "out" topic'e yazmaya başlar. 

Ancak bazen kullanmak istenilen topic ismi farklı olabilir. Bu durumda application.properties dosyasında 

dinlenecek topic myprocessor-in-0 altındaki destination alanında dinlemek istediğimiz topic belirtilir.

yazılacak topic ise myprocessor-out-0 altındaki destination alanında dinlemek istediğimiz topic belirtilir.

Topic İçin Index Numarası

Açıklaması şöyle

But these topics are created with a default naming standard. They are created as javaMethodName-in-<index> and javaMethodName-out-<index> where index corresponds to the index of the application instance. So, when this app is run in local, the Exchanges will get created as convertToUppercase-in-0 and convertToUppercase-out-0. But the Producer microservice publishes the event to an Exchange named as values-topic. So, unless we override the default Exchange names created by Spring, the message sent by Producer will not be read by Processor as they’ll be sending and listening to different Exchanges.

Örnek

Şöyle yaparız

spring:
  cloud:
    function:
      definition: consumer;producer
    stream:
      bindings:
        producer-out-0:
            destination : first-topic
        consumer-in-0:
            destination : first-topic

Örnek

application.yaml şöyledir. processbean-in-0 ve processbean-out-0 başlıkları altında okunacak ve yazılacak topic isimleri belirtilir.

server:
  port: 9001

spring:
  cloud:
    stream:
      function:
        definition: fizzBuzzProducer;fizzBuzzProcessor;fizzBuzzConsumer

      bindings:
        fizzBuzzProducer-out-0:
          destination: numbers
        fizzBuzzProcessor-in-0:
          destination: numbers
        fizzBuzzProcessor-out-0:
          destination: fizz-buzz
        fizzBuzzConsumer-in-0:
          destination: fizz-buzz
      kafka:
        binder:
          brokers: localhost:9092
          auto-create-topics: true

Processor girdi ve çıktı olarak Flux kullanan bir Function döndürür. Şöyle yaparız

@Bean
public Function<Flux<Integer>, Flux<String>> fizzBuzzProcessor(){
  return longFlux -> longFlux
			.map(i -> evaluateFizzBuzz(i))
			.log();
}

String evaluateFizzBuzz(Integer value) {
  if (value % 15 == 0) {
    return "FizzBuzz";
  } else if (value % 5 == 0) {
    return "Buzz";
  } else if (value % 3 == 0) {
    return "Fizz";
  } else {
    return String.valueOf(value);
  }
}

Örnek

application.properties şöyledir

spring:
  cloud:
    stream:
      bindings:
        convertToUppercase-in-0:
          destination: values-topic
          group: processor
        convertToUppercase-out-0:
          destination: uppercase-values-topic

Şöyle yaparız

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.function.Function;

@Slf4j
@Component
public class ValueProcessor {

    @Bean
    public Function<String, String> convertToUppercase() {
        return (value) -> {
            log.info("Received {}", value);
            String upperCaseValue = value.toUpperCase();
            log.info("Sending {}", upperCaseValue);
            return upperCaseValue;
        };
    }
}

Örnek

Şöyle yaparız. Burada enrichAndSendToRabbit isimli bean Kafka'daki product-topic isimli kuyruğu okur ve Rabbit'teki inventory.message.exchange isimili exchange'e yazar. Bu exchange için kullanılacak routing key değerleri de aşağıda belirtiliyor.

spring:
  cloud:
    stream:
      bindings:
        enrichAndSendToRabbit-in-0:
          destination: product-topic
          binder: kafka
          group: product-enrich-group
        enrichAndSendToRabbit-out-0:
          destination: inventory.message.exchange
          requiredGroups: inventory_group
          binder: rabbit
      rabbit:
        bindings:
          enrichAndSendToRabbit-out-0:
            producer:
              bindingRoutingKey: inventory_item_publication
              routing-key-expression: "'inventory_item_publication'"
              exchangeAutoDelete: false
              exchangeType: direct

Açıklaması şöyle

We have to bind both the input and the output streams to the corresponding channels. We already know how we can bind Kafka channels. RabbitMQ is similar, but brings in some special binding properties like the type of the exchange and the routing key. With the direct type the consumers will use the routing key to redirect the message from the given exchange towards the queue declared by the consumer. It is very useful in case of point-to-point communication.

Örnek

Şöyle yaparız. Rabbit üzerindeki inventory.message.exchange ve Kafka üzerindeki product-topic kuyruklarını okur

spring:
  cloud:
    stream:
      bindings:
        multiInMultiOut-in-0:
          group: multi_message_group
          destination: inventory.message.exchange
          binder: rabbit
        multiInMultiOut-in-1:
          destination: product-topic
          binder: kafka
          group: product-multimessage-group
        multiInMultiOut-out-0:
          destination: multi-name-topic
        multiInMultiOut-out-1:
          destination: multi-quantity-topic
      rabbit:
        bindings:
          multiInMultiOut-in-0:
            consumer:
              bindingRoutingKey: inventory_item_publication
              exchangeType: direct

Örnek

application.yml dosyası şöyle olsun

spring.cloud.stream:
  function:
    definition: orderSupplier;orderProcessor
  bindings:
    orderSupplier-out-0:
      destination: order-received
      producer:
        useNativeEncoding: true
    orderProcessor-in-0:
      destination: order-received
    orderProcessor-out-0:
      destination: order-validated
  kafka:
    bindings:
      orderSupplier-out-0:
        producer:
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            schema.registry.url: http://localhost:8081
    streams:
      binder:
        applicationId: kafka-cqrs-command-processor
        configuration:
          schema.registry.url: http://localhost:8081
          commit.interval.ms: 100
          default:
            key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

server.port: 9001

Şöyle yaparız

import org.apache.kafka.streams.kstream.KStream;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.function.Function;

@Component
public class CommandProcessor {
  @Bean
  public Function<KStream<String, ReceivedOrder>, KStream<String, ValidatedOrder>>
  orderProcessor() {

    return receivedOrdersStream -> receivedOrdersStream
      .mapValues(ProcessorUtil::validateOrder);
  }
}

Örnek

application.yml dosyası şöyle olsun

spring.cloud.stream:
  function:
    definition: itemProcessor
  # Processors
  bindings:
    # green
    itemProcessor-in-0:
      destination: order-validated
    itemProcessor-out-0:
      destination: cheap-item-ordered
    itemProcessor-out-1:
      destination: affordable-item-ordered
    itemProcessor-out-2:
      destination: expensive-item-ordered
  kafka:
    streams:
      binder:
        applicationId: kafka-cqrs-query-processor
        configuration:
          schema.registry.url: http://localhost:8081
          commit.interval.ms: 100
          default:
            key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

server.port: 9002

Processor için aggregation işlemi yapan kod şöyle olsun

import org.apache.kafka.streams.KeyValue;
import org.mapstruct.factory.Mappers;

public class ProcessorUtil {
  
  public static KeyValue<String, OrderedItem> getItem(String customerId,
    ValidatedOrder validatedOrder) {
    return new KeyValue<>(customerId,
      Mappers.getMapper(QueryMapper.class).getOrderedItemMessage(validatedOrder));
  }

  public static OrderedItemsList initializeItems() {
    return OrderedItemsList.newBuilder().setItems(new ArrayList<>()).build();
  }

  public static OrderedItemsList aggregateItems(String aggKey, OrderedItem newValue,
    OrderedItemsList aggValue) {
    int index = aggValue.getItems().indexOf(newValue);
    if (index >= 0) {
      int quantity = aggValue.getItems().get(index).getQuantity();
      aggValue.getItems().get(index).setQuantity(quantity + 1);
    } else {
      aggValue.getItems().add(newValue);
    }
    return aggValue;
  }
}

Şöyle yaparız. Burada aggregation sonucu KTable nesnelerine yazılır.

import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Predicate;

@Component
public class QueryProcessor {
  public static final String ITEM_STORE_SUFFIX = "-items-store";

  Predicate<String, OrderedItem> isItemCheap = (k, v) -> v.getPrice() < 5;
  Predicate<String, OrderedItem> isItemAffordable = (k, v) -> v.getPrice() >= 5 && v.getPrice() < 50;
  Predicate<String, OrderedItem> isItemExpensive = (k, v) -> v.getPrice() > 50;

  @Bean
  public Function<KStream<String, ValidatedOrder>, KStream<String, OrderedItem>[]> itemProcessor() {
    return validatedOrdersStream -> {
      // group the ordered items by price
      KStream<String, OrderedItem>[] orderedItemsByPriceStream = validatedOrdersStream
        .map(ProcessorUtil::getItem)
        .branch(isItemCheap, isItemAffordable, isItemExpensive);

      // materialize the groups items into separate state stores.
      // Cheap items:
      orderedItemsByPriceStream[0].groupByKey().aggregate(
        ProcessorUtil::initializeItems,
        ProcessorUtil::aggregateItems,
        Materialized.as(Price.CHEAP.label + ITEM_STORE_SUFFIX));
      // Affordable items:
      orderedItemsByPriceStream[1].groupByKey().aggregate(
        ProcessorUtil::initializeItems,
        ProcessorUtil::aggregateItems,
        Materialized.as(Price.AFFORDABLE.label + ITEM_STORE_SUFFIX));
      // Expensive items:
      orderedItemsByPriceStream[2].groupByKey().aggregate(
        ProcessorUtil::initializeItems,
        ProcessorUtil::aggregateItems,
        Materialized.as(Price.EXPENSIVE.label + ITEM_STORE_SUFFIX));

      return orderedItemsByPriceStream;
  };
}

SpringCloud Stream Producer Bean

Giriş

Producer Bean için sadece "foo-out-0" şeklinde bir topic vermek yeterli.

Örnek

Şöyle yaparız. Burada iki tane bean ismi belirtilmiş.

spring:
  cloud:
    stream:
      function:
        definition: consumeMessage;produceMessage

Producer Bean Supplier arayüzünü gerçekleştirirler. Bir Sink'e yazarlar. Producer'ın yazdığı topic'i belirtmek için şöyle yaparız. produceMessage isimli bean Kafka üzerindeki product-topic isimli kuyruğa yazacak

spring:
  cloud:
    stream:
      bindings:
        produceMessage-out-0:
          destination: product-topic
          binder: kafka

Örnek

StreamBridge ile  yazma yapılabilir.

Örnek - Reactor Sink

Elimizde şöyle bir application.yaml olsun. orderSupplier bean, order-event kuyruğuna yazar.

server:
  port: 8080
spring.cloud.stream:
  function:
    definition: orderSupplier;paymentEventConsumer;inventoryEventConsumer
  bindings:
    orderSupplier-out-0:
      destination: order-event
    paymentEventConsumer-in-0:
      destination: payment-event
    inventoryEventConsumer-in-0:
      destination: inventory-event

Kuyruğu işleyen taraf şöyledir. Processor order-event kuyruğundan okur ve payment-event kuyruğuna yazar.

spring.cloud.stream:
  function:
    definition: paymentProcessor
  bindings:
    paymentProcessor-in-0:
      destination: order-event
    paymentProcessor-out-0:
      destination: payment-event

Event gönderen tarafı şöyle yaparız. Sinks.Many Flux arayüzüne çevriliyor.

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.function.Supplier;

@Configuration
public class OrderConfig {

  @Bean
  public Sinks.Many<OrderEvent> orderSink(){
    return Sinks.many().unicast().onBackpressureBuffer();
  }

  @Bean
  public Supplier<Flux<OrderEvent>> orderSupplier(Sinks.Many<OrderEvent> sink){
    return sink::asFlux;
  }
}

Event gönderen tarafı kullanmak için şöyle yaparız. Burada Sink için Sinks.Many kullanılıyor.

@Service
public class OrderStatusPublisher {

  @Autowired
  private Sinks.Many<OrderEvent> orderSink;

  public void raiseOrderEvent(...){
    ...
    OrderEvent orderEvent = ...;
    this.orderSink.tryEmitNext(orderEvent);
  }
}

Kullanmak için şöyle yaparız

@Service
public class OrderCommandService {
    
  @Autowired
  private OrderStatusPublisher publisher;

  @Transactional
  public PurchaseOrder createOrder(OrderRequestDto orderRequestDTO){
    PurchaseOrder purchaseOrder = ...;
    this.publisher.raiseOrderEvent(purchaseOrder, OrderStatus.ORDER_CREATED);
    return purchaseOrder;
  }
}

Örnek - Reactor EmitterProcessor

Elimizde şöyle bir application.yaml dosyası olsun

spring.cloud.stream:
  function:
    definition: orderSupplier;orderProcessor
  bindings:
    orderSupplier-out-0:
      destination: order-received
      producer:
        useNativeEncoding: true
    orderProcessor-in-0:
      destination: order-received
    orderProcessor-out-0:
      destination: order-validated
  kafka:
    bindings:
      orderSupplier-out-0:
        producer:
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            schema.registry.url: http://localhost:8081
    streams:
      binder:
        applicationId: kafka-cqrs-command-processor
        configuration:
          schema.registry.url: http://localhost:8081
          commit.interval.ms: 100
          default:
            key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
            value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

server.port: 9001

Şöyle yaparız. Burada Sink için EmitterProcessor.create() kullanılıyor.

import org.mapstruct.factory.Mappers;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;

@RestController
public class CommandController {
  private final EmitterProcessor<Message<ReceivedOrder>> messageEmitterProcessor =
    EmitterProcessor.create();

  @PostMapping(value = "/orders", consumes = MediaType.APPLICATION_JSON_VALUE)
  public ResponseEntity<Order> createOrder(@RequestBody Order order) {
    // initiate asynchronous processing of the order
    ReceivedOrder receivedOrderMessage = Mappers.getMapper(CommandMapper.class)
      .getReceivedOrderMessage(order);
    messageEmitterProcessor.onNext(MessageBuilder.withPayload(receivedOrderMessage)
      .setHeader(KafkaHeaders.MESSAGE_KEY, receivedOrderMessage.getCustomerId()).build());

    // send back a response confirming the recipient of the order
    return ResponseEntity.status(HttpStatus.ACCEPTED).body(order);
  }

  @Bean
  public Supplier<Flux<Message<ReceivedOrder>>> orderSupplier() {
    return () -> messageEmitterProcessor;
  }
}

SpringCloud Stream InteractiveQueryService Sınıfı

Giriş

Şu satırı dahil ederiz

import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;

getQueryableStore metodu

KTable nesnesinin sorgulanmasını sağlar.

Örnek

Şöyle yaparız

@RestController
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;

public class QueryController {
  @Autowired
  private InteractiveQueryService queryService;

  @GetMapping(value = "/orders", produces = MediaType.APPLICATION_JSON_VALUE)
  public ResponseEntity<List<Item>> getItemsByCustomerIdAndPrice(
    @RequestParam(value = "customerId") String customerId, 
    @RequestParam(value = "price") Price price) {

    // get the item store for the given colour
    String storeName = ...;
    ReadOnlyKeyValueStore<String, OrderedItemsList> orderedItemsStore = queryServicea
      .getQueryableStore(storeName,QueryableStoreTypes.keyValueStore());

    // get the items for the given customer
    OrderedItemsList orderedItems = orderedItemsStore.get(customerId);
    if (orderedItems != null) {
      List<Item> response = Mappers.getMapper(QueryMapper.class).getItems(orderedItems
        .getItems());
      return ResponseEntity.ok(response);
    } else {
      return ResponseEntity.notFound().build();
    }
  }
}