source service impl

cqrs-source/src/main/java/com/amaljoyc/cqrs/SourceApplication.java

@@ -2,8 +2,11 @@ package com.amaljoyc.cqrs;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Source;
 
 @SpringBootApplication
+@EnableBinding(Source.class)
 public class SourceApplication {
 
 	public static void main(String[] args) {

cqrs-source/src/main/java/com/amaljoyc/cqrs/api/BankAccountController.java

@@ -0,0 +1,30 @@
+package com.amaljoyc.cqrs.api;
+
+import com.amaljoyc.cqrs.application.BankAccountService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+@RestController("/bankAccount")
+public class BankAccountController {
+
+    @Autowired
+    private BankAccountService bankAccountService;
+
+    @PostMapping("/credit")
+    ResponseEntity credit(@RequestBody CreditCommand creditCommand) {
+        bankAccountService.creditAmount(creditCommand.getAccountNumber(), creditCommand.getAmount());
+        return ResponseEntity.ok().build();
+    }
+
+    @PostMapping("/debit")
+    ResponseEntity debit(@RequestBody DebitCommand debitCommand) {
+        bankAccountService.debitAmount(debitCommand.getAccountNumber(), debitCommand.getAmount());
+        return ResponseEntity.ok().build();
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/api/CreditCommand.java

@@ -0,0 +1,16 @@
+package com.amaljoyc.cqrs.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class CreditCommand {
+    private String accountNumber;
+    private Long amount;
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/api/DebitCommand.java

@@ -0,0 +1,16 @@
+package com.amaljoyc.cqrs.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class DebitCommand {
+    private String accountNumber;
+    private Long amount;
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/application/BankAccountService.java

@@ -0,0 +1,42 @@
+package com.amaljoyc.cqrs.application;
+
+import com.amaljoyc.cqrs.event.AmountCredited;
+import com.amaljoyc.cqrs.event.AmountDebited;
+import com.amaljoyc.cqrs.event.DomainEventPublisher;
+import com.amaljoyc.cqrs.model.BankAccount;
+import com.amaljoyc.cqrs.persistence.BankAccountRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.transaction.Transactional;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+@Service
+public class BankAccountService {
+
+    @Autowired
+    private BankAccountRepository bankAccountRepository;
+
+    @Autowired
+    private DomainEventPublisher domainEventPublisher;
+
+    @Transactional
+    public void creditAmount(String accountNumber, Long amount) {
+        BankAccount bankAccount = bankAccountRepository.findByAccountNumber(accountNumber)
+                .orElseThrow(() -> new IllegalArgumentException("Invalid accountNumber :" + accountNumber));
+
+        bankAccount.credit(amount);
+        domainEventPublisher.publish(new AmountCredited(accountNumber, amount));
+    }
+
+    @Transactional
+    public void debitAmount(String accountNumber, Long amount) {
+        BankAccount bankAccount = bankAccountRepository.findByAccountNumber(accountNumber)
+                .orElseThrow(() -> new IllegalArgumentException("Invalid accountNumber :" + accountNumber));
+
+        bankAccount.debit(amount);
+        domainEventPublisher.publish(new AmountDebited(accountNumber, amount));
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/AccountProcessType.java

@@ -0,0 +1,10 @@
+package com.amaljoyc.cqrs.event;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+public enum AccountProcessType {
+
+    CREDIT,
+    DEBIT;
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/AmountCredited.java

@@ -0,0 +1,31 @@
+package com.amaljoyc.cqrs.event;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+@Data
+@NoArgsConstructor
+public class AmountCredited implements DomainEvent {
+
+    private String accountNumber;
+    private AccountProcessType type;
+    private Long amount;
+    private Instant timestamp;
+
+    public AmountCredited(String accountNumber, Long amount) {
+        this.accountNumber = accountNumber;
+        this.type = AccountProcessType.CREDIT;
+        this.amount = amount;
+        this.timestamp = Instant.now();
+    }
+
+    @Override
+    public String getType() {
+        return this.getClass().getName();
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/AmountDebited.java

@@ -0,0 +1,33 @@
+package com.amaljoyc.cqrs.event;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class AmountDebited implements DomainEvent {
+
+    private String accountNumber;
+    private AccountProcessType type;
+    private Long amount;
+    private Instant timestamp;
+
+    public AmountDebited(String accountNumber, Long amount) {
+        this.accountNumber = accountNumber;
+        this.type = AccountProcessType.DEBIT;
+        this.amount = amount;
+        this.timestamp = Instant.now();
+    }
+
+    @Override
+    public String getType() {
+        return this.getClass().getName();
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/DomainEvent.java

@@ -0,0 +1,9 @@
+package com.amaljoyc.cqrs.event;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+public interface DomainEvent {
+
+    String getType();
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/DomainEventPublisher.java

@@ -0,0 +1,9 @@
+package com.amaljoyc.cqrs.event;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+public interface DomainEventPublisher {
+
+    void publish(DomainEvent event);
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/event/KafkaDomainEventPublisher.java

@@ -0,0 +1,37 @@
+package com.amaljoyc.cqrs.event;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Created by amaljoyc on 18.10.18.
+ */
+@Component
+public class KafkaDomainEventPublisher implements DomainEventPublisher {
+
+    @Autowired
+    private Source source;
+
+    @Autowired
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void publish(DomainEvent event) {
+        try {
+            Map<String, Object> headers = new HashMap<>();
+            headers.put("type", event.getType());
+            String content = objectMapper.writeValueAsString(event);
+
+            source.output().send(new GenericMessage<>(content, headers));
+        } catch (JsonProcessingException e) {
+            e.printStackTrace();
+        }
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/model/BankAccount.java

@@ -0,0 +1,36 @@
+package com.amaljoyc.cqrs.model;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.Id;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+@Entity
+@NoArgsConstructor
+public class BankAccount {
+
+    @Id
+    @GeneratedValue
+    @Getter
+    private Long id;
+
+    private String accountNumber;
+    private Long balance;
+
+    public void credit(Long amount) {
+        balance += amount;
+    }
+
+    public void debit(Long amount) {
+        if (amount > balance) {
+            throw new IllegalArgumentException("Not enough balance available for debit");
+        }
+
+        balance -= amount;
+    }
+}

cqrs-source/src/main/java/com/amaljoyc/cqrs/persistence/BankAccountRepository.java

@@ -0,0 +1,15 @@
+package com.amaljoyc.cqrs.persistence;
+
+import com.amaljoyc.cqrs.model.BankAccount;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.Optional;
+
+/**
+ * Created by amaljoyc on 17.10.18.
+ */
+public interface BankAccountRepository extends JpaRepository<BankAccount, Long> {
+
+    Optional<BankAccount> findByAccountNumber(String accountNumber);
+
+}