diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json index f4e4c93ea68e..0ac4fc9adcc7 100644 --- a/handwritten/pubsub/package.json +++ b/handwritten/pubsub/package.json @@ -72,6 +72,7 @@ "devDependencies": { "@grpc/proto-loader": "^0.8.0", "@opentelemetry/sdk-trace-base": "^2.8.0", + "@opentelemetry/sdk-trace-node": "^2.8.0", "@types/duplexify": "^3.6.4", "@types/extend": "^3.0.4", "@types/lodash.snakecase": "^4.1.9", @@ -82,6 +83,7 @@ "@types/proxyquire": "^1.3.31", "@types/sinon": "^21.0.0", "@types/tmp": "^0.2.6", + "avro-js": "^1.12.1", "c8": "^10.1.3", "codecov": "^3.8.3", "execa": "~5.1.0", diff --git a/handwritten/pubsub/src/telemetry-tracing.ts b/handwritten/pubsub/src/telemetry-tracing.ts index 54a516d1b6b4..77bc15883487 100644 --- a/handwritten/pubsub/src/telemetry-tracing.ts +++ b/handwritten/pubsub/src/telemetry-tracing.ts @@ -756,9 +756,7 @@ export function injectSpan(span: Span, message: MessageWithAttributes): void { return; } - if (!message.attributes) { - message.attributes = {}; - } + message.attributes = Object.assign({}, message.attributes); if (message.attributes[modernAttributeName]) { console.warn( diff --git a/handwritten/pubsub/system-test/avro-js.d.ts b/handwritten/pubsub/system-test/avro-js.d.ts new file mode 100644 index 000000000000..20937ee94b06 --- /dev/null +++ b/handwritten/pubsub/system-test/avro-js.d.ts @@ -0,0 +1,25 @@ +// Copyright 2021 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// This one doesn't seem to have typings. +declare module 'avro-js' { + function parse(def: string): Parser; + + class Parser { + fromBuffer(buf: Buffer): T; + fromString(str: string): T; + toBuffer(item: T): Buffer; + toString(item: T): string; + } +} diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts new file mode 100644 index 000000000000..49cfadca3881 --- /dev/null +++ b/handwritten/pubsub/system-test/avro-samples.test.ts @@ -0,0 +1,146 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {Message, PubSub, Schema} from '../src'; +import * as assert from 'assert'; +import {describe, it, after, before} from 'mocha'; +import {TestResources} from './testResources'; +import * as avro from 'avro-js'; +import * as fs from 'fs'; +import {waitForMessage} from './common'; + +describe('Avro Samples System Tests', () => { + const pubsub = new PubSub(); + const resources = new TestResources('ps-sys-avro'); + + let schemaId: string; + + before(async () => { + schemaId = resources.generateName('schema'); + + const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString(); + await pubsub.createSchema(schemaId, 'AVRO', definition); + }); + + after(async () => { + const [subscriptions] = await pubsub.getSubscriptions(); + await Promise.all( + resources.filterForCleanup(subscriptions).map(x => x.delete?.()) + ); + + const [topics] = await pubsub.getTopics(); + await Promise.all( + resources.filterForCleanup(topics).map((x: any) => x.delete?.()) + ); + + const schemas: any[] = []; + for await (const s of pubsub.listSchemas()) { + schemas.push(pubsub.schema(s.name!)); + } + await Promise.all( + resources.filterForCleanup(schemas).map(x => x.delete?.()) + ); + }); + + async function publishAndListen(encoding: 'BINARY' | 'JSON') { + const topicName = resources.generateName(`topic-${encoding}`); + const subName = resources.generateName(`sub-${encoding}`); + + const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString(); + const [topic] = await pubsub.createTopic({ + name: topicName, + schemaSettings: { + schema: await pubsub.schema(schemaId).getName(), + encoding, + } + }); + const [subscription] = await pubsub.subscription(subName).get(); + + const type = avro.parse(definition); + + const province = { + name: 'Ontario', + post_abbr: 'ON', + }; + + const dataBuffer = type.toBuffer(province); + const messageId = await topic.publishMessage({data: dataBuffer}); + assert.ok(messageId); + + const message = await waitForMessage(subscription, { + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for Avro record', + }); + + const schemaMetadata = Schema.metadataFromMessage(message.attributes); + assert.strictEqual(schemaMetadata.encoding, encoding); + + const result = type.fromBuffer(message.data) as any; + assert.strictEqual(result.name, 'Ontario'); + assert.strictEqual(result.post_abbr, 'ON'); + } + + it('should publish and listen for avro records (binary encoding)', async () => { + publishAndListen('BINARY'); + }); + + it('should publish and listen for avro records (json encoding)', async () => { + publishAndListen('JSON'); + }); + + it('should listen for avro records with revisions', async () => { + const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString(); + + const schemaClient = await pubsub.getSchemaClient(); + + const topicName = resources.generateName(`topic-rev`); + const subName = resources.generateName(`sub-rev`); + const [topic] = await pubsub.createTopic({ + name: topicName, + schemaSettings: { + schema: await pubsub.schema(schemaId).getName(), + encoding: 'BINARY', + }, + }); + const [subscription] = await pubsub.createSubscription(topicName, subName); + + const type = avro.parse(definition); + const province = { + name: 'Ontario', + post_abbr: 'ON', + }; + + const dataBuffer = type.toBuffer(province); + await topic.publishMessage({data: dataBuffer}); + + const message = await waitForMessage(subscription, { + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for Avro revision', + }); + + const schemaMetadata = Schema.metadataFromMessage(message.attributes); + const revision = schemaMetadata.revision!; + assert.ok(revision); + + const [fetchedSchema] = await schemaClient.getSchema({ + name: `${schemaMetadata.name}@${schemaMetadata.revision}`, + }); + + const reader = avro.parse(fetchedSchema.definition!); + const result = reader.fromBuffer(message.data) as any; + + assert.strictEqual(result.name, 'Ontario'); + assert.strictEqual(result.post_abbr, 'ON'); + }); +}); diff --git a/handwritten/pubsub/system-test/batch-flow-samples.test.ts b/handwritten/pubsub/system-test/batch-flow-samples.test.ts new file mode 100644 index 000000000000..08b72744fb34 --- /dev/null +++ b/handwritten/pubsub/system-test/batch-flow-samples.test.ts @@ -0,0 +1,109 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {Message, PubSub, PublishOptions} from '../src'; +import * as assert from 'assert'; +import {describe, it, after, before} from 'mocha'; +import {TestResources} from './testResources'; +import {waitForMessages} from './common'; + +describe('Batch and Flow Control Samples System Tests', () => { + const pubsub = new PubSub(); + const resources = new TestResources('ps-sys-batch'); + + let topicName: string; + let subName: string; + + before(async () => { + topicName = resources.generateName('topic'); + subName = resources.generateName('sub'); + }); + + after(async () => { + const [subscriptions] = await pubsub.getSubscriptions(); + await Promise.all( + resources.filterForCleanup(subscriptions).map(x => x.delete?.()) + ); + + const [topics] = await pubsub.getTopics(); + await Promise.all( + resources.filterForCleanup(topics).map((x: any) => x.delete?.()) + ); + }); + + it('should publish batched messages', async () => { + const [topic] = await pubsub.createTopic(topicName); + const [subscription] = await topic.createSubscription(subName); + + const publishOptions: PublishOptions = { + batching: { + maxMessages: 10, + maxMilliseconds: 2000, + }, + }; + const batchPublisher = pubsub.topic(topicName, publishOptions); + + const promises: Promise[] = []; + for (let i = 0; i < 10; i++) { + promises.push(batchPublisher.publishMessage({data: Buffer.from(`message ${i}`)})); + } + + const messageIds = await Promise.all(promises); + assert.strictEqual(messageIds.length, 10); + + const messages = await waitForMessages(subscription, { + count: 10, + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for batched messages', + }); + + assert.strictEqual(messages.length, 10); + }); + + it('should publish with flow control', async () => { + const flowTopicName = resources.generateName('flow'); + const flowSubName = resources.generateName('flowsub'); + + const [topic] = await pubsub.createTopic(flowTopicName); + const [subscription] = await topic.createSubscription(flowSubName); + + const options = { + flowControlOptions: { + maxOutstandingMessages: 5, + maxOutstandingBytes: 1024, + }, + }; + + const topicWithFlow = pubsub.topic(flowTopicName, options); + const flow = topicWithFlow.flowControlled(); + + for (let i = 0; i < 10; i++) { + const wait = flow.publish({data: Buffer.from('flow control message')}); + if (wait) { + await wait; + } + } + + const messageIds = await flow.all(); + assert.strictEqual(messageIds.length, 10); + + const messages = await waitForMessages(subscription, { + count: 10, + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for flow control messages', + }); + + assert.strictEqual(messages.length, 10); + }); +}); diff --git a/handwritten/pubsub/system-test/common.ts b/handwritten/pubsub/system-test/common.ts new file mode 100644 index 000000000000..dfad1c665952 --- /dev/null +++ b/handwritten/pubsub/system-test/common.ts @@ -0,0 +1,173 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {Message, Subscription} from '../src'; + +export interface WaitForMessagesOptions { + /** Number of messages to collect before resolving. Defaults to 1. */ + count?: number; + /** Timeout in milliseconds. Defaults to 15,000ms. */ + timeoutMs?: number; + /** Whether to automatically ack received messages. Defaults to true. */ + autoAck?: boolean; + /** Optional filter predicate. Only matching messages increment count and get collected. */ + filter?: (message: Message) => boolean; + /** Optional per-message inspection hook called for matching messages before ack/resolve. */ + onMessage?: (message: Message) => void | Promise; + /** Error message prefix if timeout occurs. */ + timeoutErrorMessage?: string; + /** Whether to call subscription.close() after receiving all expected messages. Defaults to false. */ + closeWhenDone?: boolean; +} + +/** + * Listens for messages on a subscription until the expected count is reached, + * a timeout occurs, or an error is emitted. + * + * Guarantees that all attached listeners ('message', 'error') and any active timeouts + * are removed on success, error, or timeout to avoid leaking resources. + */ +export async function waitForMessages( + subscription: Subscription, + options: WaitForMessagesOptions = {} +): Promise { + const { + count = 1, + timeoutMs = 15000, + autoAck = true, + filter = () => true, + onMessage, + timeoutErrorMessage = 'Timeout waiting for messages', + closeWhenDone = false, + } = options; + + return new Promise((resolve, reject) => { + const received: Message[] = []; + let timeoutId: NodeJS.Timeout | undefined; + + let cleanedUp = false; + const cleanup = () => { + if (cleanedUp) return; + cleanedUp = true; + + if (timeoutId) { + clearTimeout(timeoutId); + timeoutId = undefined; + } + subscription.removeListener('message', messageHandler); + subscription.removeListener('error', errorHandler); + + if (closeWhenDone) { + void subscription.close(); + } + }; + + const messageHandler = async (msg: Message) => { + try { + if (!filter(msg)) { + return; + } + + if (onMessage) { + await onMessage(msg); + } + + if (autoAck) { + msg.ack(); + } + + received.push(msg); + + if (received.length >= count) { + cleanup(); + resolve(received); + } + } catch (err) { + cleanup(); + reject(err); + } + }; + + const errorHandler = (err: Error) => { + cleanup(); + reject(err); + }; + + timeoutId = setTimeout(() => { + cleanup(); + reject(new Error(`${timeoutErrorMessage} (${timeoutMs}ms)`)); + }, timeoutMs); + + subscription.on('error', errorHandler); + subscription.on('message', messageHandler); + }); +} + +/** + * Listens for a single message on a subscription with guaranteed cleanup. + */ +export async function waitForMessage( + subscription: Subscription, + options: Omit = {} +): Promise { + const messages = await waitForMessages(subscription, {...options, count: 1}); + return messages[0]; +} + +/** + * Tracks event listeners attached to a Subscription and removes them on cleanup. + */ +export class SubscriptionScope { + private cleanups: Array<() => void> = []; + + constructor(public readonly subscription: Subscription) {} + + /** Attaches an event listener and tracks it for automatic teardown */ + // eslint-disable-next-line @typescript-eslint/no-explicit-any + on(event: string, listener: (...args: any[]) => void): this { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + this.subscription.on(event as any, listener as any); + this.cleanups.push(() => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + this.subscription.removeListener(event as any, listener as any); + }); + return this; + } + + /** Cleans up all listeners registered through this scope */ + cleanup(): void { + while (this.cleanups.length > 0) { + const fn = this.cleanups.pop(); + try { + fn?.(); + } catch {} + } + } +} + +/** + * Runs an asynchronous test block with a managed subscription scope, guaranteeing + * that all registered listeners are removed in a finally block. + */ +export async function withSubscriptionScope( + subscription: Subscription, + fn: (scope: SubscriptionScope) => Promise +): Promise { + const scope = new SubscriptionScope(subscription); + try { + return await fn(scope); + } finally { + scope.cleanup(); + } +} diff --git a/handwritten/pubsub/system-test/fixtures/provinces.proto b/handwritten/pubsub/system-test/fixtures/provinces.proto new file mode 100644 index 000000000000..08f05488efc0 --- /dev/null +++ b/handwritten/pubsub/system-test/fixtures/provinces.proto @@ -0,0 +1,8 @@ +syntax = "proto3"; + +package utilities; + +message Province { + string name = 1; + string post_abbr = 2; +} diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts new file mode 100644 index 000000000000..6bd8a05e4709 --- /dev/null +++ b/handwritten/pubsub/system-test/otel-samples.test.ts @@ -0,0 +1,87 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {Message, PubSub, Subscription, Topic} from '../src'; +import * as tracing from '../src/telemetry-tracing'; +import * as assert from 'assert'; +import {describe, it, after, before} from 'mocha'; +import {TestResources} from './testResources'; +import {NodeTracerProvider} from '@opentelemetry/sdk-trace-node'; +import {SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base'; +import {waitForMessage} from './common'; + +describe('OpenTelemetry Samples System Tests', () => { + const pubsub = new PubSub({enableOpenTelemetryTracing: true}); + const resources = new TestResources('ps-sys-otel'); + + let topicName: string; + let subName: string; + let topic: Topic; + let subscription: Subscription; + let processor: SimpleSpanProcessor; + let provider: NodeTracerProvider; + let exporter: InMemorySpanExporter; + + before(async () => { + topicName = resources.generateName('ot'); + subName = resources.generateName('ot'); + topic = (await pubsub.createTopic(topicName))[0]; + subscription = (await topic.createSubscription(subName))[0]; + + exporter = new InMemorySpanExporter(); + + // Build a tracer provider and a span processor to do + // something with the spans we're generating. + processor = new SimpleSpanProcessor(exporter); + provider = new NodeTracerProvider({ + spanProcessors: [processor], + }); + provider.register(); + }); + + after(async () => { + // Don't interfere with other tests. + tracing.setGloballyEnabled(false); + provider.shutdown(); + + const [subscriptions] = await pubsub.getSubscriptions(); + await Promise.all( + resources.filterForCleanup(subscriptions).map(x => x.delete?.()) + ); + + const [topics] = await pubsub.getTopics(); + await Promise.all( + resources.filterForCleanup(topics).map((x: any) => x.delete?.()) + ); + }); + + it('should publish and listen with OpenTelemetry tracing', async () => { + const data = 'Hello, world!'; + const dataBuffer = Buffer.from(data); + + const messageId = await topic.publishMessage({data: dataBuffer}); + assert.ok(messageId); + + const message = await waitForMessage(subscription, { + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for OTel message', + }); + + assert.strictEqual(message.data.toString(), data); + + await processor.forceFlush(); + const spans = exporter.getFinishedSpans(); + assert.ok(spans.length > 0, 'Should have generated spans'); + }); +}); diff --git a/handwritten/pubsub/system-test/proto-js.d.ts b/handwritten/pubsub/system-test/proto-js.d.ts new file mode 100644 index 000000000000..c83a47f02562 --- /dev/null +++ b/handwritten/pubsub/system-test/proto-js.d.ts @@ -0,0 +1,18 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +interface ProvinceObject { + name: string; + postAbbr: string; +} diff --git a/handwritten/pubsub/system-test/protobuf-samples.test.ts b/handwritten/pubsub/system-test/protobuf-samples.test.ts new file mode 100644 index 000000000000..9b0b46f4e230 --- /dev/null +++ b/handwritten/pubsub/system-test/protobuf-samples.test.ts @@ -0,0 +1,95 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {Message, PubSub, Schema} from '../src'; +import * as assert from 'assert'; +import {describe, it, after, before} from 'mocha'; +import {TestResources} from './testResources'; +import * as protobuf from 'protobufjs'; +import * as fs from 'fs'; +import {waitForMessage} from './common'; + +describe('Protobuf Samples System Tests', () => { + const pubsub = new PubSub(); + const resources = new TestResources('ps-sys-proto'); + + after(async () => { + const [subscriptions] = await pubsub.getSubscriptions(); + await Promise.all( + resources.filterForCleanup(subscriptions).map(x => x.delete?.()) + ); + + const [topics] = await pubsub.getTopics(); + await Promise.all( + resources.filterForCleanup(topics).map((x: any) => x.delete?.()) + ); + + const schemas: any[] = []; + for await (const s of pubsub.listSchemas()) { + schemas.push(pubsub.schema(s.name!)); + } + await Promise.all( + resources.filterForCleanup(schemas).map(x => x.delete?.()) + ); + }); + + it('should publish and listen for protobuf messages', async () => { + const topicName = resources.generateName('topic'); + const subName = resources.generateName('sub'); + const schemaId = resources.generateName('schema'); + const definition = fs.readFileSync('system-test/fixtures/provinces.proto').toString(); + await pubsub.createSchema(schemaId, 'PROTOCOL_BUFFER', definition); + + const [topic] = await pubsub.createTopic({ + name: topicName, + schemaSettings: { + schema: await pubsub.schema(schemaId).getName(), + encoding: 'BINARY', + }, + }); + + const [subscription] = await topic.createSubscription(subName); + + // Make an encoder using the protobufjs library. + // + // Since we're providing the test message for a specific schema here, we'll + // also code in the path to a sample proto definition. + const root = await protobuf.load('system-test/fixtures/provinces.proto'); + const Province = root.lookupType('utilities.Province'); + const province: ProvinceObject = { + name: 'Ontario', + postAbbr: 'ON', + }; + + const message = Province.create(province); + const dataBuffer = Buffer.from(Province.encode(message).finish()); + + const messageId = await topic.publishMessage({data: dataBuffer}); + assert.ok(messageId); + + const received = await waitForMessage(subscription, { + timeoutMs: 15000, + timeoutErrorMessage: 'Timeout waiting for Proto message', + }); + + const schemaMetadata = Schema.metadataFromMessage(received.attributes); + assert.strictEqual(schemaMetadata.encoding, 'BINARY'); + + for (let i = 0; i < received.length; i++) { + const result = Province.decode(received.data) as any; + assert.strictEqual(result.name, 'Ontario'); + assert.strictEqual(result.postAbbr || result.post_abbr, 'ON'); + } + }); +}); diff --git a/handwritten/pubsub/system-test/pubsub.ts b/handwritten/pubsub/system-test/pubsub.ts index 44aaff741107..9b63cfd72ebc 100644 --- a/handwritten/pubsub/system-test/pubsub.ts +++ b/handwritten/pubsub/system-test/pubsub.ts @@ -36,6 +36,7 @@ import { import {MessageOptions} from '../src/topic'; import {TestResources} from '../test/testResources'; import {GoogleError} from 'google-gax'; +import {waitForMessage, waitForMessages, withSubscriptionScope} from './common'; const pubsub = new PubSub(); @@ -122,10 +123,7 @@ describe('pubsub', () => { for (let i = 0; i < 6; i++) { await topic.topic.publishMessage(message); } - return new Promise((resolve, reject) => { - sub.sub.on('error', reject); - sub.sub.once('message', resolve); - }); + return waitForMessage(sub.sub); } before(async () => { @@ -217,14 +215,25 @@ describe('pubsub', () => { }); it('should publish a message', async () => { - const testTopic = await generateTopic('pub-msg'); - const topic = testTopic.topic; - const message = { - data: Buffer.from('message from me'), - orderingKey: 'a', - }; - - const result = await topic.publishMessage(message); + const tname = generateTopicName('publish'); + const sname = generateSubName('publish'); + + const [topic] = await pubsub.topic(tname).get({autoCreate: true}); + const [subscription] = await topic.subscription(sname).get({autoCreate: true}); + + // --- From sample (publishMessage.js) --- + const data = 'Hello, world!'; + const dataBuffer = Buffer.from(data); + const messageId = await topic.publishMessage({data: dataBuffer}); + console.log(`Message ${messageId} published.`); + + // --- From test (topics.test.ts) --- + const message = await waitForMessage(subscription, { + timeoutMs: 10000, + timeoutErrorMessage: 'Timeout', + }); + + assert.strictEqual(message.data.toString(), data); }); it('should publish a message with attributes', async () => { @@ -536,107 +545,79 @@ describe('pubsub', () => { const testTopic = await generateTopic('dne-sub'); const subscription = testTopic.topic.subscription(generateSubName('dne-sub')); - await new Promise((res, rej) => { - subscription.on('error', (err: {code: number}) => { - assert.strictEqual(err.code, 5); - subscription.close(res); - }); + await withSubscriptionScope(subscription, scope => { + return new Promise((res, rej) => { + scope.on('error', async (err: {code: number}) => { + try { + assert.strictEqual(err.code, 5); + await subscription.close(); + res(); + } catch (e) { + rej(e); + } + }); - subscription.on('message', () => { - rej(new Error('Should not have been called.')); + scope.on('message', () => { + rej(new Error('Should not have been called.')); + }); }); }); }); it('should receive the published messages', async () => { const pop = await subPop('recv', 1); - let messageCount = 0; const subscription = pop.subs[0]; - await new Promise((res, rej) => { - subscription.on('error', rej); - - // eslint-disable-next-line @typescript-eslint/no-explicit-any - subscription.on('message', message => { - assert.deepStrictEqual(message.data, Buffer.from('hello')); - message.ack(); - - if (++messageCount === 10) { - subscription.close(res); - } - }); + const messages = await waitForMessages(subscription, { + count: 10, + filter: msg => { + assert.deepStrictEqual(msg.data, Buffer.from('hello')); + return true; + }, + closeWhenDone: true, }); + assert.strictEqual(messages.length, 10); }); it('should ack the message', async () => { const pop = await subPop('ack', 1); const subscription = pop.subs[0]; - await new Promise((res, rej) => { - let finished = false; - subscription.on('error', () => { - if (!finished) { - finished = true; - subscription.close(rej); - } - }); - subscription.on('message', ack); - - function ack(message: Message) { - if (!finished) { - finished = true; - message.ack(); - subscription.close(res); - } - } + const message = await waitForMessage(subscription, { + autoAck: true, + closeWhenDone: true, }); + assert.ok(message); }); it('should nack the message', async () => { const pop = await subPop('nack', 1); const subscription = pop.subs[0]; - await new Promise((res, rej) => { - let finished = false; - subscription.on('error', () => { - if (!finished) { - finished = true; - subscription.close(rej); - } - }); - subscription.on('message', nack); - - function nack(message: Message) { - if (!finished) { - finished = true; - message.nack(); - subscription.close(res); - } - } + const message = await waitForMessage(subscription, { + autoAck: false, + onMessage: msg => { + msg.nack(); + }, + closeWhenDone: true, }); + assert.ok(message); }); it('should respect flow control limits', async () => { const maxMessages = 3; - let messageCount = 0; const pop = await subPop('fcl', 1); const subscription = pop.topic.subscription(pop.testSubs[0].name, { flowControl: {maxMessages, allowExcessMessages: false}, }); - await new Promise((res, rej) => { - subscription.on('error', rej); - subscription.on('message', onMessage); - - function onMessage() { - if (++messageCount < maxMessages) { - return; - } - - subscription.close(res); - } + const messages = await waitForMessages(subscription, { + count: maxMessages, + autoAck: false, + closeWhenDone: true, }); + assert.strictEqual(messages.length, maxMessages); }); it('should send and receive large messages', async () => { @@ -645,16 +626,12 @@ describe('pubsub', () => { const data = crypto.randomBytes(9000000); // 9mb const messageId = await pop.topic.publishMessage({data}); - await new Promise((res, rej) => { - subscription.on('error', rej).on('message', (message: Message) => { - if (message.id !== messageId) { - return; - } - - assert.deepStrictEqual(data, message.data); - subscription.close(res); - }); + const message = await waitForMessage(subscription, { + filter: msg => msg.id === messageId, + autoAck: true, + closeWhenDone: true, }); + assert.deepStrictEqual(data, message.data); }); it('should detach subscriptions', async () => { @@ -862,9 +839,13 @@ describe('pubsub', () => { type WorkCallback = (arg: Message, resolve: Function) => void; function makeMessagePromise(subscription: Subscription, workCallback: WorkCallback): Promise { return new Promise(resolve => { - subscription.on('message', (arg: Message) => { - workCallback(arg, resolve); - }); + const messageHandler = (arg: Message) => { + workCallback(arg, () => { + subscription.removeListener('message', messageHandler); + resolve(); + }); + }; + subscription.on('message', messageHandler); }); } diff --git a/handwritten/pubsub/system-test/testResources.test.ts b/handwritten/pubsub/system-test/testResources.test.ts new file mode 100644 index 000000000000..b63e10280508 --- /dev/null +++ b/handwritten/pubsub/system-test/testResources.test.ts @@ -0,0 +1,84 @@ +// Copyright 2022-2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {describe, it, beforeEach} from 'mocha'; +import {TestResources} from './testResources'; +import * as assert from 'node:assert'; + +describe('testResources (unit)', () => { + const fixedId = 'fixed'; + const fixedTime = Date.now(); + const fakeTokenMaker = { + uuid: () => fixedId, + timestamp: () => fixedTime, + }; + + const suiteId = 'someSuite'; + let testResources!: TestResources; + + beforeEach(() => { + testResources = new TestResources(suiteId, fakeTokenMaker); + }); + + it('has predictable prefixes', () => { + const prefix = testResources.getPrefix('testId'); + assert.strictEqual(prefix, `${suiteId}-${fixedTime}-testId`); + + const normalizedPrefix = testResources.getPrefix('test-id-dashes'); + assert.strictEqual( + normalizedPrefix, + `${suiteId}-${fixedTime}-test_id_dashes` + ); + + const suitePrefix = testResources.getPrefix(); + assert.strictEqual(suitePrefix, `${suiteId}-${fixedTime}`); + }); + + it('generates names', () => { + const prefix = testResources.getPrefix('testId'); + const name = testResources.generateName('testId'); + assert.strictEqual(name, `${prefix}-${fixedId}`); + }); + + it('filters for cleanup', () => { + const resources = [ + { + // Not related + name: 'ooga', + }, + { + // For current test run + name: `${suiteId}-${fixedTime}-bob-98719284791`, + }, + { + // For previous test run, but not very old + name: `${suiteId}-${fixedTime - 100}-bob-124897912`, + }, + { + // For previous test run, but old + name: `${suiteId}-${fixedTime - 3000 * 60 * 60}-bob-57823975`, + }, + ]; + const filtered = testResources.filterForCleanup(resources); + assert.strictEqual(filtered.length, 2); + assert.strictEqual( + 1, + filtered.filter(r => r.name?.includes('bob-9871')).length + ); + assert.strictEqual( + 1, + filtered.filter(r => r.name?.includes('bob-5782')).length + ); + }); +}); diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts new file mode 100644 index 000000000000..0fd11552ad60 --- /dev/null +++ b/handwritten/pubsub/system-test/testResources.ts @@ -0,0 +1,182 @@ +// Copyright 2022-2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// I don't like that these two files (this plus ".test") are duplicated +// across the two test structures, but because of the tangle of rootDirs +// and package.json "files", it's hard to avoid it. + +import * as crypto from 'node:crypto'; + +// Returns a shortened UUID that can be used to identify a +// specific run of a specific test. +function shortUUID() { + return crypto.randomUUID().split('-').shift()!; +} + +export interface TokenMaker { + uuid(): string; + timestamp(): number; +} + +export const defaultMaker = { + uuid: shortUUID, + timestamp: () => Date.now(), +}; + +export interface Resource { + name?: string | null | undefined; + delete?(): Promise; +} + +function normalizeId(id: string): string { + return id.replace(/-/g, '_'); +} + +/** + * Manages the names of testing resources during a test run. It's + * easily to accidentally leak resources, and it's easy to accidentally + * have conflicts with tests running concurrently, so this class helps + * you manage them. + * + * Used nomenclature: + * Test - a single test for a single aspect of code; for example, + * "create a topic in pub/sub" + * Test Suite - a collection of tests that are generally run together; + * for example, "test topic operations in pub/sub" + * Test Run - a single run of a test suite (or single test within a suite); + * for example, "run the tests for PR #1234, 5th attempt" + */ +export class TestResources { + testSuiteId: string; + currentTime: string; + tokenMaker: TokenMaker; + + /** + * @param testSuiteId [string] A unique ID for a test suite (e.g. + * pubsub-topics). + */ + constructor(testSuiteId: string, tokenMaker: TokenMaker = defaultMaker) { + this.testSuiteId = normalizeId(testSuiteId); + this.currentTime = `${tokenMaker.timestamp()}`; + this.tokenMaker = tokenMaker; + } + + /** + * Returns the resource prefix for the current run of the test suite. + * Optionally, testId may specify the specific ID of a test in the + * suite. + */ + getPrefix(testId?: string): string { + if (testId) { + return [this.testSuiteId, this.currentTime, normalizeId(testId)].join( + '-', + ); + } else { + return [this.testSuiteId, this.currentTime].join('-'); + } + } + + /** + * Generates a unique resource name for one run of a test within + * a test suite. + */ + generateName(testId: string): string { + return [this.getPrefix(testId), this.tokenMaker.uuid()].join('-'); + } + + /** + * Generates a unique resource name for one run of a test within + * a test suite for BigQuery resources. + */ + generateBigQueryName(testId: string): string { + return [normalizeId(this.getPrefix(testId)), this.tokenMaker.uuid()].join( + '_', + ); + } + + /** + * Generates a unique resource name for one run of a test within + * a test suite for Cloud Storage resources. + */ + generateStorageName(testId: string): string { + return [normalizeId(this.getPrefix(testId)), this.tokenMaker.uuid()].join( + '_', + ); + } + + /** + * Given a list of resource names (and a test ID), this will return + * a list of all resources that should be deleted to clean up for + * the current run of that particular test. + */ + filterForTest(testId: string, allResources: Resource[]): Resource[] { + const prefix = this.getPrefix(testId); + return allResources.filter(n => n.name?.includes(prefix)); + } + + /** + * Given a list of resource names, this will return a list of all + * resources that should be deleted to clean up after the current + * run of a test suite. + */ + filterForCurrentRun(allResources: Resource[]): Resource[] { + const prefix = this.getPrefix(); + return allResources.filter(n => n.name?.includes(prefix)); + } + + /** + * Given a list of resource names, this will return a list of all + * resources that should be deleted to clean up after any run + * of the current test suite. Note that some of the names may + * still be in use. + */ + filterForSuite(allResources: Resource[]): Resource[] { + return allResources.filter(n => n.name?.includes(this.testSuiteId)); + } + + /** + * Given a list of resource names, this will return a list of all + * resources that should be deleted to generally clean up after any + * run of the current test suite. This is much like filterForSuite(), + * but it also filters by age - items that are less than 2 hours + * old will not be cleaned. + */ + filterForCleanup(allResources: Resource[]): Resource[] { + const currentRunPrefix = this.getPrefix(); + return allResources.filter(n => { + let name = n.name || undefined; + if (name === undefined) { + return false; + } + + // We'll always get at least one thing. + name = name.split('/').pop()!; + + if (name.startsWith(currentRunPrefix)) { + return true; + } + + if (name.startsWith(this.testSuiteId)) { + const parts = name.split('-'); + const createdAt = Number(parts[1]); + const timeDiff = (this.tokenMaker.timestamp() - createdAt) / (1000 * 60 * 60); + if (timeDiff >= 2) { + return true; + } + } + + return false; + }); + } +} diff --git a/handwritten/pubsub/test/message-queues.ts b/handwritten/pubsub/test/message-queues.ts index 35bda63ebcc2..1b9738023f35 100644 --- a/handwritten/pubsub/test/message-queues.ts +++ b/handwritten/pubsub/test/message-queues.ts @@ -17,10 +17,10 @@ import * as assert from 'assert'; import {describe, it, before, beforeEach, afterEach} from 'mocha'; import {EventEmitter} from 'events'; -import {CallOptions, GoogleError, loggingUtils, Status} from 'google-gax'; +import {CallOptions, GoogleError, Status} from 'google-gax'; import * as sinon from 'sinon'; -import * as crypto from 'crypto'; import defer = require('p-defer'); +import * as crypto from 'node:crypto'; import * as messageTypes from '../src/message-queues'; import {BatchError} from '../src/message-queues'; @@ -418,10 +418,7 @@ describe('MessageQueues', () => { fakeLog.remove(); assert.strictEqual(fakeLog.called, true); - assert.strictEqual( - fakeLog.fields!.severity, - 'INFO', - ); + assert.strictEqual(fakeLog.fields!.severity, 'INFO'); assert.strictEqual(fakeLog.args![1] as string, 'logtest'); }); @@ -679,10 +676,7 @@ describe('MessageQueues', () => { fakeLog.remove(); assert.strictEqual(fakeLog.called, true); - assert.strictEqual( - fakeLog.fields!.severity, - 'INFO', - ); + assert.strictEqual(fakeLog.fields!.severity, 'INFO'); assert.strictEqual(fakeLog.args![1] as string, 'logtest'); }); diff --git a/handwritten/pubsub/test/message-stream.ts b/handwritten/pubsub/test/message-stream.ts index 00bc31fff02e..3836b2620989 100644 --- a/handwritten/pubsub/test/message-stream.ts +++ b/handwritten/pubsub/test/message-stream.ts @@ -20,9 +20,9 @@ import {grpc} from 'google-gax'; import * as proxyquire from 'proxyquire'; import * as sinon from 'sinon'; import {Duplex, PassThrough} from 'stream'; -import * as crypto from 'crypto'; import * as defer from 'p-defer'; import {promisify} from 'util'; +import * as crypto from 'node:crypto'; import * as messageTypes from '../src/message-stream'; import {Subscriber} from '../src/subscriber'; @@ -534,7 +534,7 @@ describe('MessageStream', () => { assert.strictEqual(spy.callCount, 5); const {args} = spy.firstCall; - const request = args[0] as any; + const request = args[0] as {protocolVersion?: string | number}; assert.strictEqual(String(request.protocolVersion), '1'); diff --git a/handwritten/pubsub/test/subscriber.ts b/handwritten/pubsub/test/subscriber.ts index 64dcbb06973f..3cec7ccd06af 100644 --- a/handwritten/pubsub/test/subscriber.ts +++ b/handwritten/pubsub/test/subscriber.ts @@ -22,10 +22,10 @@ import {common as protobuf} from 'protobufjs'; import * as proxyquire from 'proxyquire'; import * as sinon from 'sinon'; import {PassThrough} from 'stream'; -import * as crypto from 'crypto'; import * as opentelemetry from '@opentelemetry/api'; import {google} from '../protos/protos'; import * as defer from 'p-defer'; +import * as crypto from 'node:crypto'; import {HistogramOptions} from '../src/histogram'; import {FlowControlOptions, LeaseManager} from '../src/lease-manager'; @@ -37,7 +37,6 @@ import {SpanKind} from '@opentelemetry/api'; import {Duration} from '../src'; import * as tracing from '../src/telemetry-tracing'; import {FakeLog, TestUtils} from './test-utils'; -import {loggingUtils} from 'google-gax'; type PullResponse = google.pubsub.v1.IStreamingPullResponse; @@ -450,10 +449,7 @@ describe('Subscriber', () => { await subscriber.ack(message); assert.strictEqual(fakeLog.called, true); - assert.strictEqual( - fakeLog.fields!.severity, - 'INFO', - ); + assert.strictEqual(fakeLog.fields!.severity, 'INFO'); assert.strictEqual(fakeLog.args![1], message.id); }); @@ -468,10 +464,7 @@ describe('Subscriber', () => { await subscriber.ack(message); assert.strictEqual(fakeLog.called, true); - assert.strictEqual( - fakeLog.fields!.severity, - 'INFO', - ); + assert.strictEqual(fakeLog.fields!.severity, 'INFO'); assert.strictEqual(fakeLog.args![1], message.id); }); @@ -822,9 +815,10 @@ describe('Subscriber', () => { timeout: Duration.from({milliseconds: 100}), }, }); - const prom = subscriber.close().then(() => { + const prom = (async () => { + await subscriber.close(); closed = true; - }); + })(); // Advance time past the timeout clock.tick(200); @@ -854,9 +848,10 @@ describe('Subscriber', () => { timeout: Duration.from({milliseconds: 100}), }, }); - const prom = subscriber.close().then(() => { + const prom = (async () => { + await subscriber.close(); closed = true; - }); + })(); // Resolve drains quickly ackDrainDeferred.resolve(); @@ -944,10 +939,7 @@ describe('Subscriber', () => { await subscriber.nack(message); assert.strictEqual(fakeLog.called, true); - assert.strictEqual( - fakeLog.fields!.severity, - 'INFO', - ); + assert.strictEqual(fakeLog.fields!.severity, 'INFO'); assert.strictEqual(fakeLog.args![1], message.id); }); diff --git a/handwritten/pubsub/test/testResources.test.ts b/handwritten/pubsub/test/testResources.test.ts index b1cf3a67371d..b759389df23a 100644 --- a/handwritten/pubsub/test/testResources.test.ts +++ b/handwritten/pubsub/test/testResources.test.ts @@ -14,7 +14,7 @@ import {describe, it, beforeEach} from 'mocha'; import {TestResources} from './testResources'; -import * as assert from 'assert'; +import * as assert from 'node:assert'; describe('testResources (unit)', () => { const fixedId = 'fixed'; diff --git a/handwritten/pubsub/test/testResources.ts b/handwritten/pubsub/test/testResources.ts index 4503138e51cd..93df7dbc9c6e 100644 --- a/handwritten/pubsub/test/testResources.ts +++ b/handwritten/pubsub/test/testResources.ts @@ -16,7 +16,7 @@ // across the two test structures, but because of the tangle of rootDirs // and package.json "files", it's hard to avoid it. -import * as crypto from 'crypto'; +import * as crypto from 'node:crypto'; // Returns a shortened UUID that can be used to identify a // specific run of a specific test.