adding department model

.idea/misc.xml

 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
+  <component name="ExternalStorageConfigurationManager" enabled="true" />
   <component name="MavenProjectsManager">
     <option name="originalFiles">
       <list>
         <option value="$PROJECT_DIR$/pom.xml" />
         <option value="$PROJECT_DIR$/../springMono/pom.xml" />
         <option value="$USER_HOME$/Downloads/loyalty-events-producer-master/pom.xml" />
+        <option value="$PROJECT_DIR$/../Java8Test/pom.xml" />
       </list>
     </option>
   </component>

.idea/modules.xml

@@ -2,9 +2,7 @@
 <project version="4">
   <component name="ProjectModuleManager">
     <modules>
-      <module fileurl="file://$USER_HOME$/Downloads/loyalty-events-producer-master/loyalty-events-producer.iml" filepath="$USER_HOME$/Downloads/loyalty-events-producer-master/loyalty-events-producer.iml" />
-      <module fileurl="file://$PROJECT_DIR$/../springMono/mono.iml" filepath="$PROJECT_DIR$/../springMono/mono.iml" />
-      <module fileurl="file://$PROJECT_DIR$/.idea/spring-webflux-kafka.iml" filepath="$PROJECT_DIR$/.idea/spring-webflux-kafka.iml" />
+      <module fileurl="file://$PROJECT_DIR$/.idea/modules/Java8Demo.main.iml" filepath="$PROJECT_DIR$/.idea/modules/Java8Demo.main.iml" />
     </modules>
   </component>
 </project>
\ No newline at end of file

src/main/java/com/example/kafka/controller/DepartmentController.java

+package com.example.kafka.controller;
+
+import com.example.kafka.entity.Department;
+import com.example.kafka.entity.User;
+import com.example.kafka.entity.UserResponse;
+import com.example.kafka.service.DepartmentServiceImpl;
+import com.example.kafka.service.UserServiceImpl;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@RestController
+@RequestMapping("v1/department")
+public class DepartmentController {
+    @Autowired
+    private DepartmentServiceImpl departmentService;
+
+    @Autowired
+    private UserServiceImpl userService;
+
+    /**
+     * Create user
+     * @param department The user
+     * @return The mono of User
+     */
+    @PostMapping("/create")
+    @ResponseStatus(HttpStatus.ACCEPTED)
+    public Mono<Department> createDepartment(@RequestBody Department department){
+        return departmentService.createDepartment(department);
+    }
+
+    /**
+     * Display all users data
+     * @return The Flux of user
+     */
+    @GetMapping("/all")
+    public Flux<Department> getAllDepartments(){
+        return departmentService.getAllDepartment();
+    }
+
+    /**
+     * Fetch particular user by user id
+     * @param departmentId The userId
+     * @return ResponseEntity of user
+     */
+    @GetMapping("/{departmentId}")
+    public Mono<ResponseEntity<Department>> getDepartmentById(@PathVariable String departmentId){
+        Mono<Department> department = departmentService.findById(departmentId);
+        return department.map(ResponseEntity::ok)
+                .defaultIfEmpty(ResponseEntity.notFound().build());
+    }
+
+    /**
+     * Update the user data
+     * @param departmentID The userId
+     * @param department The user data
+     * @return ResponseEntity of user
+     */
+    @PutMapping("/{departmentID}")
+    public Mono<ResponseEntity<Department>> updateDepartmentById(@PathVariable String departmentID, @RequestBody Department department){
+        return departmentService.updateDepartment(departmentID,department)
+                .map(ResponseEntity::ok)
+                .defaultIfEmpty(ResponseEntity.badRequest().build());
+    }
+
+    @GetMapping("/fetchAll/{departmentId}")
+    public Mono<UserResponse> getDepartmentAndUsersById(@PathVariable String departmentId){
+        Mono<Department> department = departmentService.findById(departmentId);
+        return department.flatMap(d-> userService.fetchUserByDepartmentName("d.getName()").collectList().map(users->{
+            UserResponse userResponse = new UserResponse();
+            userResponse.setUserList(users);
+            return userResponse;
+        }));
+    }
+
+}
+
+

src/main/java/com/example/kafka/controller/UserController.java

@@ -6,6 +6,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.reactive.function.server.ServerResponse;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -52,6 +53,11 @@ public class UserController {
                 .defaultIfEmpty(ResponseEntity.notFound().build());
     }
 
+    @GetMapping
+    public Flux<User> getUserByDepartments(@RequestParam(value = "departmentName") String departmentName){
+        return userService.fetchUserByDepartmentName(departmentName);
+    }
+
     /**
      * Update the user data
      * @param userId The userId

src/main/java/com/example/kafka/entity/Department.java

+package com.example.kafka.entity;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import lombok.*;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString
+@EqualsAndHashCode(of={"id","name"})
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@Document(value = "department")
+public class Department {
+    @Id
+    private String id;
+    private String name;
+
+}

src/main/java/com/example/kafka/entity/User.java

@@ -35,4 +35,6 @@ public class User {
      * User's balance
      */
     private double balance;
+
+    private String departmentName;
 }

src/main/java/com/example/kafka/entity/UserResponse.java

+package com.example.kafka.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString
+public class UserResponse {
+
+    /**
+     * User id
+     */
+    private String id;
+
+    /**
+     * User name
+     */
+    private List<User> userList;
+}

src/main/java/com/example/kafka/repository/DepartmentRepository.java

+package com.example.kafka.repository;
+
+import com.example.kafka.entity.Department;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+
+public interface DepartmentRepository extends ReactiveMongoRepository<Department,String> {
+
+}

src/main/java/com/example/kafka/service/DepartmentService.java

+package com.example.kafka.service;
+
+import com.example.kafka.entity.Department;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface DepartmentService {
+    Mono<Department> createDepartment(Department department);
+
+    Flux<Department> getAllDepartment();
+
+    Mono<Department> findById(String departmentId);
+
+    Mono<Department> updateDepartment(String departmentId, Department department);
+}

src/main/java/com/example/kafka/service/DepartmentServiceImpl.java

+package com.example.kafka.service;
+
+import com.example.kafka.entity.Department;
+import com.example.kafka.repository.DepartmentRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@Service
+public class DepartmentServiceImpl implements DepartmentService {
+
+    @Autowired
+    private DepartmentRepository departmentRepository;
+
+    @Override
+    public Mono<Department> createDepartment(Department department) {
+        return departmentRepository.save(department);
+    }
+
+    @Override
+    public Flux<Department> getAllDepartment() {
+        return departmentRepository.findAll();
+    }
+
+    @Override
+    public Mono<Department> findById(String departmentId) {
+        return departmentRepository.findById(departmentId);
+    }
+
+    @Override
+    public Mono<Department> updateDepartment(String departmentId, Department department) {
+        return departmentRepository.findById(departmentId).flatMap(DBdepartment -> {
+           DBdepartment.setName(department.getName());
+           return departmentRepository.save(DBdepartment);
+        });
+    }
+}

src/main/java/com/example/kafka/service/UserService.java

@@ -2,6 +2,7 @@ package com.example.kafka.service;
 
 import com.example.kafka.entity.Order;
 import com.example.kafka.entity.User;
+import org.springframework.beans.factory.annotation.Autowired;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -17,4 +18,6 @@ public interface UserService {
     Mono<User> findById(String userId);
 
     Mono<User> updateUser(String userId, User user);
+
+    Flux<User> fetchUserByDepartmentName(String department);
 }

src/main/java/com/example/kafka/service/UserServiceImpl.java

@@ -5,8 +5,14 @@ import com.example.kafka.entity.User;
 import com.example.kafka.enumerator.OrderStatus;
 import com.example.kafka.repository.OrderRepository;
 import com.example.kafka.repository.UserRepository;
+import com.mongodb.reactivestreams.client.MongoClients;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.stereotype.Service;
+import org.springframework.util.ObjectUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -62,9 +68,26 @@ public class UserServiceImpl implements UserService {
     @Override
     public Mono<User> updateUser(String userId, User user){
         return userRepository.findById(userId).flatMap(dbUser -> {
-                    dbUser.setAge(user.getAge());
-                    dbUser.setBalance(user.getBalance());
+                    if(!ObjectUtils.isEmpty(user.getAge()) && user.getAge()>0) {
+                        dbUser.setAge(user.getAge());
+                    }
+                    if(!ObjectUtils.isEmpty(user.getBalance()) && user.getBalance()>0) {
+                        dbUser.setBalance(user.getBalance());
+                    }
+                    if(!ObjectUtils.isEmpty(user.getDepartmentName())) {
+                        dbUser.setDepartmentName(user.getDepartmentName());
+                    }
                     return userRepository.save(dbUser);
                 });
     }
+
+    @Override
+    public Flux<User> fetchUserByDepartmentName(String department) {
+        ReactiveMongoOperations ops = new ReactiveMongoTemplate(MongoClients.create(), "test");
+        Criteria c = Criteria.where("departmentName").is(department);
+        Query qry = new Query(c);
+        return ops.find(qry, User.class);
+    }
+
+
 }