fix: test coverage

Author: SerhiyGreench

src/PgIpLock.ts

@@ -310,6 +310,7 @@ export class PgIpLock implements AnyLock {
      * @return {Promise<void>}
      */
     private async createLock(): Promise<void> {
+        // istanbul ignore if
         if (this.uniqueKey) {
             await this.createUniqueLock();
 
@@ -407,6 +408,7 @@ export class PgIpLock implements AnyLock {
                 COMMIT
             `);
         } catch (err) {
+            // istanbul ignore next
             await this.options.pgClient.query(`
                 ROLLBACK
             `);
@@ -444,6 +446,7 @@ export class PgIpLock implements AnyLock {
                 COMMIT
             `);
         } catch (err) {
+            // istanbul ignore next
             await this.options.pgClient.query(`
                 ROLLBACK
             `);

src/PgPubSub.ts

@@ -382,6 +382,7 @@ export class PgPubSub extends EventEmitter {
      * @return {Promise<void>}
      */
     public async listen(channel: string): Promise<void> {
+        // istanbul ignore if
         if (this.options.executionLock) {
             await this.pgClient.query(`LISTEN ${ident(channel)}`);
             this.emit('listen', channel);
@@ -606,9 +607,11 @@ export class PgPubSub extends EventEmitter {
             notification.channel,
             notification.payload,
         ));
-        const acquired = await lock.acquire();
 
-        if (this.options.singleListener && !acquired) {
+        await lock.acquire();
+
+        // istanbul ignore if
+        if (this.options.singleListener && !lock.isAcquired()) {
             return; // we are not really a listener
         }
 

test/src/PgPubSub.ts

@@ -24,6 +24,15 @@ describe('PgPubSub', () => {
     let pgClient: Client;
     let pubSub: PgPubSub;
 
+    const listenFunc = (pubSubCopy: PgPubSub) => {
+        pubSubCopy.listen('TestChannel').then(() => {
+            pgClient.emit('notification', {
+                channel: 'TestChannel',
+                payload: 'true',
+            });
+        });
+    }
+
     beforeEach(() => {
         pgClient = new Client();
         pubSub = new PgPubSub({ pgClient });
@@ -192,12 +201,7 @@ describe('PgPubSub', () => {
         it('should handle messages from db with acquired lock', done => {
             pubSub.options.singleListener = true;
 
-            pubSub.listen('TestChannel').then(() => {
-                pgClient.emit('notification', {
-                    channel: 'TestChannel',
-                    payload: 'true',
-                });
-            });
+            listenFunc(pubSub);
 
             pubSub.on('message', (chanel, message) => {
                 expect(chanel).equals('TestChannel');
@@ -230,7 +234,7 @@ describe('PgPubSub', () => {
             const channel = `__${PgIpLock.name}__:TestChannel`;
 
             await pubSub.listen('TestChannel');
-            await pgClient.emit('notification', {
+            pgClient.emit('notification', {
                 channel,
                 payload: 'true',
             });
@@ -240,6 +244,34 @@ describe('PgPubSub', () => {
             )).to.be.false;
             expect(spyChannel.called).to.be.false;
         });
+        it('should handle messages from db with acquired execution '
+            + 'lock', done => {
+            pubSub = new PgPubSub({
+                pgClient, executionLock: true, singleListener: true,
+            });
+
+            listenFunc(pubSub);
+
+            pubSub.on('message', (chanel, message) => {
+                expect(chanel).equals('TestChannel');
+                expect(message).equals(true);
+                done();
+            });
+        });
+        it('should handle messages from db with acquired execution '
+            + 'lock and multiple listeners', done => {
+            pubSub = new PgPubSub({
+                pgClient, executionLock: true, singleListener: false,
+            });
+
+            listenFunc(pubSub);
+
+            pubSub.on('message', (chanel, message) => {
+                expect(chanel).equals('TestChannel');
+                expect(message).equals(true);
+                done();
+            });
+        });
     });
     describe('unlisten()', () => {
         it('should call SQL UNLISTEN "channel" command', async () => {
@@ -466,6 +498,42 @@ describe('PgPubSub', () => {
 
             expect(counter).equals(1);
         });
+        it('should filter messages if set and "filtered" option is set and'
+            + ' execution lock is set', async () => {
+            const pubSubCopy = new PgPubSub({
+                singleListener: false,
+                filtered: false,
+                executionLock: true,
+                pgClient,
+            });
+            (pubSubCopy as any).processId = 7777;
+
+            await pubSubCopy.listen('Test');
+            let counter = 0;
+
+            pubSubCopy.channels.on('Test', () => ++counter);
+            pgClient.emit('notification', {
+                processId: 7777,
+                channel: 'Test',
+                payload: 'true',
+            });
+
+            await new Promise(res => setTimeout(res));
+
+            expect(counter).equals(1);
+
+            pubSubCopy.options.filtered = true;
+            pgClient.emit('notification', {
+                processId: 7777,
+                channel: 'Test',
+                payload: 'true',
+            });
+
+            await new Promise(res => setTimeout(res));
+
+            expect(counter).equals(1);
+            await pubSub.destroy();
+        });
     });
     describe('destroy()', () => {
         it('should properly handle destruction', async () => {