myWorkAround included kafka and exception and security

03.microservices/currency-conversion-service/pom.xml

@@ -28,6 +28,21 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-thymeleaf</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-security</artifactId>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
+		<dependency>
+		    <groupId>com.google.guava</groupId>
+		    <artifactId>guava</artifactId>
+		    <version>19.0</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-starter-config</artifactId>
@@ -56,7 +71,11 @@
 			<groupId>org.springframework.amqp</groupId>
 			<artifactId>spring-rabbit</artifactId>
 		</dependency>
-
+
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
 
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
@@ -69,6 +88,30 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>io.jsonwebtoken</groupId>
+			<artifactId>jjwt-api</artifactId>
+			<version>0.11.5</version>
+		</dependency>
+		<dependency>
+			<groupId>io.jsonwebtoken</groupId>
+			<artifactId>jjwt-impl</artifactId>
+			<version>0.11.5</version>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.jsonwebtoken</groupId>
+			<artifactId>jjwt-jackson</artifactId> <!-- or jjwt-gson if Gson is preferred -->
+			<version>0.11.5</version>
+			<scope>runtime</scope>
+		</dependency>
 	</dependencies>
 
 	<dependencyManagement>
@@ -89,6 +132,14 @@
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
 			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>16</source>
+					<target>16</target>
+				</configuration>
+			</plugin>
 		</plugins>
 	</build>
 

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/CurrencyConversionController.java

@@ -5,18 +5,22 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.client.RestTemplate;
 
 @RestController
+@RequestMapping("/api")
 public class CurrencyConversionController {
 
 	@Autowired
 	private CurrencyExchangeProxy proxy;
 
 	@GetMapping("/currency-conversion/from/{from}/to/{to}/quantity/{quantity}")
+	@PreAuthorize("hasAnyRole('ROLE_STUDENT')")
 	public CurrencyConversion calculateCurrencyConversion(
 			@PathVariable String from,
 			@PathVariable String to,
@@ -42,7 +46,8 @@ public CurrencyConversion calculateCurrencyConversion(
 	}
 
 	@GetMapping("/currency-conversion-feign/from/{from}/to/{to}/quantity/{quantity}")
-	public CurrencyConversion calculateCurrencyConversionFeign(
+	@PreAuthorize("hasAuthority('course:write')")
+    public CurrencyConversion calculateCurrencyConversionFeign(
 			@PathVariable String from,
 			@PathVariable String to,
 			@PathVariable BigDecimal quantity

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/CurrencyConversionServiceApplication.java

@@ -1,8 +1,11 @@
 package com.in28minutes.microservices.currencyconversionservice;
 
+import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.KafkaTemplate;
 
 @SpringBootApplication
 @EnableFeignClients
@@ -12,4 +15,8 @@ public static void main(String[] args) {
 		SpringApplication.run(CurrencyConversionServiceApplication.class, args);
 	}
 
+	@Bean
+	CommandLineRunner commandLineRunner(KafkaTemplate<String,String> kafkaTemplate){
+		return args -> {kafkaTemplate.send("xecTest","hello xec from kafka");};
+	}
 }

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/Message/KafkaConsumerConfig.java

@@ -0,0 +1,58 @@
+package com.in28minutes.microservices.currencyconversionservice.Message;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+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.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfig {
+
+    @Value("localhost:9092")
+    private String bootstrapServers;
+
+    public Map<String, Object> consumerConfig() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "hello");
+        return props;
+    }
+
+
+    //creating consumer instances
+    @Bean
+    public ConsumerFactory<String, String> consumerFactory() {
+        return new DefaultKafkaConsumerFactory<>((Map<String, Object>) consumerConfig());
+    }
+
+    //listerner container
+    @Bean
+    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> factory(
+            ConsumerFactory<String, String> consumerFactory
+    ) {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+            factory.setConsumerFactory(consumerFactory);
+        return factory;
+    }
+
+    public KafkaConsumer<String, String> getConsumer() {
+        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig());
+        consumer.subscribe(Collections.singletonList("currencyConversion"));
+        return consumer;
+    }
+}

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/Message/KafkaProducerConfig.java

@@ -0,0 +1,42 @@
+package com.in28minutes.microservices.currencyconversionservice.Message;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.ConfigDef;
+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.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    public Map<String,Object> producerConfig(){
+        Map<String,Object> props=new HashMap<>();
+        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;
+    }
+
+
+    //creating producer instances
+   @Bean
+    public ProducerFactory<String,String> producerFactory(){
+        return new DefaultKafkaProducerFactory<>(producerConfig());
+    }
+
+    //send messages
+    @Bean
+    public KafkaTemplate<String,String> kafkaTemplate(ProducerFactory<String,String> producerFactory){
+        return new KafkaTemplate<>(producerFactory);
+    }
+}

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/Message/KafkaTopicConfig.java

@@ -0,0 +1,20 @@
+package com.in28minutes.microservices.currencyconversionservice.Message;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+
+@Configuration
+public class KafkaTopicConfig {
+
+    @Bean
+    public NewTopic xecTopic(){
+        return TopicBuilder.name("xecTest").build();
+    }
+
+    @Bean
+    public NewTopic currencyConversion(){
+        return TopicBuilder.name("currencyConversion").build();
+    }
+}

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/Message/api/ConsumerSuperThread.java

@@ -0,0 +1,68 @@
+package com.in28minutes.microservices.currencyconversionservice.Message.api;
+
+import com.in28minutes.microservices.currencyconversionservice.CurrencyConversion;
+import com.in28minutes.microservices.currencyconversionservice.Message.KafkaConsumerConfig;
+import io.jsonwebtoken.lang.Collections;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+@Component
+public class ConsumerSuperThread extends Thread {
+
+    public ConsumerSuperThread() {
+    }
+
+    Logger logger= LoggerFactory.getLogger(this.getClass());
+
+    @Autowired
+    private KafkaConsumerConfig kafkaConsumerConfig;
+
+    public Map<String,String> MY_MAP=new HashMap<>();
+
+    public void triggerConsumerData(String from,String to,BigDecimal quantity){
+       if(Collections.isEmpty(MY_MAP)) {
+           this.start();
+       }else {
+           BigDecimal multiple = BigDecimal.valueOf(Long.parseLong(MY_MAP.get("multiply")));
+
+           CurrencyConversion hello_from_messaging_kafka = new CurrencyConversion(1L,
+                   from, to, quantity,
+                   multiple,
+                   quantity.multiply(multiple),
+                   "Hello from Messaging kafka");
+           System.out.println(hello_from_messaging_kafka);
+           MY_MAP.clear();
+       }
+    }
+    @Override
+    public void run() {
+        BigDecimal multiple = BigDecimal.ZERO;
+        KafkaConsumer<String, String> consumer = kafkaConsumerConfig.getConsumer();
+        try {
+            while (MY_MAP.isEmpty()) {
+                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
+                for (ConsumerRecord<String, String> record : records)
+                {
+                    logger.info("topic = %s, partition = %d, offset = %d, multipleKey = %s multipleValue = %s",
+                            record.topic(), record.partition(), record.offset(),
+                            record.key(), record.value());
+
+                    MY_MAP.put("multiply", record.value().split("-")[0]);
+                }
+            }
+        } finally {
+            consumer.close();
+        }
+    }
+}

03.microservices/currency-conversion-service/src/main/java/com/in28minutes/microservices/currencyconversionservice/Message/api/CurrencyConversionKafkaController.java

@@ -0,0 +1,46 @@
+package com.in28minutes.microservices.currencyconversionservice.Message.api;
+
+import com.in28minutes.microservices.currencyconversionservice.CurrencyExchangeProxy;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.concurrent.ExecutionException;
+
+@RestController
+    @RequestMapping("/app")
+    public class CurrencyConversionKafkaController {
+        @Autowired
+        private CurrencyExchangeProxy proxy;
+        @Autowired
+        private KafkaTemplate<String,String> kafkaTemplate;
+
+        @Autowired
+        private ConsumerSuperThread consumerSuperThread;
+
+        @GetMapping("/currency-conversion-feign/from/{from}/to/{to}/quantity/{quantity}")
+        @PreAuthorize("hasAuthority('course:write')")
+        public String calculateCurrencyConversionKafka(
+                @PathVariable String from,
+                @PathVariable String to,
+                @PathVariable BigDecimal quantity
+        ) throws ExecutionException, InterruptedException {
+            kafkaTemplate.send("currencyConversion", from + "-" + to + "-" + new Date());
+            consumerSuperThread.triggerConsumerData(from, to, quantity);
+            return "Success";
+        }
+
+    }
+
+
+
+
+
+
+