Publish messages to a topic with a schema

namespacepubsub=::google::cloud::pubsub;
using::google::cloud::future;
using::google::cloud::StatusOr;
[](pubsub::Publisherpublisher){
autoconstexprkNewYork=
R"js({ "name": "New York", "post_abbr": "NY" })js";
autoconstexprkPennsylvania=
R"js({ "name": "Pennsylvania", "post_abbr": "PA" })js";
std::vector<future<void>>done;
autohandler=[](future<StatusOr<std::string>>f){
autoid=f.get();
if(!id)throwstd::move(id).status();
};
for(autoconst*data:{kNewYork,kPennsylvania}){
done.push_back(
publisher.Publish(pubsub::MessageBuilder{}.SetData(data).Build())
.then(handler));
}
// Block until all messages are published.
for(auto&d:done)d.get();
}

namespacepubsub=::google::cloud::pubsub;
using::google::cloud::future;
using::google::cloud::StatusOr;
[](pubsub::Publisherpublisher){
std::vector<std::pair<std::string,std::string>>states{
{"New York","NY"},
{"Pennsylvania","PA"},
};
std::vector<future<void>>done;
autohandler=[](future<StatusOr<std::string>>f){
autoid=f.get();
if(!id)throwstd::move(id).status();
};
for(auto&data:states){
google::cloud::pubsub::samples::Statestate;
state.set_name(data.first);
state.set_post_abbr(data.second);
done.push_back(publisher
.Publish(pubsub::MessageBuilder{}
.SetData(state.SerializeAsString())
.Build())
.then(handler));
}
// Block until all messages are published.
for(auto&d:done)d.get();
}

usingAvro.IO;
usingAvro.Specific;
usingGoogle.Cloud.PubSub.V1;
usingSystem;
usingSystem.Collections.Generic;
usingSystem.IO;
usingSystem.Linq;
usingSystem.Threading;
usingSystem.Threading.Tasks;
publicclassPublishAvroMessagesAsyncSample
{
publicasyncTask<int>PublishAvroMessagesAsync(stringprojectId,stringtopicId,IEnumerable<AvroUtilities.State>messageStates)
{
TopicNametopicName=TopicName.FromProjectTopic(projectId,topicId);
PublisherClientpublisher=awaitPublisherClient.CreateAsync(topicName);
PublisherServiceApiClientpublishApi=PublisherServiceApiClient.Create();
vartopic=publishApi.GetTopic(topicName);
intpublishedMessageCount=0;
varpublishTasks=messageStates.Select(asyncstate=>
{
try
{
stringmessageId=null;
switch(topic.SchemaSettings.Encoding)
{
caseEncoding.Binary:
using(varms=newMemoryStream())
{
varencoder=newBinaryEncoder(ms);
varwriter=newSpecificDefaultWriter(state.Schema);
writer.Write(state,encoder);
messageId=awaitpublisher.PublishAsync(ms.ToArray());
}
break;
caseEncoding.Json:
varjsonMessage=AvroUtilities.StateUtils.StateToJsonString(state);
messageId=awaitpublisher.PublishAsync(jsonMessage);
break;
}
Console.WriteLine($"Published message {messageId}");
Interlocked.Increment(refpublishedMessageCount);
}
catch(Exceptionexception)
{
Console.WriteLine($"An error occurred when publishing message {state}: {exception.Message}");
}
});
awaitTask.WhenAll(publishTasks);
// PublisherClient instance should be shutdown after use.
// The TimeSpan specifies for how long to attempt to publish locally queued messages.
awaitpublisher.ShutdownAsync(TimeSpan.FromSeconds(15));
returnpublishedMessageCount;
}
}

usingGoogle.Cloud.PubSub.V1;
usingGoogle.Protobuf;
usingSystem;
usingSystem.Collections.Generic;
usingSystem.Linq;
usingSystem.Threading;
usingSystem.Threading.Tasks;
publicclassPublishProtoMessagesAsyncSample
{
publicasyncTask<int>PublishProtoMessagesAsync(stringprojectId,stringtopicId,IEnumerable<Utilities.State>messageStates)
{
TopicNametopicName=TopicName.FromProjectTopic(projectId,topicId);
PublisherClientpublisher=awaitPublisherClient.CreateAsync(topicName);
PublisherServiceApiClientpublishApi=PublisherServiceApiClient.Create();
vartopic=publishApi.GetTopic(topicName);
intpublishedMessageCount=0;
varpublishTasks=messageStates.Select(asyncstate=>
{
try
{
stringmessageId=null;
switch(topic.SchemaSettings.Encoding)
{
caseEncoding.Binary:
varbinaryMessage=state.ToByteString();
messageId=awaitpublisher.PublishAsync(binaryMessage);
break;
caseEncoding.Json:
varjsonMessage=JsonFormatter.Default.Format(state);
messageId=awaitpublisher.PublishAsync(jsonMessage);
break;
}
Console.WriteLine($"Published message {messageId}");
Interlocked.Increment(refpublishedMessageCount);
}
catch(Exceptionexception)
{
Console.WriteLine($"An error occurred when publishing message {state}: {exception.Message}");
}
});
awaitTask.WhenAll(publishTasks);
// PublisherClient instance should be shutdown after use.
// The TimeSpan specifies for how long to attempt to publish locally queued messages.
awaitpublisher.ShutdownAsync(TimeSpan.FromSeconds(15));
returnpublishedMessageCount;
}
}

import(
"context"
"fmt"
"io"
"os"
"cloud.google.com/go/pubsub/v2"
"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
"github.com/linkedin/goavro/v2"
)
funcpublishAvroRecords(wio.Writer,projectID,topicID,avscFilestring)error{
// projectID := "my-project-id"
// topicID := "my-topic"
// avscFile = "path/to/an/avro/schema/file(.avsc)/formatted/in/json"
ctx:=context.Background()
client,err:=pubsub.NewClient(ctx,projectID)
iferr!=nil{
returnfmt.Errorf("pubsub.NewClient: %w",err)
}
avroSource,err:=os.ReadFile(avscFile)
iferr!=nil{
returnfmt.Errorf("os.ReadFile err: %w",err)
}
codec,err:=goavro.NewCodec(string(avroSource))
iferr!=nil{
returnfmt.Errorf("goavro.NewCodec err: %w",err)
}
record:=map[string]interface{}{"name":"Alaska","post_abbr":"AK"}
// Get the topic encoding type.
req:=&pubsubpb.GetTopicRequest{
Topic:fmt.Sprintf("projects/%s/topics/%s",projectID,topicID),
}
t,err:=client.TopicAdminClient.GetTopic(ctx,req)
iferr!=nil{
returnfmt.Errorf("got err in GetTopic: %w",err)
}
encoding:=t.SchemaSettings.Encoding
varmsg[]byte
switchencoding{
casepubsubpb.Encoding_BINARY:
msg,err=codec.BinaryFromNative(nil,record)
iferr!=nil{
returnfmt.Errorf("codec.BinaryFromNative err: %w",err)
}
casepubsubpb.Encoding_JSON:
msg,err=codec.TextualFromNative(nil,record)
iferr!=nil{
returnfmt.Errorf("codec.TextualFromNative err: %w",err)
}
default:
returnfmt.Errorf("invalid encoding: %v",encoding)
}
// client.Publisher can be passed a topic ID (e.g. "my-topic") or
// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
// If a topic ID is provided, the project ID from the client is used.
publisher:=client.Publisher(topicID)
result:=publisher.Publish(ctx,&pubsub.Message{
Data:msg,
})
_,err=result.Get(ctx)
iferr!=nil{
returnfmt.Errorf("result.Get: %w",err)
}
fmt.Fprintf(w,"Published avro record: %s\n",string(msg))
returnnil
}

import(
"context"
"fmt"
"io"
"cloud.google.com/go/pubsub/v2"
"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
statepb"github.com/GoogleCloudPlatform/golang-samples/internal/pubsub/schemas"
"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
)
funcpublishProtoMessages(wio.Writer,projectID,topicIDstring)error{
// projectID := "my-project-id"
// topicID := "my-topic"
ctx:=context.Background()
client,err:=pubsub.NewClient(ctx,projectID)
iferr!=nil{
returnfmt.Errorf("pubsub.NewClient: %w",err)
}
state:=&statepb.State{
Name:"Alaska",
PostAbbr:"AK",
}
// Get the topic encoding type.
req:=&pubsubpb.GetTopicRequest{
Topic:fmt.Sprintf("projects/%s/topics/%s",projectID,topicID),
}
t,err:=client.TopicAdminClient.GetTopic(ctx,req)
iferr!=nil{
returnfmt.Errorf("got err in GetTopic: %w",err)
}
encoding:=t.SchemaSettings.Encoding
varmsg[]byte
switchencoding{
casepubsubpb.Encoding_BINARY:
msg,err=proto.Marshal(state)
iferr!=nil{
returnfmt.Errorf("proto.Marshal err: %w",err)
}
casepubsubpb.Encoding_JSON:
msg,err=protojson.Marshal(state)
iferr!=nil{
returnfmt.Errorf("protojson.Marshal err: %w",err)
}
default:
returnfmt.Errorf("invalid encoding: %v",encoding)
}
// client.Publisher can be passed a topic ID (e.g. "my-topic") or
// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
// If a topic ID is provided, the project ID from the client is used.
publisher:=client.Publisher(topicID)
result:=publisher.Publish(ctx,&pubsub.Message{
Data:msg,
})
_,err=result.Get(ctx)
iferr!=nil{
returnfmt.Errorf("result.Get: %w",err)
}
fmt.Fprintf(w,"Published proto message with %#v encoding: %s\n",encoding,string(msg))
returnnil
}

importcom.google.api.core.ApiFuture ;
importcom.google.cloud.pubsub.v1.Publisher ;
importcom.google.cloud.pubsub.v1.TopicAdminClient ;
importcom.google.protobuf.ByteString ;
importcom.google.pubsub.v1.Encoding ;
importcom.google.pubsub.v1.PubsubMessage ;
importcom.google.pubsub.v1.TopicName ;
importjava.io.ByteArrayOutputStream;
importjava.io.IOException;
importjava.util.concurrent.ExecutionException;
importjava.util.concurrent.TimeUnit;
importorg.apache.avro.io.Encoder;
importorg.apache.avro.io.EncoderFactory;
importutilities.State;
publicclass PublishAvroRecordsExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
// Use a topic created with an Avro schema.
StringtopicId="your-topic-id";
publishAvroRecordsExample(projectId,topicId);
}
publicstaticvoidpublishAvroRecordsExample(StringprojectId,StringtopicId)
throwsIOException,ExecutionException,InterruptedException{
Encoding encoding=null;
TopicName topicName=TopicName .of(projectId,topicId);
// Get the topic encoding type.
try(TopicAdminClient topicAdminClient=TopicAdminClient .create()){
encoding=topicAdminClient.getTopic(topicName).getSchemaSettings().getEncoding();
}
// Instantiate an avro-tools-generated class defined in `us-states.avsc`.
Statestate=State.newBuilder().setName("Alaska").setPostAbbr("AK").build();
Publisher publisher=null;
block:
try{
publisher=Publisher .newBuilder(topicName).build();
// Prepare to serialize the object to the output stream.
ByteArrayOutputStreambyteStream=newByteArrayOutputStream();
Encoderencoder=null;
// Prepare an appropriate encoder for publishing to the topic.
switch(encoding){
caseBINARY:
System.out.println("Preparing a BINARY encoder...");
encoder=EncoderFactory.get().directBinaryEncoder(byteStream,/* reuse= */null);
break;
caseJSON:
System.out.println("Preparing a JSON encoder...");
encoder=EncoderFactory.get().jsonEncoder(State.getClassSchema(),byteStream);
break;
default:
breakblock;
}
// Encode the object and write it to the output stream.
state.customEncode(encoder);
encoder.flush ();
// Publish the encoded object as a Pub/Sub message.
ByteString data=ByteString .copyFrom (byteStream.toByteArray());
PubsubMessage message=PubsubMessage .newBuilder().setData (data).build();
System.out.println("Publishing message: "+message);
ApiFuture<String>future=publisher.publish (message);
System.out.println("Published message ID: "+future.get());
}finally{
if(publisher!=null){
publisher.shutdown ();
publisher.awaitTermination (1,TimeUnit.MINUTES);
}
}
}
}

importcom.google.api.core.ApiFuture ;
importcom.google.cloud.pubsub.v1.Publisher ;
importcom.google.cloud.pubsub.v1.TopicAdminClient ;
importcom.google.protobuf.ByteString ;
importcom.google.protobuf.util.JsonFormat ;
importcom.google.pubsub.v1.Encoding ;
importcom.google.pubsub.v1.PubsubMessage ;
importcom.google.pubsub.v1.TopicName ;
importjava.io.IOException;
importjava.util.concurrent.ExecutionException;
importjava.util.concurrent.TimeUnit;
importutilities.StateProto.State;
publicclass PublishProtobufMessagesExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
// Use a topic created with a proto schema.
StringtopicId="your-topic-id";
publishProtobufMessagesExample(projectId,topicId);
}
publicstaticvoidpublishProtobufMessagesExample(StringprojectId,StringtopicId)
throwsIOException,ExecutionException,InterruptedException{
Encoding encoding=null;
TopicName topicName=TopicName .of(projectId,topicId);
// Get the topic encoding type.
try(TopicAdminClient topicAdminClient=TopicAdminClient .create()){
encoding=topicAdminClient.getTopic(topicName).getSchemaSettings().getEncoding();
}
Publisher publisher=null;
// Instantiate a protoc-generated class defined in `us-states.proto`.
Statestate=State.newBuilder().setName("Alaska").setPostAbbr("AK").build();
block:
try{
publisher=Publisher .newBuilder(topicName).build();
PubsubMessage .Buildermessage=PubsubMessage .newBuilder();
// Prepare an appropriately formatted message based on topic encoding.
switch(encoding){
caseBINARY:
message.setData (state.toByteString());
System.out.println("Publishing a BINARY-formatted message:\n"+message);
break;
caseJSON:
StringjsonString=JsonFormat .printer().omittingInsignificantWhitespace().print(state);
message.setData (ByteString .copyFromUtf8 (jsonString));
System.out.println("Publishing a JSON-formatted message:\n"+message);
break;
default:
breakblock;
}
// Publish the message.
ApiFuture<String>future=publisher.publish (message.build());
System.out.println("Published message ID: "+future.get());
}finally{
if(publisher!=null){
publisher.shutdown ();
publisher.awaitTermination (1,TimeUnit.MINUTES);
}
}
}
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
const{PubSub,Encodings}=require('@google-cloud/pubsub');
// And the Apache Avro library
constavro=require('avro-js');
constfs=require('fs');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
asyncfunctionpublishAvroRecords(topicNameOrId){
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
// Get the topic metadata to learn about its schema encoding.
const[topicMetadata]=awaittopic.getMetadata();
consttopicSchemaMetadata=topicMetadata.schemaSettings;
if(!topicSchemaMetadata){
console.log(`Topic ${topicNameOrId} doesn't seem to have a schema.`);
return;
}
constschemaEncoding=topicSchemaMetadata.encoding;
// Make an encoder using the official avro-js library.
constdefinition=fs
.readFileSync('system-test/fixtures/provinces.avsc')
.toString();
consttype=avro.parse(definition);
// Encode the message.
constprovince={
name:'Ontario',
post_abbr:'ON',
};
letdataBuffer;
switch(schemaEncoding){
caseEncodings .Binary:
dataBuffer=type.toBuffer(province);
break;
caseEncodings .Json:
dataBuffer=Buffer.from (type.toString(province));
break;
default:
console.log(`Unknown schema encoding: ${schemaEncoding}`);
break;
}
if(!dataBuffer){
console.log(`Invalid encoding ${schemaEncoding} on the topic.`);
return;
}
constmessageId=awaittopic.publish(dataBuffer);
console.log(`Avro record ${messageId} published.`);
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
const{PubSub,Encodings}=require('@google-cloud/pubsub');
// And the protobufjs library
constprotobuf=require('protobufjs');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
asyncfunctionpublishProtobufMessages(topicNameOrId){
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
// Get the topic metadata to learn about its schema.
const[topicMetadata]=awaittopic.getMetadata();
consttopicSchemaMetadata=topicMetadata.schemaSettings;
if(!topicSchemaMetadata){
console.log(`Topic ${topicNameOrId} doesn't seem to have a schema.`);
return;
}
constschemaEncoding=topicSchemaMetadata.encoding;
// Encode the message.
constprovince={
name:'Ontario',
postAbbr:'ON',
};
// 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.
constroot=awaitprotobuf.load('system-test/fixtures/provinces.proto');
constProvince=root.lookupType('utilities.Province');
constmessage=Province.create(province);
letdataBuffer;
switch(schemaEncoding){
caseEncodings .Binary:
dataBuffer=Buffer.from (Province.encode(message).finish());
break;
caseEncodings .Json:
dataBuffer=Buffer.from (JSON.stringify(message.toJSON()));
break;
default:
console.log(`Unknown schema encoding: ${schemaEncoding}`);
break;
}
if(!dataBuffer){
console.log(`Invalid encoding ${schemaEncoding} on the topic.`);
return;
}
constmessageId=awaittopic.publishMessage ({data:dataBuffer});
console.log(`Protobuf message ${messageId} published.`);
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
import{PubSub,Encodings}from'@google-cloud/pubsub';
// And the Apache Avro library
import*asavrofrom'avro-js';
import*asfsfrom'fs';
// Creates a client; cache this for further use
constpubSubClient=newPubSub();
interfaceProvinceObject{
name:string;
post_abbr:string;
}
asyncfunctionpublishAvroRecords(topicNameOrId:string){
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
// Get the topic metadata to learn about its schema encoding.
const[topicMetadata]=awaittopic.getMetadata();
consttopicSchemaMetadata=topicMetadata.schemaSettings;
if(!topicSchemaMetadata){
console.log(`Topic ${topicNameOrId} doesn't seem to have a schema.`);
return;
}
constschemaEncoding=topicSchemaMetadata.encoding;
// Make an encoder using the official avro-js library.
constdefinition=fs
.readFileSync('system-test/fixtures/provinces.avsc')
.toString();
consttype=avro.parse(definition);
// Encode the message.
constprovince:ProvinceObject={
name:'Ontario',
post_abbr:'ON',
};
letdataBuffer:Buffer|undefined;
switch(schemaEncoding){
caseEncodings.Binary:
dataBuffer=type.toBuffer(province);
break;
caseEncodings.Json:
dataBuffer=Buffer.from(type.toString(province));
break;
default:
console.log(`Unknown schema encoding: ${schemaEncoding}`);
break;
}
if(!dataBuffer){
console.log(`Invalid encoding ${schemaEncoding} on the topic.`);
return;
}
constmessageId=awaittopic.publish(dataBuffer);
console.log(`Avro record ${messageId} published.`);
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
import{PubSub,Encodings}from'@google-cloud/pubsub';
// And the protobufjs library
import*asprotobuffrom'protobufjs';
// Creates a client; cache this for further use
constpubSubClient=newPubSub();
interfaceProvinceObject{
name:string;
postAbbr:string;
}
asyncfunctionpublishProtobufMessages(topicNameOrId:string){
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
// Get the topic metadata to learn about its schema.
const[topicMetadata]=awaittopic.getMetadata();
consttopicSchemaMetadata=topicMetadata.schemaSettings;
if(!topicSchemaMetadata){
console.log(`Topic ${topicNameOrId} doesn't seem to have a schema.`);
return;
}
constschemaEncoding=topicSchemaMetadata.encoding;
// Encode the message.
constprovince:ProvinceObject={
name:'Ontario',
postAbbr:'ON',
};
// 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.
constroot=awaitprotobuf.load('system-test/fixtures/provinces.proto');
constProvince=root.lookupType('utilities.Province');
constmessage=Province.create(province);
letdataBuffer:Buffer|undefined;
switch(schemaEncoding){
caseEncodings.Binary:
dataBuffer=Buffer.from(Province.encode(message).finish());
break;
caseEncodings.Json:
dataBuffer=Buffer.from(JSON.stringify(message.toJSON()));
break;
default:
console.log(`Unknown schema encoding: ${schemaEncoding}`);
break;
}
if(!dataBuffer){
console.log(`Invalid encoding ${schemaEncoding} on the topic.`);
return;
}
constmessageId=awaittopic.publishMessage({data:dataBuffer});
console.log(`Protobuf message ${messageId} published.`);
}

use Google\Cloud\PubSub\PubSubClient;
use Google\Cloud\PubSub\V1\Encoding;
use AvroStringIO;
use AvroSchema;
use AvroIODatumWriter;
use AvroIOBinaryEncoder;
/**
 * Publish a message using an AVRO schema.
 *
 * This sample uses `wikimedia/avro` for AVRO encoding.
 *
 * @param string $projectId
 * @param string $topicId
 * @param string $definitionFile
 */
function publish_avro_records($projectId, $topicId, $definitionFile)
{
 $pubsub = new PubSubClient([
 'projectId' => $projectId,
 ]);
 $definition = (string) file_get_contents($definitionFile);
 $messageData = [
 'name' => 'Alaska',
 'post_abbr' => 'AK',
 ];
 $topic = $pubsub->topic($topicId);
 // get the encoding type.
 $topicInfo = $topic->info();
 $encoding = '';
 if (isset($topicInfo['schemaSettings']['encoding'])) {
 $encoding = $topicInfo['schemaSettings']['encoding'];
 }
 // if encoding is not set, we can't continue.
 if ($encoding === '') {
 printf('Topic %s does not have schema enabled', $topicId);
 return;
 }
 // If you are using gRPC, encoding may be an integer corresponding to an
 // enum value on Google\Cloud\PubSub\V1\Encoding.
 if (!is_string($encoding)) {
 $encoding = Encoding::name($encoding);
 }
 $encodedMessageData = '';
 if ($encoding == 'BINARY') {
 // encode as AVRO binary.
 $io = new AvroStringIO();
 $schema = AvroSchema::parse($definition);
 $writer = new AvroIODatumWriter($schema);
 $encoder = new AvroIOBinaryEncoder($io);
 $writer->write($messageData, $encoder);
 $encodedMessageData = $io->string();
 } else {
 // encode as JSON.
 $encodedMessageData = json_encode($messageData);
 }
 $topic->publish(['data' => $encodedMessageData]);
 printf('Published message with %s encoding', $encoding);
}

use Google\Cloud\PubSub\PubSubClient;
use Google\Cloud\PubSub\V1\Encoding;
use Utilities\StateProto;
/**
 * Publish a message using a protocol buffer schema.
 *
 * Relies on a proto message of the following form:
 * ```
 * syntax = "proto3";
 *
 * package utilities;
 *
 * message StateProto {
 * string name = 1;
 * string post_abbr = 2;
 * }
 * ```
 *
 * @param string $projectId
 * @param string $topicId
 * @return void
 */
function publish_proto_messages($projectId, $topicId)
{
 $pubsub = new PubSubClient([
 'projectId' => $projectId,
 ]);
 $messageData = new StateProto([
 'name' => 'Alaska',
 'post_abbr' => 'AK',
 ]);
 $topic = $pubsub->topic($topicId);
 // get the encoding type.
 $topicInfo = $topic->info();
 $encoding = '';
 if (isset($topicInfo['schemaSettings']['encoding'])) {
 $encoding = $topicInfo['schemaSettings']['encoding'];
 }
 // if encoding is not set, we can't continue.
 if ($encoding === '') {
 printf('Topic %s does not have schema enabled', $topicId);
 return;
 }
 // If you are using gRPC, encoding may be an integer corresponding to an
 // enum value on Google\Cloud\PubSub\V1\Encoding.
 if (!is_string($encoding)) {
 $encoding = Encoding::name($encoding);
 }
 $encodedMessageData = '';
 if ($encoding == 'BINARY') {
 // encode as protobuf binary.
 $encodedMessageData = $messageData->serializeToString();
 } else {
 // encode as JSON.
 $encodedMessageData = $messageData->serializeToJsonString();
 }
 $topic->publish(['data' => $encodedMessageData]);
 printf('Published message with %s encoding', $encoding);
}

fromavro.ioimport BinaryEncoder, DatumWriter
importavro.schemaasschema
importio
importjson
fromgoogle.api_core.exceptionsimport NotFound
fromgoogle.cloud.pubsubimport PublisherClient
fromgoogle.pubsub_v1.typesimport Encoding
# TODO(developer): Replace these variables before running the sample.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# avsc_file = "path/to/an/avro/schema/file/(.avsc)/formatted/in/json"
publisher_client = PublisherClient()
topic_path = publisher_client .topic_path(project_id, topic_id)
# Prepare to write Avro records to the binary output stream.
with open(avsc_file, "rb") as file:
 avro_schema = schema.parse(file.read())
writer = DatumWriter(avro_schema)
bout = io.BytesIO()
# Prepare some data using a Python dictionary that matches the Avro schema
record = {"name": "Alaska", "post_abbr": "AK"}
try:
 # Get the topic encoding type.
 topic = publisher_client .get_topic(request={"topic": topic_path})
 encoding = topic.schema_settings.encoding
 # Encode the data according to the message serialization type.
 if encoding == Encoding.BINARY:
 encoder = BinaryEncoder(bout)
 writer.write(record, encoder)
 data = bout.getvalue()
 print(f"Preparing a binary-encoded message:\n{data .decode()}")
 elif encoding == Encoding.JSON:
 data_str = json.dumps(record)
 print(f"Preparing a JSON-encoded message:\n{data_str}")
 data = data_str.encode("utf-8")
 else:
 print(f"No encoding specified in {topic_path}. Abort.")
 exit(0)
 future = publisher_client.publish (topic_path, data)
 print(f"Published message ID: {future.result()}")
except NotFound:
 print(f"{topic_id} not found.")

fromgoogle.api_core.exceptionsimport NotFound
fromgoogle.cloud.pubsubimport PublisherClient
fromgoogle.protobuf.json_formatimport MessageToJson
fromgoogle.pubsub_v1.typesimport Encoding
fromutilitiesimport us_states_pb2 # type: ignore
# TODO(developer): Replace these variables before running the sample.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
publisher_client = PublisherClient()
topic_path = publisher_client .topic_path(project_id, topic_id)
try:
 # Get the topic encoding type.
 topic = publisher_client .get_topic(request={"topic": topic_path})
 encoding = topic.schema_settings.encoding
 # Instantiate a protoc-generated class defined in `us-states.proto`.
 state = us_states_pb2.StateProto()
 state.name = "Alaska"
 state.post_abbr = "AK"
 # Encode the data according to the message serialization type.
 if encoding == Encoding.BINARY:
 data = state.SerializeToString()
 print(f"Preparing a binary-encoded message:\n{data }")
 elif encoding == Encoding.JSON:
 json_object = MessageToJson(state)
 data = str(json_object).encode("utf-8")
 print(f"Preparing a JSON-encoded message:\n{data }")
 else:
 print(f"No encoding specified in {topic_path}. Abort.")
 exit(0)
 future = publisher_client.publish (topic_path, data)
 print(f"Published message ID: {future.result()}")
except NotFound:
 print(f"{topic_id} not found.")

# topic_id = "your-topic-id"
# avsc_file = "path/to/an/avro/schema/file/(.avsc)/formatted/in/json"
pubsub=Google::Cloud::PubSub .new
topic_admin=pubsub.topic_admin
publisher=pubsub.publisher topic_id
record={"name"=>"Alaska","post_abbr"=>"AK"}
topic=topic_admin.get_topictopic:pubsub.topic_path(topic_id)
encoding=topic.schema_settings .encoding
caseencoding
when:BINARY
require"avro"
avro_schema=Avro::Schema.parseFile.read(avsc_file)
writer=Avro::IO::DatumWriter.new avro_schema
buffer=StringIO.new
encoder=Avro::IO::BinaryEncoder.new buffer
writer.writerecord,encoder
publisher.publishbuffer
puts"Published binary-encoded AVRO message."
when:JSON
require"json"
publisher.publishrecord.to_json
puts"Published JSON-encoded AVRO message."
else
raise"No encoding specified in #{topic.name}."
end

# topic_id = "your-topic-id"
pubsub=Google::Cloud::PubSub .new
topic_admin=pubsub.topic_admin
publisher=pubsub.publisher topic_id
state=Utilities::StateProto.new name:"Alaska",post_abbr:"AK"
topic=topic_admin.get_topictopic:pubsub.topic_path(topic_id)
encoding=topic.schema_settings .encoding
caseencoding
when:BINARY
publisher.publishUtilities::StateProto.encode(state)
puts"Published binary-encoded protobuf message."
when:JSON
publisher.publishUtilities::StateProto.encode_json(state)
puts"Published JSON-encoded protobuf message."
else
raise"No encoding specified in #{topic.name}."
end