package net.lightoze.jooq.postgresql.notify;

import com.google.common.util.concurrent.AbstractExecutionThreadService;
import org.postgresql.PGConnection;
import org.postgresql.PGNotification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

public abstract class NotificationListener extends AbstractExecutionThreadService {

    private final Logger log = LoggerFactory.getLogger(getClass());

    private String[] channels = new String[0];
    private long retryDelayMillis = 5000;
    private long fetchIntervalMillis = 500;

    public String[] getChannels() {
        return channels;
    }

    public void setChannels(String[] channels) {
        this.channels = channels;
    }

    public long getRetryDelayMillis() {
        return retryDelayMillis;
    }

    public void setRetryDelayMillis(long retryDelayMillis) {
        this.retryDelayMillis = retryDelayMillis;
    }

    public long getFetchIntervalMillis() {
        return fetchIntervalMillis;
    }

    public void setFetchIntervalMillis(long fetchIntervalMillis) {
        this.fetchIntervalMillis = fetchIntervalMillis;
    }

    /**
     * Obtain a new database connection. The connection should not have a timeout
     * because it will be kept open as long as new notifications are listened for.
     *
     * @return database connection
     * @throws SQLException when database connection failed
     */
    protected abstract Connection getConnection() throws SQLException;

    /**
     * Close database connection and make sure it is not reused by anyone else
     * (e.g. by connection pool) to prevent leaking notification subscriptions.
     *
     * @param connection database connection
     * @throws SQLException when error occurred
     */
    protected abstract void closeConnection(Connection connection) throws SQLException;

    /**
     * Process incoming notification.
     *
     * @param notification notification
     */
    protected abstract void receiveNotification(PGNotification notification);

    @Override
    protected void run() throws Exception {
        while (isRunning()) {
            Connection conn;
            try {
                conn = getConnection();
                conn.setAutoCommit(true);
            } catch (Throwable e) {
                log.error("Could not get a connection", e);
                Thread.sleep(retryDelayMillis);
                continue;
            }
            try {
                for (String channel : channels) {
                    try (Statement stmt = conn.createStatement()) {
                        stmt.execute("LISTEN " + channel);
                    }
                }
                while (isRunning()) {
                    try (Statement stmt = conn.createStatement()) {
                        stmt.execute("SELECT 1");
                    }

                    PGNotification[] notifications = conn.unwrap(PGConnection.class).getNotifications();
                    if (notifications == null || notifications.length == 0) {
                        Thread.sleep(fetchIntervalMillis);
                    } else {
                        for (PGNotification notification : notifications) {
                            try {
                                receiveNotification(notification);
                            } catch (Throwable e) {
                                log.error("Error processing notification {} ({}) from {}",
                                        notification.getName(), notification.getParameter(), notification.getPID(), e);
                            }
                        }
                    }
                }
            } catch (Throwable e) {
                log.error("Error while listening for notifications", e);
                Thread.sleep(retryDelayMillis);
            } finally {
                try {
                    closeConnection(conn);
                } catch (Throwable e) {
                    log.warn("Error closing connection", e);
                }
            }
        }
    }
}

@@ -50,72 +50,70 @@
 ort 
-com.google.common.util.concurrent.AbstractExecutionThreadService
+lombok.AccessLevel;%0Aimport lombok.Getter;%0Aimport lombok.Setter
 ;%0Aim
@@ -375,47 +375,8 @@
 ner 
-extends AbstractExecutionThreadService 
 %7B%0A%0A 
@@ -439,24 +439,36 @@
 tClass());%0A%0A
+    @Getter%0A
     private 
@@ -502,16 +502,40 @@
 ing%5B0%5D;%0A
+    @Getter%0A    @Setter%0A
     priv
@@ -576,611 +576,238 @@
     
-private long fetchIntervalMillis = 500;%0A%0A    public String%5B%5D getChannels() %7B%0A        return channels;%0A    %7D%0A%0A    public void setChannels(String%5B%5D channels) %7B%0A        this.channels = channels;%0A    %7D%0A%0A    public long getRetryDelayMillis() %7B%0A        return retryDelayMillis;%0A    %7D%0A%0A    public void setRetryDelayMillis(long retryDelayMillis) %7B%0A        this.retryDelayMillis = retryDelayMillis;%0A    %7D%0A%0A    public long getFetchIntervalMillis() %7B%0A        return fetchIntervalMillis;%0A    %7D%0A%0A    public void setFetchIntervalMillis(long fetchIntervalMillis) %7B%0A        this.fetchIntervalMillis = fetchIntervalMilli
+@Getter%0A    @Setter%0A    private int fetchTimeoutMillis = 1000;%0A    @Getter%0A    @Setter(AccessLevel.PROTECTED)%0A    private volatile boolean running;%0A%0A    public void setChannels(String... channels) %7B%0A        this.channels = channel
 s;%0A 
@@ -1715,22 +1715,8 @@
 );%0A%0A
-    @Override%0A
     
@@ -1739,25 +1739,8 @@
 un()
- throws Exception
  %7B%0A 
@@ -1859,50 +1859,8 @@
 ();%0A
-                conn.setAutoCommit(true);%0A
     
@@ -1943,32 +1943,58 @@
 onnection%22, e);%0A
+                try %7B%0A    
                 
@@ -2017,32 +2017,132 @@
 ryDelayMillis);%0A
+                %7D catch (InterruptedException ex) %7B%0A                    // ignore%0A                %7D%0A
                 
@@ -2175,32 +2175,74 @@
           try %7B%0A
+                conn.setAutoCommit(true);%0A
                 
@@ -2616,25 +2616,24 @@
           %7D%0A
-%0A
             
@@ -2722,16 +2722,34 @@
 cations(
+fetchTimeoutMillis
 );%0A     
@@ -2785,133 +2785,16 @@
 ons 
-=
+!
 = null
- %7C%7C notifications.length == 0) %7B%0A                        Thread.sleep(fetchIntervalMillis);%0A                    %7D else
+)
  %7B%0A 
@@ -3433,24 +3433,50 @@
 tions%22, e);%0A
+                try %7B%0A    
             
@@ -3503,32 +3503,132 @@
 ryDelayMillis);%0A
+                %7D catch (InterruptedException ex) %7B%0A                    // ignore%0A                %7D%0A
             %7D fi

Remove Guava dependency from NotificationListener