Receive messages that could be for different schema revisions
Stay organized with collections
Save and categorize content based on your preferences.
Receive messages that could be for different schema revisions
Explore further
For detailed documentation that includes this code sample, see the following:
Code sample
Go
Before trying this sample, follow the Go setup instructions in the
Pub/Sub quickstart using
client libraries.
For more information, see the
Pub/Sub Go API
reference documentation.
To authenticate to Pub/Sub, set up Application Default Credentials.
For more information, see
Set up authentication for a local development environment.
import (
"context"
"fmt"
"io"
"strings"
"sync"
"time"
"cloud.google.com/go/pubsub"
"github.com/linkedin/goavro/v2"
)
func subscribeWithAvroSchemaRevisions(w io.Writer, projectID, subID, avscFile string) 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)
if err != nil {
return fmt.Errorf("pubsub.NewClient: %w", err)
}
schemaClient, err := pubsub.NewSchemaClient(ctx, projectID)
if err != nil {
return fmt.Errorf("pubsub.NewSchemaClient: %w", err)
}
// Create the cache for the codecs for different revision IDs.
revisionCodecs := make(map[string]*goavro.Codec)
sub := client.Subscription(subID)
ctx2, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
var mu sync.Mutex
sub.Receive(ctx2, func(ctx context.Context, msg *pubsub.Message) {
mu.Lock()
defer mu.Unlock()
name := msg.Attributes["googclient_schemaname"]
revision := msg.Attributes["googclient_schemarevisionid"]
codec, ok := revisionCodecs[revision]
// If the codec doesn't exist in the map, this is the first time we
// are seeing this revision. We need to fetch the schema and cache the
// codec. It would be more typical to do this asynchronously, but is
// shown here in a synchronous way to ease readability.
if !ok {
// Extract just the schema resource name
path := strings.Split(name, "/")
name = path[len(path)-1]
schema, err := schemaClient.Schema(ctx, fmt.Sprintf("%s@%s", name, revision), pubsub.SchemaViewFull)
if err != nil {
fmt.Fprintf(w, "Nacking, cannot read message without schema: %v\n", err)
msg.Nack()
return
}
codec, err = goavro.NewCodec(schema.Definition)
if err != nil {
msg.Nack()
fmt.Fprintf(w, "goavro.NewCodec err: %v\n", err)
}
revisionCodecs[revision] = codec
}
encoding := msg.Attributes["googclient_schemaencoding"]
var state map[string]interface{}
if encoding == "BINARY" {
data, _, err := codec.NativeFromBinary(msg.Data)
if err != nil {
fmt.Fprintf(w, "codec.NativeFromBinary err: %v\n", err)
msg.Nack()
return
}
fmt.Fprintf(w, "Received a binary-encoded message:\n%#v\n", data)
state = data.(map[string]interface{})
} else if encoding == "JSON" {
data, _, err := codec.NativeFromTextual(msg.Data)
if err != nil {
fmt.Fprintf(w, "codec.NativeFromTextual err: %v\n", err)
msg.Nack()
return
}
fmt.Fprintf(w, "Received a JSON-encoded message:\n%#v\n", data)
state = data.(map[string]interface{})
} else {
fmt.Fprintf(w, "Unknown message type(%s), nacking\n", encoding)
msg.Nack()
return
}
fmt.Fprintf(w, "%s is abbreviated as %s\n", state["name"], state["post_abbr"])
msg.Ack()
})
return nil
}
Except as otherwise noted, the content of this page is licensed under the Creative Commons Attribution 4.0 License, and code samples are licensed under the Apache 2.0 License. For details, see the Google Developers Site Policies. Java is a registered trademark of Oracle and/or its affiliates.
[{
"type": "thumb-down",
"id": "hardToUnderstand",
"label":"Hard to understand"
},{
"type": "thumb-down",
"id": "incorrectInformationOrSampleCode",
"label":"Incorrect information or sample code"
},{
"type": "thumb-down",
"id": "missingTheInformationSamplesINeed",
"label":"Missing the information/samples I need"
},{
"type": "thumb-down",
"id": "otherDown",
"label":"Other"
}]
[{
"type": "thumb-up",
"id": "easyToUnderstand",
"label":"Easy to understand"
},{
"type": "thumb-up",
"id": "solvedMyProblem",
"label":"Solved my problem"
},{
"type": "thumb-up",
"id": "otherUp",
"label":"Other"
}]