Background worker with Listen

["Ihnat Peter | TSS Group a.s." <Ihnat@xxxxxxxxxxx>]

I am trying to create background worker which listens to notifications and do some work after receiving one.

I got 2 problems:

-          Worker is receiving notifications from every channel not only the registered channel (in my case “foo”)

-          Notifications are not logged in the server log – I cannot store the payloads for further work

Any help is welcomed.

 

Here is the code:

PG_MODULE_MAGIC;

 

void _PG_init(void);

void _PG_fini(void);

 

static volatile sig_atomic_t got_sigterm = false;

static volatile sig_atomic_t got_sigusr1 = false;

static char *notify_database = NULL;

static emit_log_hook_type prev_log_hook = NULL;

                                                

static void

bgw_sigterm(SIGNAL_ARGS)

{

                int save_errno = errno;

                got_sigterm = true;

                if (MyProc)

                                SetLatch(&MyProc->procLatch);

                errno = save_errno;

}

 

static void

bgw_sigusr1(SIGNAL_ARGS)

{

                int save_errno = errno;

                got_sigusr1 = true;

                if (MyProc)

                                SetLatch(&MyProc->procLatch);

               errno = save_errno;

}     

 

static void

notify_main(Datum main_arg)

{

                pqsignal(SIGTERM, bgw_sigterm);

                pqsignal(SIGUSR1, bgw_sigusr1);

 

                BackgroundWorkerUnblockSignals();

                BackgroundWorkerInitializeConnection(notify_database, NULL);

                 EnableNotifyInterrupt();

 

                pgstat_report_activity(STATE_RUNNING, "background_worker");

                StartTransactionCommand();

                Async_Listen("foo");

                CommitTransactionCommand();

                pgstat_report_activity(STATE_IDLE, NULL);

  

                while (!got_sigterm)

                {

                                int           rc;

 

                                rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 10000L);

                                ResetLatch(&MyProc->procLatch);

 

                                if (rc & WL_POSTMASTER_DEATH)

                                                proc_exit(1);

      

                                if (got_sigusr1)

                                {

                                                got_sigusr1 = false;

                                                elog(INFO, " background_worker: notification received");

                                                // DO SOME WORK WITH STORED NOTIFICATIONS

                                }            

          

                }

 

                elog(LOG, "background_worker: finished");

                proc_exit(0);

}

 

static void

store_notification(ErrorData *edata)

{

                // HERE STORE THE NOTIFICATION FROM SERVER LOG

 

                if (prev_log_hook)

                                (*prev_log_hook) (edata);

}

 

void

_PG_init(void)

{

                BackgroundWorker worker;

                DefineCustomStringVariable("postgres", NULL, NULL, &notify_database,

                                           "postgres",

                                           PGC_POSTMASTER, 0, NULL, NULL, NULL);

 

                MemSet(&worker, 0, sizeof(BackgroundWorker));

                snprintf(worker.bgw_name, BGW_MAXLEN, "background_worker");

                worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;

                worker.bgw_start_time = BgWorkerStart_RecoveryFinished;

                worker.bgw_main = notify_main;

                worker.bgw_restart_time = 10;

                worker.bgw_main_arg = (Datum) 0;

                worker.bgw_notify_pid = 0;

                RegisterBackgroundWorker(&worker);

   

                prev_log_hook = emit_log_hook;

                emit_log_hook = store_notification;

}

 

void

_PG_fini(void)

{

                emit_log_hook = prev_log_hook;

}