Feature/21 ticketing queue

build.gradle

@@ -59,6 +59,14 @@ dependencies {
 
 	testImplementation 'org.springframework.boot:spring-boot-starter-test'
 	testImplementation 'org.springframework.security:spring-security-test'
+
+	/* kafka */
+	implementation 'org.springframework.boot:spring-boot-starter-web'
+	implementation 'org.springframework.kafka:spring-kafka'
+
+	testImplementation 'org.mockito:mockito-core:4.0.0'
+
+	implementation 'org.springframework.boot:spring-boot-starter-aop'
 }
 
 tasks.named('test') {

src/main/java/com/kboticket/KboticketApplication.java

@@ -2,10 +2,12 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
 @EnableScheduling
 @SpringBootApplication
+@EnableAspectJAutoProxy(proxyTargetClass=true)
 public class KboticketApplication {
 	public static void main(String[] args) {
 

src/main/java/com/kboticket/common/CustomSpringELParser.java

@@ -0,0 +1,24 @@
+package com.kboticket.common;
+
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+
+@Slf4j
+@NoArgsConstructor
+public class CustomSpringELParser {
+
+    public static Object getDynamicValue(String[] parameterNames, Object[] args, String key) {
+        ExpressionParser parser = new SpelExpressionParser();
+        StandardEvaluationContext context = new StandardEvaluationContext();
+
+        for (int i = 0; i < parameterNames.length; i++) {
+            context.setVariable(parameterNames[i], args[i]);
+        }
+
+        return parser.parseExpression(key).getValue(context, Object.class);
+    }
+
+}

src/main/java/com/kboticket/common/constants/KboConstant.java

@@ -11,8 +11,8 @@ public class KboConstant {
     public static final String BASIC_DLIIMITER = ":";
     public static final String AUTH_HEADER_PREFIX = "Basic ";
 
-    public static final long CONNECT_TIMEOUT = 1 * 1000;
+    public static final long CONNECT_TIMEOUT = 3 * 1000;
     public static final long READ_TIMEOUT = 60 * 1000;
-    public static final long WAIT_TIME = 3 * 1000;
-    public static final long EXPIRED_TIME = 8 * 60 * 1000;
+    public static final long WAIT_TIME = 3L;
+    public static final long LEASE_TIME = 480L;
 }

src/main/java/com/kboticket/config/KafkaConsumerConfig.java

@@ -0,0 +1,39 @@
+package com.kboticket.config;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+@Configuration
+@EnableKafka
+public class KafkaConsumerConfig {
+
+    @Bean
+    public ConsumerFactory<String, Object> consumerFactory() {
+        Map<String, Object> config = new HashMap<>();
+        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        config.put(ConsumerConfig.GROUP_ID_CONFIG, "ticketing-group");
+        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+
+        return new DefaultKafkaConsumerFactory<>(config);
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory());
+
+        return factory;
+    }
+}

src/main/java/com/kboticket/config/KafkaProducerConfig.java

@@ -0,0 +1,30 @@
+package com.kboticket.config;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Bean
+    public ProducerFactory<String, Object> producerFactory() {
+        Map<String, Object> config = new HashMap<>();
+        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+        return new DefaultKafkaProducerFactory<>(config);
+    }
+
+    @Bean
+    public KafkaTemplate<String, Object> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

src/main/java/com/kboticket/config/RedissonConfig.java

@@ -0,0 +1,29 @@
+package com.kboticket.config;
+
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RedissonConfig {
+
+    @Value("${spring.data.redis.host}")
+    private String host;
+
+    @Value("${spring.data.redis.port}")
+    private int port;
+
+    private static final String REDISSON_HOST_PREFIX = "redis://";
+
+    @Bean
+    public RedissonClient redissonClient() {
+        RedissonClient redissonClient = null;
+        Config config = new Config();
+        config.useSingleServer().setAddress(REDISSON_HOST_PREFIX + host + ":" + port);
+        redissonClient = Redisson.create(config);
+        return redissonClient;
+    }
+}

src/main/java/com/kboticket/config/WebSecurityConfig.java

@@ -45,8 +45,8 @@ public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
                     "/terms/**",
                     "/games/**",
                     "/game/**",
-                    "/seat/**", "/payment-page",
-                    "/payment/**"
+                    "/seat/**", "/payment-page","/favicon.ico",
+                    "/payment/**", "/ticket-page/**"
                 ).permitAll()
                 .anyRequest().authenticated())
             .logout(logout -> logout

src/main/java/com/kboticket/config/aop/AopForTransaction.java

@@ -0,0 +1,15 @@
+package com.kboticket.config.aop;
+
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+@Component
+public class AopForTransaction {
+
+    @Transactional (propagation = Propagation.REQUIRES_NEW)
+    public Object proceed(final ProceedingJoinPoint joinPoint) throws Throwable {
+        return joinPoint.proceed();
+    }
+}

src/main/java/com/kboticket/config/kafka/Consumer/KafkaConsumer.java

@@ -0,0 +1,32 @@
+package com.kboticket.config.kafka.Consumer;
+
+import com.kboticket.controller.QueueService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class KafkaConsumer {
+
+    private final QueueService queueService;
+
+    @KafkaListener(topics="ticketing-queue", groupId="ticketing_group")
+    public void receive(ConsumerRecord<String, String> record) {
+        if (record == null) {
+            log.info("대기열이 비어 있습니다.");
+        }
+
+        long offset = record.offset();
+        String email = record.value();
+
+        processUserInQueue(offset, email);
+    }
+
+    private void processUserInQueue(long offset, String email) {
+        queueService.addToRedisQueue(email, offset);
+    }
+}

src/main/java/com/kboticket/config/kafka/hadler/KafkaErrorHandler.java

@@ -0,0 +1,15 @@
+package com.kboticket.config.kafka.hadler;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.listener.CommonErrorHandler;
+import org.springframework.stereotype.Component;
+
+@Component
+public class KafkaErrorHandler implements CommonErrorHandler {
+
+    public void handle(Exception thrownException, ConsumerRecord record, Consumer consumer) {
+        System.err.println("Error processing message: " + record.value());
+        thrownException.printStackTrace();
+    }
+}

src/main/java/com/kboticket/config/kafka/producer/KafkaProducer.java

@@ -0,0 +1,19 @@
+package com.kboticket.config.kafka.producer;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class KafkaProducer {
+
+    private final KafkaTemplate<String, Object> kafkaTemplate;
+
+    public void create(Long gameId, String email) {
+        kafkaTemplate.send("ticketing-queue", email, email);
+        log.info("User added to queue: {}", email);
+    }
+}

src/main/java/com/kboticket/config/redis/RedisConfig.java

@@ -36,7 +36,7 @@ public RedisTemplate<String, Object>  redisTemplate() {
     }
 
     @Bean
-    public RedissonClient redissonClient() {
+    public RedissonClient redissonClient_redis() {
         RedissonClient redissonClient = null;
         Config config = new Config();
         config.useSingleServer().setAddress("redis://" + redisProperties.getHost() + ":" + redisProperties.getPort());

src/main/java/com/kboticket/config/redisson/DistributedLock.java

@@ -0,0 +1,21 @@
+package com.kboticket.config.redisson;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.util.concurrent.TimeUnit;
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface DistributedLock {
+
+    String key();
+
+    TimeUnit timeUnit() default TimeUnit.SECONDS;
+
+    long waitTime() default 3L;
+
+    long leaseTime() default 480L;
+}
+

src/main/java/com/kboticket/config/redisson/DistributedLockAop.java

@@ -0,0 +1,56 @@
+package com.kboticket.config.redisson;
+
+import com.kboticket.common.CustomSpringELParser;
+import com.kboticket.config.aop.AopForTransaction;
+import com.kboticket.enums.ErrorCode;
+import com.kboticket.exception.KboTicketException;
+import java.lang.reflect.Method;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.reflect.MethodSignature;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Aspect
+@Component
+@RequiredArgsConstructor
+public class DistributedLockAop {
+
+    private final RedissonClient redissonClient;
+    private final AopForTransaction aopForTransaction;
+
+    @Around("@annotation(DistributedLock)")
+    public Object lock(final ProceedingJoinPoint joinPoint) throws Throwable {
+        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
+        Method method = signature.getMethod();
+        DistributedLock distributedLock = method.getAnnotation(DistributedLock.class);
+
+        String lockKey = (String) CustomSpringELParser
+            .getDynamicValue(signature.getParameterNames(), joinPoint.getArgs(), distributedLock.key());
+
+        RLock rLock = redissonClient.getLock(lockKey);
+
+        if (rLock.isLocked()) {
+            throw new KboTicketException(ErrorCode.ALREADY_SELECTED_SEATS);
+        }
+
+        try {
+            boolean available = rLock.tryLock(distributedLock.waitTime(), distributedLock.leaseTime(), distributedLock.timeUnit());
+            if (!available) {
+                throw new KboTicketException(ErrorCode.FAILED_TRY_ROCK);
+            }
+
+            return aopForTransaction.proceed(joinPoint);
+        } catch (KboTicketException e) {
+            if (rLock.isHeldByCurrentThread()) {
+                rLock.unlock();
+            }
+            throw new KboTicketException(ErrorCode.FAILED_DURING_TRANSACTION);
+        }
+     }
+}