tests(pubsub): integrate pub/sub sample system tests that are still useful

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",

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(

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<T>(buf: Buffer): T;
+    fromString<T>(str: string): T;
+    toBuffer<T>(item: T): Buffer;
+    toString<T>(item: T): string;
+  }
+}

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');
+  });
+});

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<string>[] = [];
+    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);
+  });
+});