Initial commit

.gitignore

+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### IntelliJ IDEA ###
+.idea/modules.xml
+.idea/jarRepositories.xml
+.idea/compiler.xml
+.idea/libraries/
+*.iws
+*.iml
+*.ipr
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store
\ No newline at end of file

pom.xml

+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>3.4.1</version>
+    <relativePath/> <!-- lookup parent from repository -->
+  </parent>
+
+  <groupId>org.nisum</groupId>
+  <artifactId>Notification-system</artifactId>
+  <version>1.0-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <name>Notification-system</name>
+  <url>http://maven.apache.org</url>
+
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-webflux -->
+    <!--    For Reactive web support-->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-webflux</artifactId>
+      <version>3.4.0</version>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
+    <!-- To reduce boilerplate code-->
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <version>1.18.34</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
+    <!-- Starter for kafka    -->
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka</artifactId>
+      <version>3.3.2</version>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-mongodb-reactive -->
+    <!-- Starter for MongoDB    -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+      <version>3.4.1</version>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-logging -->
+    <!-- Starter for logging and testing    -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-logging</artifactId>
+      <version>3.3.6</version>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test -->
+    <!-- Starter test for reactive testing   -->
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <version>3.4.0</version>
+      <scope>test</scope>
+    </dependency>
+
+
+
+
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>3.8.1</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

src/main/java/org/nisum/NotificationApplication.java

+package org.nisum;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * Hello world!
+ *
+ */
+@SpringBootApplication
+public class NotificationApplication
+{
+    public static void main( String[] args )
+    {
+
+        SpringApplication.run(NotificationApplication.class, args);
+    }
+}

src/main/java/org/nisum/consumer/NotificationConsumer.java

+package org.nisum.consumer;
+
+import org.nisum.model.Notification;
+import org.nisum.repository.NotificationRepository;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+
+import java.time.LocalDateTime;
+
+@Service
+public class NotificationConsumer {
+
+    private final NotificationRepository notificationRepository;
+
+    public NotificationConsumer(NotificationRepository notificationRepository) {
+        this.notificationRepository = notificationRepository;
+    }
+    @KafkaListener(topics = "notification-topic", groupId = "notification-group")
+    public void listenNotifications(String message){
+        Notification notification = new Notification();
+        notification.setMessage(message);
+        notification.setTimestamp(LocalDateTime.now());
+        notification.setRead(false);
+
+        notificationRepository.save(notification).subscribe();
+    }
+}

src/main/java/org/nisum/controller/NotificationController.java

+package org.nisum.controller;
+
+import org.nisum.model.Notification;
+import org.nisum.service.NotificationService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequestMapping("/api/notifications")
+public class NotificationController {
+
+    @Autowired
+    private final NotificationService notificationService;
+
+    public NotificationController(NotificationService notificationService) {
+        this.notificationService = notificationService;
+    }
+
+    @PostMapping("/send")
+    public Mono<String> sendNotification(@RequestParam String userId,@RequestParam String message){
+        notificationService.sendNotification(userId,message);
+        return Mono.just("Success");
+    }
+
+    @GetMapping("/send-notification")
+    public Mono<String> sendNotification(){
+        return notificationService.sendNotification();
+    }
+
+    @GetMapping("/user/{userId}")
+    public Flux<Notification> getNotification(@PathVariable String userId){
+        return notificationService.getNotificationForUser(userId);
+    }
+
+
+}

src/main/java/org/nisum/model/Notification.java

+package org.nisum.model;
+
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+
+
+@Document(collection = "notifications")
+public class Notification {
+
+    @Id
+    private String id;
+    private String userId;
+    private String message;
+    private boolean read;
+    private LocalDateTime timestamp;
+
+    public Notification() {
+    }
+
+    public Notification(String id, String userId, String message, boolean read, LocalDateTime timestamp) {
+        this.id = id;
+        this.userId = userId;
+        this.message = message;
+        this.read = read;
+        this.timestamp = timestamp;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    public boolean isRead() {
+        return read;
+    }
+
+    public void setRead(boolean read) {
+        this.read = read;
+    }
+
+    public LocalDateTime getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(LocalDateTime timestamp) {
+        this.timestamp = timestamp;
+    }
+}

src/main/java/org/nisum/producer/NotificationProducer.java

+package org.nisum.producer;
+
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class NotificationProducer {
+
+    private final KafkaTemplate<String,String> kafkaTemplate;
+
+    public NotificationProducer(KafkaTemplate<String,String> kafkaTemplate){
+        this.kafkaTemplate = kafkaTemplate;
+    }
+
+    //Method to send notification
+    public void sendNotification(String message){
+        kafkaTemplate.send("notification-topic",message);
+    }
+}

src/main/java/org/nisum/repository/NotificationRepository.java

+package org.nisum.repository;
+
+import org.nisum.model.Notification;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Flux;
+
+@Repository
+public interface NotificationRepository extends ReactiveMongoRepository<Notification,String> {
+    Flux<Notification> findByUserId(String userId);
+}

src/main/java/org/nisum/service/NotificationService.java

+package org.nisum.service;
+
+import org.nisum.model.Notification;
+import org.nisum.producer.NotificationProducer;
+import org.nisum.repository.NotificationRepository;
+import org.nisum.webclientconfig.WebClientConfig;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.LocalDateTime;
+
+@Service
+public class NotificationService {
+
+    @Autowired
+    private final NotificationProducer notificationProducer;
+    @Autowired
+    private final NotificationRepository notificationRepository;
+
+
+    private final WebClient webClient;
+
+
+    public NotificationService(NotificationProducer notificationProducer, NotificationRepository notificationRepository,WebClient.Builder webClientBuilder) {
+        this.notificationProducer = notificationProducer;
+        this.notificationRepository = notificationRepository;
+        this.webClient = webClientBuilder.baseUrl("https://jsonplaceholder.typicode.com").build();
+    }
+
+    public void sendNotification(String userId, String message){
+        notificationProducer.sendNotification(message);
+        Notification notification = new Notification();
+        notification.setUserId(userId);
+        notification.setMessage(message);
+        notification.setRead(false);
+        notification.setTimestamp(LocalDateTime.now());
+
+        notificationRepository.save(notification).subscribe();
+    }
+
+    public Mono<String> sendNotification(){
+            return this.webClient.get()
+                    .uri("/posts")
+                    .retrieve()
+                    .bodyToMono(String.class);
+    }
+
+    public Flux<Notification> getNotificationForUser(String userId){
+        return notificationRepository.findByUserId(userId);
+    }
+}

src/main/java/org/nisum/webclientconfig/WebClientConfig.java

+package org.nisum.webclientconfig;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration
+public class WebClientConfig {
+
+    @Bean
+    public WebClient.Builder webClientBuilder(){
+        return WebClient.builder();
+    }
+}

src/main/resources/application.yml

+server:
+  port: 8081
+spring:
+  kafka:
+    consumer:
+      group-id: notification-group
+      auto-offset-reset: earliest
+      bootstrap-servers: localhost:9092
+    producer:
+      bootstrap-servers: localhost:9092
+
+  data:
+    mongodb:
+      uri: mongodb://localhost:27017/notificationdb
+
+  webflux:
+    base-path: /
+

src/test/java/org/nisum/NotificationApplicationTest.java

+package org.nisum;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+/**
+ * Unit test for simple App.
+ */
+public class NotificationApplicationTest
+    extends TestCase
+{
+    /**
+     * Create the test case
+     *
+     * @param testName name of the test case
+     */
+    public NotificationApplicationTest(String testName )
+    {
+        super( testName );
+    }
+
+    /**
+     * @return the suite of tests being tested
+     */
+    public static Test suite()
+    {
+        return new TestSuite( NotificationApplicationTest.class );
+    }
+
+    /**
+     * Rigourous Test :-)
+     */
+    public void testApp()
+    {
+        assertTrue( true );
+    }
+}