[BAEL-6275] PostgreSQL NOTIFY/LISTEN (#14153)

* [BAEL-4849] Article code

* [BAEL-4968] Article code

* [BAEL-4968] Article code

* [BAEL-4968] Article code

* [BAEL-4968] Remove extra comments

* [BAEL-5258] Article Code

* [BAEL-2765] PKCE Support for Secret Clients

* [BAEL-5698] Article code

* [BAEL-5698] Article code

* [BAEL-5905] Initial code

* [BAEL-5905] Article code

* [BAEL-5905] Relocate article code to new module

* [BAEL-6275] PostgreSQL NOTIFY/LISTEN

* [BAEL-6275] Minor correction

---------

Co-authored-by: Philippe Sevestre <psevestre@gmail.com>

messaging-modules/pom.xml

@@ -22,6 +22,7 @@
         <module>spring-amqp</module>
         <module>spring-apache-camel</module>
         <module>spring-jms</module>
+		<module>postgres-notify</module>
     </modules>
 
 </project>

messaging-modules/postgres-notify/.gitignore

@@ -0,0 +1 @@
+/application-local.properties

messaging-modules/postgres-notify/pom.xml

@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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>
+    <artifactId>postgres-notify</artifactId>
+    <name>postgres-notify</name>
+    <description>PostgreSQL as a Message Broker</description>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>messaging-modules</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-jdbc</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-devtools</artifactId>
+            <scope>runtime</scope>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+        
+    </dependencies>
+
+
+	<properties>
+		<java.version>1.8</java.version>
+	</properties>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+	
+	<profiles>
+		<profile>
+			<id>instance1</id>
+			<build>
+				<plugins>
+					<plugin>
+						<groupId>org.springframework.boot</groupId>
+						<artifactId>spring-boot-maven-plugin</artifactId>
+						<configuration>
+							<jvmArguments>-Dserver.port=8081</jvmArguments>						
+						</configuration>						
+					</plugin>
+				</plugins>
+			</build>
+		</profile>
+		
+	</profiles>
+</project>

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/Application.java

@@ -0,0 +1,12 @@
+package com.baeldung.messaging.postgresql;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class Application {
+
+    public static void main(String[] args) {
+        SpringApplication.run(Application.class, args);        
+    }
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/config/CacheConfiguration.java

@@ -0,0 +1,34 @@
+package com.baeldung.messaging.postgresql.config;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.concurrent.ConcurrentMapCache;
+import org.springframework.cache.support.SimpleCacheManager;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.util.ConcurrentLruCache;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+
+@Configuration
+public class CacheConfiguration {
+    
+    @Bean
+    Cache ordersCache(CacheManager cm) {
+        return cm.getCache("orders");
+    }
+    
+    @Bean
+    @ConditionalOnMissingBean
+    CacheManager defaultCacheManager() {
+        SimpleCacheManager cm = new SimpleCacheManager();
+        Cache cache = new ConcurrentMapCache("orders",false);
+        cm.setCaches(Arrays.asList(cache));
+        
+        return cm;
+    }
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/config/ListenerConfiguration.java

@@ -0,0 +1,25 @@
+package com.baeldung.messaging.postgresql.config;
+
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.baeldung.messaging.postgresql.service.NotifierService;
+import com.baeldung.messaging.postgresql.service.NotificationHandler;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Configuration
+@Slf4j
+public class ListenerConfiguration {
+    
+    @Bean
+    CommandLineRunner startListener(NotifierService notifier, NotificationHandler handler) {
+        return (args) -> {
+            log.info("Starting order listener thread...");            
+            Runnable listener = notifier.createNotificationHandler(handler);            
+            Thread t = new Thread(listener, "order-listener");
+            t.start();
+        };
+    }
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/config/NotifierConfiguration.java

@@ -0,0 +1,30 @@
+package com.baeldung.messaging.postgresql.config;
+
+import java.util.Properties;
+
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import com.baeldung.messaging.postgresql.service.NotifierService;
+import com.zaxxer.hikari.util.DriverDataSource;
+
+@Configuration
+public class NotifierConfiguration {
+
+    @Bean
+    NotifierService notifier(DataSourceProperties props) {
+        
+        DriverDataSource ds = new DriverDataSource(
+           props.determineUrl(), 
+           props.determineDriverClassName(),
+           new Properties(), 
+           props.determineUsername(),
+           props.determinePassword());
+        
+        JdbcTemplate tpl = new JdbcTemplate(ds);
+
+        return new NotifierService(tpl);
+    }
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/controller/OrdersController.java

@@ -0,0 +1,53 @@
+package com.baeldung.messaging.postgresql.controller;
+
+import java.math.BigDecimal;
+import java.util.Optional;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+import com.baeldung.messaging.postgresql.domain.OrderType;
+import com.baeldung.messaging.postgresql.service.OrdersService;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@RestController
+@RequiredArgsConstructor
+@Slf4j
+public class OrdersController {
+    
+    private final OrdersService orders;
+
+    @PostMapping("/orders/sell")
+    public ResponseEntity<Order> postSellOrder(String symbol, BigDecimal quantity, BigDecimal price) {
+        log.info("postSellOrder: symbol={},quantity={},price={}", symbol,quantity,price);
+        Order order = orders.createOrder(OrderType.SELL, symbol, quantity, price);
+        return ResponseEntity.status(HttpStatus.CREATED).body(order);
+    }
+
+    @PostMapping("/orders/buy")
+    public ResponseEntity<Order> postBuyOrder(String symbol, BigDecimal quantity, BigDecimal price) {
+        log.info("postBuyOrder: symbol={},quantity={},price={}", symbol,quantity,price);
+        Order order = orders.createOrder(OrderType.BUY, symbol, quantity, price);
+        return ResponseEntity.status(HttpStatus.CREATED).body(order);
+    }
+    
+    @GetMapping("/orders/{id}")
+    public ResponseEntity<Order> getOrderById(@PathVariable Long id) {
+        
+        Optional<Order> o = orders.findById(id);
+        if (o.isEmpty()) {
+            return ResponseEntity.notFound().build();
+        }
+        
+        return ResponseEntity.ok(o.get());
+        
+    }
+ 
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/domain/Order.java

@@ -0,0 +1,21 @@
+package com.baeldung.messaging.postgresql.domain;
+
+import java.math.BigDecimal;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Table;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString
+@Table(name = "orders")
+public class Order {
+    @Id
+    private Long id;
+    private String symbol;
+    private OrderType orderType;
+    private BigDecimal price;
+    private BigDecimal quantity;
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/domain/OrderType.java

@@ -0,0 +1,12 @@
+package com.baeldung.messaging.postgresql.domain;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+@Getter
+public enum OrderType {    
+    BUY('B'),
+    SELL('S');
+    private final char c;
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/repository/OrdersRepository.java

@@ -0,0 +1,9 @@
+package com.baeldung.messaging.postgresql.repository;
+
+import org.springframework.data.repository.CrudRepository;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+
+public interface OrdersRepository extends CrudRepository<Order, Long>{
+
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/service/NotificationHandler.java

@@ -0,0 +1,31 @@
+package com.baeldung.messaging.postgresql.service;
+
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import org.postgresql.PGNotification;
+import org.springframework.cache.Cache;
+import org.springframework.stereotype.Component;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class NotificationHandler implements Consumer<PGNotification>{
+    
+    private final OrdersService orders;
+
+    @Override
+    public void accept(PGNotification t) {
+        log.info("Notification received: pid={}, name={}, param={}",t.getPID(),t.getName(),t.getParameter());
+        Optional<Order> order = orders.findById(Long.valueOf(t.getParameter()));
+        if ( !order.isEmpty()) {
+            log.info("order details: {}", order.get());
+        }
+    }
+
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/service/NotifierService.java

@@ -0,0 +1,55 @@
+package com.baeldung.messaging.postgresql.service;
+
+import java.sql.Connection;
+import java.util.function.Consumer;
+
+import org.postgresql.PGConnection;
+import org.postgresql.PGNotification;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RequiredArgsConstructor
+public class NotifierService {
+    private static final String ORDERS_CHANNEL = "orders";
+    private final JdbcTemplate tpl;
+    
+   
+    @Transactional
+    public void notifyOrderCreated(Order order) {
+        tpl.execute("NOTIFY " + ORDERS_CHANNEL + ", '" + order.getId() + "'");
+    }
+    
+    public Runnable createNotificationHandler(Consumer<PGNotification> consumer) {
+        
+        return () -> {
+            tpl.execute((Connection c) -> {
+                log.info("notificationHandler: sending LISTEN command...");
+                c.createStatement().execute("LISTEN " + ORDERS_CHANNEL);
+                
+                PGConnection pgconn = c.unwrap(PGConnection.class); 
+                
+                while(!Thread.currentThread().isInterrupted()) {
+                    PGNotification[] nts = pgconn.getNotifications(10000);
+                    if ( nts == null || nts.length == 0 ) {
+                        continue;
+                    }
+                    
+                    for( PGNotification nt : nts) {
+                        consumer.accept(nt);
+                    }
+                }
+                
+                return 0;
+            });
+                
+        };
+    }
+}

messaging-modules/postgres-notify/src/main/java/com/baeldung/messaging/postgresql/service/OrdersService.java

@@ -0,0 +1,61 @@
+package com.baeldung.messaging.postgresql.service;
+
+import java.math.BigDecimal;
+import java.util.Optional;
+
+import javax.sql.DataSource;
+
+import org.springframework.cache.Cache;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+import com.baeldung.messaging.postgresql.domain.OrderType;
+import com.baeldung.messaging.postgresql.repository.OrdersRepository;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class OrdersService {
+    private final OrdersRepository repo;
+    private final NotifierService notifier;
+    private final Cache ordersCache;
+    
+    @Transactional
+    public Order createOrder(OrderType orderType, String symbol, BigDecimal quantity, BigDecimal price) {
+        
+        Order order = new Order();
+        order.setOrderType(orderType);
+        order.setSymbol(symbol);
+        order.setQuantity(quantity);
+        order.setPrice(price);
+        order = repo.save(order);
+        
+        notifier.notifyOrderCreated(order);
+        
+        return order;
+
+    }
+
+    @Transactional(readOnly = true)
+    public Optional<Order> findById(Long id) {
+        Optional<Order> o = Optional.ofNullable(ordersCache.get(id, Order.class));
+        if ( !o.isEmpty() ) {
+            log.info("findById: cache hit, id={}",id);
+            return o;
+        }
+        
+        log.info("findById: cache miss, id={}",id);
+        o = repo.findById(id);
+        if ( o.isEmpty()) {
+            return o;
+        }
+        
+        ordersCache.put(id, o.get());
+        return o;
+    }
+
+}

messaging-modules/postgres-notify/src/main/resources/application.properties

@@ -0,0 +1,5 @@
+# Replace the properties below with proper values for your environment
+spring.sql.init.mode=ALWAYS
+#spring.datasource.url=jdbc:postgresql://some-postgresql-host/some-postgresql-database
+#spring.datasource.username=your-postgresql-username
+#spring.datasource.password=your-postgresql-password

messaging-modules/postgres-notify/src/main/resources/schema.sql

@@ -0,0 +1,7 @@
+create table if not exists orders (
+	id serial primary key,
+	symbol varchar(16) not null,
+	order_type varchar(8) not null,
+	price NUMERIC(10,2) not null,
+	quantity NUMERIC(10,2) not null
+);

messaging-modules/postgres-notify/src/test/java/com/baeldung/messaging/postgresql/ApplicationLiveTest.java

@@ -0,0 +1,47 @@
+package com.baeldung.messaging.postgresql;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.boot.test.web.client.TestRestTemplate;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.jdbc.Sql;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+import com.baeldung.messaging.postgresql.domain.Order;
+
+@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
+@Sql("/schema.sql")
+class ApplicationLiveTest  {
+    
+    
+    @LocalServerPort
+    int localPort;
+    
+    @Autowired
+    TestRestTemplate client;
+    
+    @Test
+    void whenCreateBuyOrder_thenSuccess() {
+        
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
+
+        MultiValueMap<String, String> data= new LinkedMultiValueMap<>();
+        data.add("symbol", "BAEL");
+        data.add("price", "14.56");
+        data.add("quantity", "100");
+        HttpEntity<MultiValueMap<String, String>> request = new HttpEntity<MultiValueMap<String, String>>(data, headers);
+        
+        client.postForEntity("http://localhost:" + localPort + "/orders/buy", data, Order.class);
+        
+    }
+
+}

messaging-modules/postgres-notify/src/test/resources/application.properties

@@ -0,0 +1,3 @@
+spring.datasource.url=jdbc:postgresql://localhost:5432/baeldung
+spring.datasource.username=baeldung
+spring.datasource.password=SqD64PtsGhDXjn9f