evnent.txt

https://www.bezkoder.com/spring-boot-security-login-jwt/
https://github.com/zees007/cookieAuthJwt-springBoot3-springSecurity6-implementation?source=post_page-----756f70664673--------------------------------

<dependencies>
    <dependency>
      <groupId>org.thepavel</groupId>
      <artifactId>spring-icomponent</artifactId>
      <version>1.0.8</version>
    </dependency>
//*********************************
package com.bws.infra.handler.event;

import com.bws.infra.annotation.event.BwsEventProxy;
import com.bws.infra.dto.common.EventMessage;
import com.bws.infra.utils.JsonUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.thepavel.icomponent.handler.MethodHandler;
import org.thepavel.icomponent.metadata.MethodMetadata;

@Service
@RequiredArgsConstructor
@Order(30)
public class BwsEventHandler implements MethodHandler {

  private final KafkaTemplate<String,String> kafkaTemplate;
  @Autowired
  private ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate;
  //private final SimpleDiscoveryProperties simpleDiscoveryProperties;
  @Value("${spring.application.name}")
  private String requestReplyTopic;

  @Override
  public Object handle(Object[] objects, MethodMetadata methodMetadata) {
    try {
      BwsEventProxy bwsEventProxy = methodMetadata.getSourceClassMetadata().getSourceClass()
          .getAnnotation(BwsEventProxy.class);

      String topicName = bwsEventProxy.value();

      EventMessage eMsg=new EventMessage();
      eMsg.setClassName(methodMetadata.getSourceClassMetadata().getSourceClass().getName());
      eMsg.setMethodName(methodMetadata.getSourceMethod().getName());
      eMsg.setDateTime(LocalDateTime.now());
      String[] params=new String[objects.length];
      ObjectMapper objectMapper=new ObjectMapper();
      String[] paramTypes=new String[objects.length];
      var i=0;
      for(Object o:objects){
        params[i]=objectMapper.writeValueAsString(o);
        paramTypes[i++]=o.getClass().getName();
      }
      eMsg.setParam(params);
      eMsg.setParameterTypes(paramTypes);
      eMsg.setReturnType(methodMetadata.getSourceMethod().getReturnType().getName());
      String body = JsonUtil.toJson(eMsg);
      if(!methodMetadata.getSourceMethod().getReturnType().getName().contains("void")) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topicName, body);
        record.headers()
            .add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, requestReplyTopic.getBytes()));
        RequestReplyFuture<String, String, String> sendAndReceive = replyingKafkaTemplate
            .sendAndReceive(record);
        SendResult<String, String> sendResult = sendAndReceive.getSendFuture()
            .get(60, TimeUnit.SECONDS);
      /*  sendResult.getProducerRecord().headers()
            .forEach(header -> System.out.println(header.key() + ":" + header.value().toString()));*/
        ConsumerRecord<String, String> consumerRecord = sendAndReceive.get(60, TimeUnit.SECONDS);

        Class<?> returnTypes =
            Class.forName(methodMetadata.getSourceMethod().getReturnType().getName());
        return objectMapper.readValue(consumerRecord.value(), returnTypes);
      }else {
        kafkaTemplate.send("faf-"+topicName,body);
      }
    } catch (Throwable ex) {
      ex.printStackTrace();
    }
    return null;
  }
}
//****************************************************************************


@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Component
@Handler("bwsEventHandler")
public @interface BwsEventProxy {
  @AliasFor(
  annotation = Component.class
  )
  String value() default "";
}
//************************************************************************


package com.bws.infra.events;

import com.bws.infra.dto.common.EventMessage;
import com.bws.infra.spring.SpringUtil;
import com.bws.infra.utils.JsonUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import lombok.SneakyThrows;
import org.reflections.Reflections;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

@Component
@EnableKafka
public class EventService {

  private static Map<String, Object> cacheMap = new HashMap<>();

  @KafkaListener(topics = "${spring.application.name}", groupId = "consume-${spring.application.name}")
  @SendTo
  public String handle(String obj,@Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {
    EventMessage eventMessage = JsonUtil.toObject(obj, EventMessage.class);
    return JsonUtil.toJson(executeMethodOfObject(eventMessage));
  }

  @KafkaListener(topics = "faf-${spring.application.name}", groupId = "consume-faf-${spring.application.name}")
  public void handleFaf(String obj) {
    EventMessage eventMessage = JsonUtil.toObject(obj, EventMessage.class);
    executeMethodOfObject(eventMessage);
  }


  private Object executeMethodOfObject(EventMessage eventMessage) {
    try {
      //todo: tuning performance
      Reflections reflections = new Reflections("com.bws");
      var list = reflections.getSubTypesOf(Class.forName(eventMessage.getClassName())).stream()
          .collect(
              Collectors.toList());
      Optional<Object> obj = list
          .stream()
          .map(c -> {
            Method[] m = c.getMethods();
            Optional<Method> l = Arrays.stream(m).filter(mm -> validMethod(eventMessage, mm))
                .findFirst();
            if (l.isPresent()) {
              Object refObject = SpringUtil.getBean(c);
              try {
                cacheMap.put(eventMessage.getClassName(), refObject);
                if (l.get().getReturnType().getName().contains("void")) {
                  l.get().invoke(refObject, eventMessage.getParam());
                  return Optional.empty();
                } else {
                  return l.get().invoke(refObject, eventMessage.getParam());
                }
              } catch (IllegalAccessException e) {
                e.printStackTrace();
              } catch (InvocationTargetException e) {
                e.printStackTrace();
              }
            }
            return Optional.empty();
          }).findFirst();

      return obj.get();
    } catch (Exception e) {
      e.printStackTrace();
    }
    return null;
  }

  @SneakyThrows
  private boolean validMethod(EventMessage eventMessage, Method mm) {
    Object[] newParams = new Object[eventMessage.getParam().length];
    ObjectMapper obj = new ObjectMapper();
    if (mm.getName().equals(eventMessage.getMethodName())) {
      var count = mm.getParameterCount();
      if (eventMessage.getParameterTypes().length == count) {
        Class<?>[] clz = mm.getParameterTypes();
        for (var j = 0; j < clz.length; j++) {
          if (!clz[j].getName().contains(eventMessage.getParameterTypes()[j])) {
            return false;
          } else {
            newParams[j] = obj.readValue(eventMessage.getParam()[j].toString(), clz[j]);
          }
        }
        if (!eventMessage.getReturnType().contains(mm.getReturnType().getName())) {
          return false;
        }
        eventMessage.setParam(newParams);
        return true;
      }
      ;
    }
    return false;
  }

}

//****************************************************************************
package com.bws.infra.config;

import com.bws.infra.dto.common.EventMessage;
import com.bws.infra.utils.JsonUtil;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;

@Configuration

public class KafkaConfig {

  @Value("${spring.kafka.bootstrap-servers}")
  private String bootstrapServers;

  @Value("${spring.application.name}")
  private String requestReplyTopic;
  @Bean
  public NewTopic createTopicFireAndForget() {
    return TopicBuilder.name("faf-"+requestReplyTopic)
        .partitions(1)
        .replicas(1)
        .build();
  }

  @Bean
  public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
  }

  @Bean
  public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroup");
    return props;
  }

  @Bean
  public KafkaMessageListenerContainer<String, String> replyContainer(
      ConsumerFactory<String, String> cf) {
    ContainerProperties containerProperties = new ContainerProperties(requestReplyTopic);
    return new KafkaMessageListenerContainer<>(cf, containerProperties);
  }

  @Bean
  public ProducerFactory<String, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
  }

  @Bean
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
  }

  @Bean
  public ReplyingKafkaTemplate<String, String, String> replyKafkaTemplate(
      ProducerFactory<String, String> pf, KafkaMessageListenerContainer<String, String> container) {
    var kafkarep = new ReplyingKafkaTemplate<>(pf, container);
    kafkarep.setSharedReplyTopic(true);
    return kafkarep;

  }

  @Bean
  public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
        new StringDeserializer());
  }

  @Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setRecordFilterStrategy(new RecordFilterStrategy<String, String>() {
      @Override
      public boolean filter(ConsumerRecord<String, String> consumerRecord) {
        EventMessage eventMessage = JsonUtil.toObject(consumerRecord.value(), EventMessage.class);
        if (eventMessage != null) {
          if (consumerRecord.topic().indexOf("faf") == -1) {
            Long minutes = Duration.between(eventMessage.getDateTime(),LocalDateTime.now()).toMinutes();
            System.out.println(minutes);
            if(minutes>1)
              return true;
            var k = Arrays.stream(consumerRecord.headers().toArray()).filter(h -> {
              return h.key().contains("kafka_replyTopic");
            }).findFirst();
            if (k.isPresent()) {
              return false;
            }
            return true;
          }
          return false;
        }else {
          return true;
        }
      }
    });

    factory.setConsumerFactory(consumerFactory());
    factory.setReplyTemplate(kafkaTemplate());
    return factory;
  }


}
//*********************************************************************

<dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>3.0.9</version>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>3.3.2</version>
      <scope>compile</scope>
    </dependency>
<dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-messaging</artifactId>
      <version>6.0.11</version>
    </dependency>
//******************************************************************

  kafka:
    bootstrap-servers: 172.20.238.193:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringDeserializer
    consumer:
      key-serializer: org.apache.kafka.common.serialization.StringDeserializer
      value-serializer: org.apache.kafka.common.serialization.StringDeserializer