diff --git a/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts b/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts index 51734e0f..496147b2 100644 --- a/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts +++ b/packages/gcp-pubsub/lib/pubsub/AbstractPubSubPublisher.ts @@ -173,4 +173,11 @@ export abstract class AbstractPubSubPublisher protected override resolveSchema(message: MessagePayloadType) { return this.messageSchemaContainer.resolveSchema(message) } + + override async close(): Promise { + // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt + // that began before the close. Matches AbstractAmqpPublisher. + this.initPromise = undefined + await super.close() + } } diff --git a/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts b/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts index 0e5cad8c..e71ed911 100644 --- a/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts +++ b/packages/gcp-pubsub/test/publishers/PubSubPermissionPublisher.spec.ts @@ -81,6 +81,30 @@ describe('PubSubPermissionPublisher', () => { await newPublisher.close() }) + it('starts a fresh lazy init when publishing after close', async () => { + const newPublisher = new PubSubPermissionPublisher(diContainer.cradle) + // The first attempt never settles, as if it were still in flight when close() ran + const initSpy = vi + .spyOn(newPublisher, 'init') + .mockImplementationOnce(() => new Promise(() => {})) + const message = (id: string) => ({ + id, + messageType: 'add' as const, + timestamp: new Date().toISOString(), + userIds: ['userId1'], + }) + + void newPublisher.publish(message('close-reinit-1')) + await newPublisher.close() + await newPublisher.publish(message('close-reinit-2')) + + const spy = await newPublisher.handlerSpy.waitForMessageWithId('close-reinit-2', 'published') + expect(spy.processingResult).toEqual({ status: 'published' }) + expect(initSpy).toHaveBeenCalledTimes(2) + + await newPublisher.close() + }) + it('publishes a message to topic', async () => { const message = { id: '1', diff --git a/packages/sns/lib/sns/AbstractSnsPublisher.ts b/packages/sns/lib/sns/AbstractSnsPublisher.ts index 8b666c16..d9da77ee 100644 --- a/packages/sns/lib/sns/AbstractSnsPublisher.ts +++ b/packages/sns/lib/sns/AbstractSnsPublisher.ts @@ -301,4 +301,11 @@ export abstract class AbstractSnsPublisher return resolvedOptions } /* c8 ignore stop */ + + override async close(): Promise { + // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt + // that began before the close. Matches AbstractAmqpPublisher. + this.initPromise = undefined + await super.close() + } } diff --git a/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts b/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts index 567001da..60bbb1af 100644 --- a/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts +++ b/packages/sns/test/publishers/SnsPermissionPublisher.spec.ts @@ -408,5 +408,23 @@ describe('SnsPermissionPublisher', () => { await newPublisher.close() }) + + it('starts a fresh lazy init when publishing after close', async () => { + const newPublisher = new SnsPermissionPublisher(diContainer.cradle) + // The first attempt never settles, as if it were still in flight when close() ran + const initSpy = vi + .spyOn(newPublisher, 'init') + .mockImplementationOnce(() => new Promise(() => {})) + + void newPublisher.publish({ id: '1', messageType: 'add' }) + await newPublisher.close() + await newPublisher.publish({ id: '2', messageType: 'add' }) + + const spy = await newPublisher.handlerSpy.waitForMessageWithId('2', 'published') + expect(spy.processingResult).toEqual({ status: 'published' }) + expect(initSpy).toHaveBeenCalledTimes(2) + + await newPublisher.close() + }) }) }) diff --git a/packages/sqs/lib/sqs/AbstractSqsPublisher.ts b/packages/sqs/lib/sqs/AbstractSqsPublisher.ts index 959d3819..aa32034e 100644 --- a/packages/sqs/lib/sqs/AbstractSqsPublisher.ts +++ b/packages/sqs/lib/sqs/AbstractSqsPublisher.ts @@ -290,4 +290,11 @@ export abstract class AbstractSqsPublisher return resolvedOptions } + + override async close(): Promise { + // Dropped so a publish after close() starts a fresh lazy init, rather than awaiting an attempt + // that began before the close. Matches AbstractAmqpPublisher. + this.initPromise = undefined + await super.close() + } } diff --git a/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts b/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts index 69464e17..abe73c20 100644 --- a/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts +++ b/packages/sqs/test/publishers/SqsPermissionPublisher.spec.ts @@ -567,5 +567,23 @@ describe('SqsPermissionPublisher', () => { await newPublisher.close() }) + + it('starts a fresh lazy init when publishing after close', async () => { + const newPublisher = new SqsPermissionPublisher(diContainer.cradle) + // The first attempt never settles, as if it were still in flight when close() ran + const initSpy = vi + .spyOn(newPublisher, 'init') + .mockImplementationOnce(() => new Promise(() => {})) + + void newPublisher.publish({ id: '1', messageType: 'add' }) + await newPublisher.close() + await newPublisher.publish({ id: '2', messageType: 'add' }) + + const spy = await newPublisher.handlerSpy.waitForMessageWithId('2', 'published') + expect(spy.processingResult).toEqual({ status: 'published' }) + expect(initSpy).toHaveBeenCalledTimes(2) + + await newPublisher.close() + }) }) })