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"));
  }
}

SpringBoot Test @AutoConfigureTestDatabase Anotasyonu

Giriş

Şu satırı dahil ederiz

import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;

Not : Bu anotasyon yerine test için @DataJpaTest tercih edilebilir.

replace Alanı

Testlerin bellekteki bir veri tabanı yerine gerçek veri tabanında koşmasını sağlar. Açıklaması şöyle

In-memory embedded databases generally work well for tests since they are fast and don’t require any developer installation. If, however, you prefer to run tests against a real database you can use the @AutoConfigureTestDatabase annotation

Örnek

Şöyle yaparız

@DataJpaTest
@AutoConfigureTestDatabase(replace=Replace.NONE)

Örnek

Eğer veri tabanını bellekte değil de TestContainers içinde istersek şöyle yaparız

@DataJpaTest
@Testcontainers
@ActiveProfiles("test-containers")
@AutoConfigureTestDatabase(replace = Replace.NONE)
class PersonRepositoryTestContainers {

  @Autowired
  private PersonRepository personRepository;

  @Test
  void shouldReturnAlLastNames() {
    personRepository.saveAndFlush(new Person().setFirstName("John").setLastName("Brown"));
    personRepository.saveAndFlush(new Person().setFirstName("Kyle").setLastName("Green"));
    personRepository.saveAndFlush(new Person().setFirstName("Paul").setLastName("Brown"));

    assertEquals(Set.of("Brown", "Green"), personRepository.findAllLastNames());
  }
}

Açıklaması şöyle

@DataJpaTest is annotated with @AutoConfigureTestDatabase itself. This annotation replaces any data source with the H2 instance by default. So, we need to override this behavior by adding replace=Replace.NONE property.

Yani aslında şöyle yapmanın bir anlamı yok

@DataJpaTest
@AutoConfigureTestDatabase(replace=Replace.NONE,
                           connection = EmbeddedDatabaseConnection.H2)

SpringSecurity InMemoryUserDetailsManager Sınıfı - Default UserDetailsManager Sınıfı Budur

Giriş

Şu satırı dahil ederiz

import org.springframework.security.provisioning.InMemoryUserDetailsManager;

Bu sınıf HttpBasic Authentication kullanılıyorsa tercih edilebilir.

constructor

Şöyle yaparız

@Bean
public InMemoryUserDetailsManager inMemoryUserDetailsManager() {
  UserDetails admin = User.builder()
    .username("gurkan")
    .password(passwordEncoder.encode("pass"))
    .authorities("ADMIN")
    .build();
  UserDetails user = User.builder()
    .username("mehmet")
    .password(passwordEncoder.encode("pass"))
    .authorities("USER")
    .build();
  return new InMemoryUserDetailsManager(admin, user);
}

createUser metodu

Örnek

Şöyle yaparız

@Configuration
public class SecurityConfiguration extends WebSecurityConfigurerAdapter {

  @Override
  protected void configure(HttpSecurity http) throws Exception {
    http.authorizeRequests().anyRequest().authenticated().and().httpBasic();
  }

  @Override
  @Bean
  protected UserDetailsService userDetailsService() {
    InMemoryUserDetailsManager inMemoryUserDetailsManager = 
      new InMemoryUserDetailsManager();

    UserDetails user1 = User.builder().username("user").password("password")
      .roles("USER").build();
    UserDetails user2 = User.builder().username("admin").password("password")
      .roles("ADMIN").build();

    inMemoryUserDetailsManager.createUser(user1);
    inMemoryUserDetailsManager.createUser(user2);

    return inMemoryUserDetailsManager;
  }

  @Bean
  public PasswordEncoder passwordEncoder() {
   return NoOpPasswordEncoder.getInstance();
  }
}

SpringData ElasticSearch Kullanımı

Maven

Şu satırı dahil ederiz

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

Kavramlar

Index : ilişkisel veri tabanındaki DB anlamına gelir

Type : ilişkisel veri tabanındaki Table anlamına gelir

Document : ilişkisel veri tabanındaki Row anlamına gelir

Field : ilişkisel veri tabanındaki Column anlamına gelir

Açıklaması şöyle

There are two ways to do operations on elastic search using spring boot.
1. By using ElasticsearchRestTemplate: This we should use when we want to create more complex queries.
2. By Using Repositories: This is very simple and it has all the methods defined and internally it creates elastic-based queries.

application.properties

Açıklaması şöyle

By default, application tries to connect with Elasticsearch on localhost. If we use another target URL we need to override it in configuration settings. 

ElasticsearchOperations ile işlem yapılabilir

ElasticsearchRepository kullanılabilir

Örnek

Şöyle yaparız

spring:
  data:
    elasticsearch:
      cluster-name: docker-cluster
      # Comma-separated cluster node addresses. If not specified, starts a client node.
      cluster-nodes: localhost:9200
      # Enable Elasticsearch repositories.
      repositories:
        enabled: true
      rest:
        uris: http://localhost:9200

ElasticsearchConfiguration Sınıfı

Örnek

Şöyle yaparız

@Configuration
public class ElasticsearchClientConfig extends ElasticsearchConfiguration {

	@Value("${spring.elasticsearch.rest.uris}")
	String connetionUrl;

	@Override
	public ClientConfiguration clientConfiguration() {
		return ClientConfiguration.builder()
				.connectedTo(connetionUrl)
				.build();
	}

}

AbstractElasticsearchConfiguration  Sınıfı

Eğer ElasticsearchRestTemplate kullanacaksak şu açıklamayı bilmek lazım. Açıklaması şöyle

ElasticSearchRestTemplate is built on the top of RestHighLevelClient. You can think of it as a Spring wrapper over RestHighLevelClient.

Örnek

Şöyle yaparız

import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;
import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
import org.springframework.data.elasticsearch.repository.config.
  EnableElasticsearchRepositories;

@Configuration
@EnableElasticsearchRepositories(basePackages = "com.search.elasticsearchapp.repo")
@ComponentScan(basePackages = { "com.search.elasticsearchapp" })
public class ElasticsearchClientConfig extends AbstractElasticsearchConfiguration {
  @Override
  @Bean
  public RestHighLevelClient elasticsearchClient() {

    ClientConfiguration clientConfiguration = ClientConfiguration
      .builder()
      .connectedTo("localhost:9200")
      .build();

      return RestClients.create(clientConfiguration).rest();
  }
}

Daha sonra bir nesne yaratırız

import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;

@Document(indexName = "product")
public class Product {

  @Id
  private String id;

  @Field(type = FieldType.Text, name = "name")
  private String name;

  @Field(type = FieldType.Double, name = "price")
  private Double price;

  @Field(type = FieldType.Text, name = "creator")
  private String creator;
}

Daha sonra repository için şöyle yaparız. Repository sınıfı save() vs gibi metodlar sağlıyor ancak kendi metodlarımızı da yazmak gerekebilir.

import com.search.elasticsearchapp.model.Product;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface ProductRepo extends ElasticsearchRepository<Product, Long> {

  List<Product> findByName(String name);

  List<Product> findByNameContaining(String name);

  List<Product> findByCreatorAndName(String creator, String name);
    
}

RestHighLevelClient Sınıfı

RestHighLevelClient Sınıfı yazısına taşıdım

SpringSecurity FilterChainProxy Sınıfı

Giriş

Şu satırı dahil ederiz

import org.springframework.security.web.FilterChainProxy;

Açıklaması şöyle 

FilterChainProxy — This is another layer of indirection provided by Spring security. This is also a servlet filter whose job is to invoke the relevant filters that would work on the incoming request. Like the DelegatingFilterProxy, it also does not perform any logic. This filter is however provided by the Spring security package. 

SecurityFilter(s) — This is the most interesting bit in the above chain. This class basically contains a list of actual filters that need to be invoked before the controller can handle the request. It contains filters for example — SecurityContextPersistenceFilter, HeaderWriterFilter, CsrfFilter, LogoutFilter, UsernamePasswordAuthenticationFilter etc. when one uses the default security config. FilterChainProxy queries this class to invoke each filter in a loop. 

 There can be multiple SecurityFilterChain, each having its own stack of filters which could be the same or different from the other SecurityFilterChain.

Filtreler Nasıl Yaratılırlar?

Şeklen şöyle. Yani istek servlet'e gelmeden önce bir dizi Filter'dan geçer. Her bir Filter ise kendi içinde bir başka zincire sahiptir.

Bu filter'lardan bir tanesi DelegatingFilterProxy. DelegatingFilterProxy ise bu yazıda anlatılan FilterChainProxy sınıfını çağırıyor.

Şeklen şöyle. 

Açıklaması şöyle

The text in bold — SecurityFilterAutoConfiguration, WebSecurityConfiguration and WebSecurityConfigurerAdapter are the java class names that are responsible for the creation of DelegatingFilterProxy, FilterChainProxy and SecurityFilterChain respectively.

The boxes colored yellow/red are the components that are used in the security filtering part of the request.

The box in green — DispatcherServlet is called once the request is validated by the filter chain.

SecurityFilterChain içindeki akış şeklen şöyle

SpringSecurity SecurityFilterChain Nasıl Oluşturulur

Giriş

SecurityFilterChain zinciri FilterChainProxy içindedir. Şeklen şöyle. DelegatingFilterProxy, FilterChainProxy sınıflarına bakabilirsiniz.