LISTEN/NOTIFY pgconnection va giù java?

Question

Utilizzo il DB PostgreSQL e l'applicazione della sua funzionalità LISTEN/NOTIFY. Quindi il mio ascoltatore è al mio AS (Application Server) e ho trigger configurati sul mio DB in modo tale che quando le operazioni CRUD vengono eseguite su un tavolo una richiesta NOTIFY sia inviata su AS.

ASCOLTATORE classe in Java:

 @Singleton 
     @Startup 
    NotificationListenerInterface.class) 
     public class NotificationListener extends Thread implements NotificationListenerInterface { 

      @Resource(mappedName="java:/RESOURCES") 
      private DataSource ds; 

      @PersistenceContext(unitName = "one") 
      EntityManager em; 

      Logger logger = Logger.getLogger(NotificationListener.class); 

      private Connection Conn; 
      private PGConnection pgConnection = null; 
      private NotifyRequest notifyRequest = null; 

      @PostConstruct 
      public void notificationListener() throws Throwable { 

       System.out.println("Notification****************"); 
       try 
       { 


        Class.forName("com.impossibl.postgres.jdbc.PGDriver"); 
        String url = "jdbc:pgsql://192.xx.xx.126:5432/postgres"; 


        Conn = DriverManager.getConnection(url,"postgres","password"); 
        this.pgConnection = (PGConnection) Conn; 

        System.out.println("PG CONNECTON: "+ pgConnection); 
        Statement listenStatement = Conn.createStatement(); 
        listenStatement.execute("LISTEN notify_channel"); 
        listenStatement.close(); 

        pgConnection.addNotificationListener(new PGNotificationListener() { 

         @Override 
         public void notification(int processId, String channelName, String payload){ 

          System.out.println("*********INSIDE NOTIFICATION*************"); 

          System.out.println("Payload: " + jsonPayload); 

} 

Così come il mio AS è in su, ho configurato che all'avvio della classe ascoltatore è chiamato (@Startup annotation) ed è iniziare l'ascolto sul canale.

Ora funziona correttamente se, come per esempio, per il test, modifica manualmente la mia tabella in DB, la notifica viene generata e LISTENER lo riceve.

Tuttavia, quando invio una richiesta UPDATE sulla tabella, UPADTE viene eseguito correttamente ma LISTENER non riceve nulla.

Sento che la mia connessione di LISTENER si interrompe quando invio una richiesta (crea anche una connessione per modificare entità), ma non ne sono sicuro. Leggo delle connessioni permanenti e delle connessioni raggruppate, ma non sono in grado di decidere come perseguirlo.

Sto utilizzando il contenitore pgjdbc (http://impossibl.github.io/pgjdbc-ng/) per le notifiche asincrone poiché la connessione jdbc richiede il polling.

EDIT:

Quando provo l'ascoltatore di cui sopra con il polling utilizzando il vaso JDBC standard (non pgjdbc), ottengo le notifiche.

faccio PGNotification notif[] = con.getNotifications() e ottengo le notifiche, ma farlo in modo asincrono come qui di seguito non ho ricevuto notifiche.

pgConnection.addNotificationListener(new PGNotificationListener() { 

     @Override 
     public void notification(int processId, String channelName, String payload){ 

      System.out.println("*********INSIDE NOTIFICATION*************"); 
     } 

RISOLTO:

mio ascoltatore stava andando fuori del campo di applicazione dopo l'esecuzione funzione è stata completata come il mio ascoltatore aveva lo scopo di funzione. Quindi l'ho mantenuto in una variabile membro della mia classe bean di avvio e poi ha funzionato.

Answer 1

Gli ascoltatori di notifiche sono internamente gestiti da quella libreria come riferimenti deboli, il che significa che è necessario tenere un riferimento fisso esternamente in modo che non vengano raccolti. Controlla le linee di classe BasicContext 642 - 655:

public void addNotificationListener(String name, String channelNameFilter, NotificationListener listener) { 

    name = nullToEmpty(name); 
    channelNameFilter = channelNameFilter != null ? channelNameFilter : ".*"; 

    Pattern channelNameFilterPattern = Pattern.compile(channelNameFilter); 

    NotificationKey key = new NotificationKey(name, channelNameFilterPattern); 

    synchronized (notificationListeners) { 
     notificationListeners.put(key, new WeakReference<NotificationListener>(listener)); 
    } 

} 

Se il GC prende il vostro ascoltatore, chiamate a "ottenere" il riferimento debole restituirà nulla e non si attiverà come si è visto dalle linee 690-710

@Override 
    public synchronized void reportNotification(int processId, String channelName, String payload) { 

    Iterator<Map.Entry<NotificationKey, WeakReference<NotificationListener>>> iter = notificationListeners.entrySet().iterator(); 
    while (iter.hasNext()) { 

     Map.Entry<NotificationKey, WeakReference<NotificationListener>> entry = iter.next(); 

     NotificationListener listener = entry.getValue().get(); 
     if (listener == null) { 

     iter.remove(); 
     } 
     else if (entry.getKey().channelNameFilter.matcher(channelName).matches()) { 

     listener.notification(processId, channelName, payload); 
     } 

    } 

} 

per risolvere questo problema, aggiungere i tuoi ascoltatori di notifica in quanto tale:

/// Do not let this reference go out of scope! 
PGNotificationListener listener = new PGNotificationListener() { 

@Override 
public void notification(int processId, String channelName, String payload) { 
    // interesting code 
}; 
pgConnection.addNotificationListener(listener); 

Piuttosto un caso d'uso dispari per i riferimenti deboli a mio parere ...