Merge pull request #7 from stephano-tri/feature-#3

Feature #3

Author: stephano-tri

README.md

@@ -3,6 +3,11 @@
 # copy to container
 docker cp $dump_file $container_id:/var
 
+# create database
+docker compose exec -it $db_container bash
+psql -U postgres
+create database dvdrental;
+
 # user pg_restore
 docker compose exec $db_container pg_restore -U $user -C -d postgres /var/$dump_file
 ```

docker-compose.yml

@@ -18,18 +18,38 @@ services:
         soft: -1
         hard: -1
 
-  db:
-    container_name: postgres
-    image: postgres
-    restart: always
+  postgresql-main:
+    image: 'bitnami/postgresql:latest'
     ports:
-      - 5432:5432
+      - '5432:5432'
+    restart: on-failure
     volumes:
-      - ./postgres-data:/var/lib/postgresql/data
+      - postgresql_main_data:/bitnami/postgresql
+      - ./init.sql:/docker-entrypoint-initdb. d/db.sql
     environment:
-      POSTGRES_USER: postgres
-      POSTGRES_PASSWORD: example
-      POSTGRES_DB: postgres
+      - POSTGRESQL_REPLICATION_MODE=master
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_USERNAME=my_user
+      - POSTGRESQL_PASSWORD=my_password
+      - POSTGRESQL_DATABASE=ins
+      - ALLOW_EMPTY_PASSWORD=yes
+
+  postgresql-standby:
+    image: 'bitnami/postgresql:latest'
+    ports:
+      - '5433:5432'
+    restart: on-failure
+    depends_on:
+      - postgresql-main
+    environment:
+      - POSTGRESQL_REPLICATION_MODE=slave
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_MASTER_HOST=postgresql-main
+      - POSTGRESQL_MASTER_PORT_NUMBER=5432
+      - POSTGRESQL_USERNAME=my_user
+      - POSTGRESQL_PASSWORD=my_password
 
   zookeeper:
     image: zookeeper:3.7
@@ -145,3 +165,7 @@ services:
       CONNECT_LOG4J_ROOT_LOGLEVEL: INFO
       CONNECT_LOG4J_LOGGERS: org.reflections=ERROR
       CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-3.3.0.jar
+
+volumes:
+  postgresql_main_data:
+    driver: local

dvdrental.zip

@@ -0,0 +1,2 @@
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO repl_user;
+ALTER USER REPL_USER WITH SUPERUSER;

init.sql

@@ -0,0 +1,10 @@
+package eom.improve.kafkaboot.model
+
+import java.io.Serializable
+import java.time.LocalDateTime
+
+data class FilmActor(
+    val actorId: Int,
+    val filmId: Int,
+    val lastUpdate: LocalDateTime
+) : Serializable

src/main/kotlin/eom/improve/kafkaboot/model/FilmActor.kt

@@ -0,0 +1,10 @@
+package eom.improve.kafkaboot.model
+
+import java.io.Serializable
+import java.time.LocalDateTime
+
+data class FilmCategory(
+    val filmId: Int,
+    val categoryId: Int,
+    val lastUpdate: LocalDateTime
+) : Serializable

src/main/kotlin/eom/improve/kafkaboot/model/FilmCategory.kt

@@ -1,4 +1,11 @@
 package eom.improve.kafkaboot.repository
 
-interface FilmActorRepository {
+import eom.improve.kafkaboot.model.FilmActor
+import org.springframework.data.r2dbc.repository.R2dbcRepository
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+interface FilmActorRepository : R2dbcRepository<FilmActor, Int> {
+    fun findAllByFilmId(filmId: Int) : Flux<FilmActor>
+    fun deleteByFilmId(filmId: Int) : Mono<Void>
 }

src/main/kotlin/eom/improve/kafkaboot/repository/FilmActorRepository.kt

@@ -1,4 +1,11 @@
 package eom.improve.kafkaboot.repository
 
-interface FilmCategoryRepository {
+import eom.improve.kafkaboot.model.FilmCategory
+import org.springframework.data.r2dbc.repository.R2dbcRepository
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+interface FilmCategoryRepository : R2dbcRepository<FilmCategory, Int> {
+    fun findAllByFilmId(filmId: Int) : Flux<FilmCategory>
+    fun deleteByFilmId(filmId: Int) : Mono<Void>
 }

src/main/kotlin/eom/improve/kafkaboot/repository/FilmCategoryRepository.kt

@@ -1,12 +1,8 @@
 package eom.improve.kafkaboot.service
 
 import eom.improve.kafkaboot.model.FilmEntity
-import eom.improve.kafkaboot.repository.FilmRepository
-import eom.improve.kafkaboot.repository.InventoryRepository
-import eom.improve.kafkaboot.repository.PaymentRepository
-import eom.improve.kafkaboot.repository.RentalRepository
+import eom.improve.kafkaboot.repository.*
 import org.springframework.stereotype.Service
-import org.springframework.transaction.annotation.Transactional
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.kotlin.core.publisher.toMono
@@ -17,7 +13,9 @@ class FilmService(
     private val filmRepository : FilmRepository,
     private val paymentRepository : PaymentRepository,
     private val inventoryRepository : InventoryRepository,
-    private val rentalRepository : RentalRepository
+    private val rentalRepository : RentalRepository,
+    private val filmActorRepository: FilmActorRepository,
+    private val filmCategoryRepository: FilmCategoryRepository
 ) {
     fun findAll() : Flux<FilmEntity> = filmRepository.findAllBy()
 
@@ -31,7 +29,6 @@ class FilmService(
         return filmRepository.save(toBeSavedFilm);
     }
 
-    @Transactional
     fun deleteFilm(filmId : Int) : Mono<Void> {
         // need to implement cascade delete(maybe soft) for table data that set foreign key
         return filmRepository.findById(filmId)
@@ -52,13 +49,24 @@ class FilmService(
                              }.then(inventoryEn.toMono())
                      }
                      .flatMap { inventoryEn ->
-                         println(inventoryEn.inventoryId)
                          inventoryRepository.deleteByInventoryId(inventoryEn.inventoryId)
                      }
                      .then(filmEn.toMono())
             }
             .flatMap { filmEn ->
-                filmRepository.deleteByFilmId(filmEn.filmId)
+                Mono.zip(
+                    filmActorRepository.findAllByFilmId(filmEn.filmId)
+                        .flatMap { filmActorRepository.deleteByFilmId(it.filmId) }
+                        .then()
+                        ,
+                    filmCategoryRepository.findAllByFilmId(filmEn.filmId)
+                        .flatMap { filmCategoryRepository.deleteByFilmId(it.filmId) }
+                        .then()
+                )
+                    .then(filmEn.toMono())
+            }
+            .flatMap {
+                filmRepository.deleteByFilmId(it.filmId)
             }
     }
 }