Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(pubsublite): settings and message transforms for Cloud Pub/Sub s…
…him (#3281) The shim implements various features in the Pub/Sub Lite Java client library for feature parity, e.g. nack handler, custom message transforms, key extractor, etc. It also implements some features from the Cloud Pub/Sub Go library, e.g. publish buffer byte limit.
- Loading branch information
Showing
8 changed files
with
716 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| // Copyright 2020 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 | ||
|
|
||
| /* | ||
| Package ps contains clients for publishing and subscribing using the Google | ||
| Cloud Pub/Sub Lite service. | ||
| If interfaces are defined, PublisherClient and SubscriberClient can be used as | ||
| substitutions for pubsub.Topic.Publish() and pubsub.Subscription.Receive(), | ||
| respectively, from the pubsub package. | ||
| As noted in comments, the two services have some differences: | ||
| - Pub/Sub Lite does not support nack for messages. By default, this will | ||
| terminate the SubscriberClient. A custom function can be provided for | ||
| ReceiveSettings.NackHandler to handle nacked messages. | ||
| - Pub/Sub Lite has no concept of ack expiration. Subscribers must ack or nack | ||
| every message received. | ||
| - Pub/Sub Lite PublisherClients can terminate when an unretryable error | ||
| occurs. | ||
| - Publishers and subscribers will be throttled if Pub/Sub Lite publish or | ||
| subscribe throughput limits are exceeded. Thus publishing can be more | ||
| sensitive to buffer overflow than Cloud Pub/Sub. | ||
| More information about Google Cloud Pub/Sub Lite is available at | ||
| https://cloud.google.com/pubsub/lite. | ||
| Information about choosing between Google Cloud Pub/Sub vs Pub/Sub Lite is | ||
| available at https://cloud.google.com/pubsub/docs/choosing-pubsub-or-lite. | ||
| See https://godoc.org/cloud.google.com/go for authentication, timeouts, | ||
| connection pooling and similar aspects of this package. | ||
| */ | ||
| package ps // import "cloud.google.com/go/pubsublite/ps" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| // Copyright 2020 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 | ||
|
|
||
| package ps | ||
|
|
||
| import ( | ||
| "encoding/base64" | ||
| "errors" | ||
| "fmt" | ||
|
|
||
| "cloud.google.com/go/pubsub" | ||
| "github.com/golang/protobuf/ptypes" | ||
| "google.golang.org/protobuf/proto" | ||
|
|
||
| tspb "github.com/golang/protobuf/ptypes/timestamp" | ||
| pb "google.golang.org/genproto/googleapis/cloud/pubsublite/v1" | ||
| ) | ||
|
|
||
| // Message transforms and event timestamp encoding mirrors the Java client | ||
| // library implementation: | ||
| // https://github.com/googleapis/java-pubsublite/blob/master/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/cloudpubsub/MessageTransforms.java | ||
| const eventTimestampAttributeKey = "x-goog-pubsublite-event-time-timestamp-proto" | ||
|
|
||
| var errInvalidMessage = errors.New("pubsublite: invalid received message") | ||
|
|
||
| // Encodes a timestamp in a way that it will be interpreted as an event time if | ||
| // published on a message with an attribute named eventTimestampAttributeKey. | ||
| func encodeEventTimestamp(eventTime *tspb.Timestamp) (string, error) { | ||
| bytes, err := proto.Marshal(eventTime) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
| return base64.StdEncoding.EncodeToString(bytes), nil | ||
| } | ||
|
|
||
| // Decodes a timestamp encoded with encodeEventTimestamp. | ||
| func decodeEventTimestamp(value string) (*tspb.Timestamp, error) { | ||
| bytes, err := base64.StdEncoding.DecodeString(value) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| eventTime := &tspb.Timestamp{} | ||
| if err := proto.Unmarshal(bytes, eventTime); err != nil { | ||
| return nil, err | ||
| } | ||
| return eventTime, nil | ||
| } | ||
|
|
||
| // extractOrderingKey extracts the ordering key from the message for routing | ||
| // during publishing. It is the default KeyExtractorFunc implementation. | ||
| func extractOrderingKey(msg *pubsub.Message) []byte { | ||
| if len(msg.OrderingKey) == 0 { | ||
| return nil | ||
| } | ||
| return []byte(msg.OrderingKey) | ||
| } | ||
|
|
||
| // transformPublishedMessage is the default PublishMessageTransformerFunc | ||
| // implementation. | ||
| func transformPublishedMessage(from *pubsub.Message, to *pb.PubSubMessage, extractKey KeyExtractorFunc) error { | ||
| to.Data = from.Data | ||
| to.Key = extractKey(from) | ||
|
|
||
| if len(from.Attributes) > 0 { | ||
| to.Attributes = make(map[string]*pb.AttributeValues) | ||
| for key, value := range from.Attributes { | ||
| if key == eventTimestampAttributeKey { | ||
| eventpb, err := decodeEventTimestamp(value) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| to.EventTime = eventpb | ||
| } else { | ||
| to.Attributes[key] = &pb.AttributeValues{Values: [][]byte{[]byte(value)}} | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // transformReceivedMessage is the default ReceiveMessageTransformerFunc | ||
| // implementation. | ||
| func transformReceivedMessage(from *pb.SequencedMessage, to *pubsub.Message) error { | ||
| if from == nil || from.GetMessage() == nil { | ||
| // This should not occur, but guard against nil. | ||
| return errInvalidMessage | ||
| } | ||
|
|
||
| var err error | ||
| msg := from.GetMessage() | ||
|
|
||
| if from.GetPublishTime() != nil { | ||
| if to.PublishTime, err = ptypes.Timestamp(from.GetPublishTime()); err != nil { | ||
| return fmt.Errorf("%s: %s", errInvalidMessage.Error(), err) | ||
| } | ||
| } | ||
| if from.GetCursor() != nil { | ||
| to.ID = fmt.Sprintf("%d", from.GetCursor().GetOffset()) | ||
| } | ||
| if len(msg.GetKey()) > 0 { | ||
| to.OrderingKey = string(msg.GetKey()) | ||
| } | ||
| to.Data = msg.GetData() | ||
| to.Attributes = make(map[string]string) | ||
|
|
||
| if msg.EventTime != nil { | ||
| val, err := encodeEventTimestamp(msg.EventTime) | ||
| if err != nil { | ||
| return fmt.Errorf("%s: %s", errInvalidMessage.Error(), err) | ||
| } | ||
| to.Attributes[eventTimestampAttributeKey] = val | ||
| } | ||
| for key, values := range msg.Attributes { | ||
| if key == eventTimestampAttributeKey { | ||
| return fmt.Errorf("%s: attribute with reserved key %q exists in API message", errInvalidMessage.Error(), eventTimestampAttributeKey) | ||
| } | ||
| if len(values.Values) > 1 { | ||
| return fmt.Errorf("%s: cannot transform API message with multiple values for attribute with key %q", errInvalidMessage.Error(), key) | ||
| } | ||
| to.Attributes[key] = string(values.Values[0]) | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.