feat(pubsublite): add sample to create a Pub/Sub Lite export subscription

pubsublite/admin/create_pubsub_export_subscription.go

@@ -0,0 +1,66 @@
+// Copyright 2022 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.
+
+package admin
+
+// [START pubsublite_create_pubsub_export_subscription]
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"cloud.google.com/go/pubsublite"
+)
+
+func createPubsubExportSubscription(w io.Writer, projectID, region, location, topicID, subID, pubsubTopicID string) error {
+	// projectID := "my-project-id"
+	// region := "us-central1"
+	// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
+	// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
+	// location := "us-central1"
+	// NOTE: topic and subscription must be in the same region/zone (e.g. "us-central1-a")
+	// topicID := "my-topic"
+	// subID := "my-subscription"
+	// pubsubTopicID := "destination-topic-id"
+	ctx := context.Background()
+	client, err := pubsublite.NewAdminClient(ctx, region)
+	if err != nil {
+		return fmt.Errorf("pubsublite.NewAdminClient: %v", err)
+	}
+	defer client.Close()
+
+	// Initialize the subscription to the oldest retained messages for each
+	// partition.
+	targetLocation := pubsublite.AtTargetLocation(pubsublite.Beginning)
+
+	sub, err := client.CreateSubscription(ctx, pubsublite.SubscriptionConfig{
+		Name:                fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", projectID, location, subID),
+		Topic:               fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID),
+		DeliveryRequirement: pubsublite.DeliverImmediately, // Can also be DeliverAfterStored.
+		// Configures an export subscription that writes messages to a Pub/Sub topic.
+		ExportConfig: &pubsublite.ExportConfig{
+			DesiredState: pubsublite.ExportActive, // Can also be ExportPaused.
+			Destination: &pubsublite.PubSubDestinationConfig{
+				Topic: fmt.Sprintf("projects/%s/topics/%s", projectID, pubsubTopicID),
+			},
+		},
+	}, targetLocation)
+	if err != nil {
+		return fmt.Errorf("client.CreateSubscription got err: %v", err)
+	}
+	fmt.Fprintf(w, "Created export subscription: %s\n", sub.Name)
+	return nil
+}
+
+// [END pubsublite_create_pubsub_export_subscription]

pubsublite/admin/pubsublite_admin_test.go

@@ -25,12 +25,13 @@ import (
 	"testing"
 	"time"
 
+	"cloud.google.com/go/pubsub"
 	"cloud.google.com/go/pubsublite"
 	"github.com/GoogleCloudPlatform/golang-samples/internal/testutil"
 	"github.com/GoogleCloudPlatform/golang-samples/pubsublite/internal/psltest"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
-	"google.golang.org/api/cloudresourcemanager/v1"
+	cloudresourcemanager "google.golang.org/api/cloudresourcemanager/v1"
 )
 
 const (
@@ -202,6 +203,21 @@ func TestSubscriptionAdmin(t *testing.T) {
 	subID := resourcePrefix + uuid.NewString()
 	subPath := fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", projNumber, testZone, subID)
 
+	exportSubID := resourcePrefix + "-export-" + uuid.NewString()
+	exportSubPath := fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", projNumber, testZone, exportSubID)
+
+	// Destination Pub/Sub topic for testing export subscriptions.
+	pubsubClient, err := pubsub.NewClient(ctx, tc.ProjectID)
+	if err != nil {
+		t.Fatalf("failed to create pubsub client: %v", err)
+	}
+	defer pubsubClient.Close()
+	pubsubTopic, err := pubsubClient.CreateTopic(ctx, topicID)
+	if err != nil {
+		t.Fatalf("CreateTopic: %v", err)
+	}
+	defer pubsubTopic.Delete(ctx)
+
 	t.Run("CreateSubscription", func(t *testing.T) {
 		buf := new(bytes.Buffer)
 		err := createSubscription(buf, tc.ProjectID, testRegion, testZone, topicID, subID)
@@ -215,6 +231,19 @@ func TestSubscriptionAdmin(t *testing.T) {
 		}
 	})
 
+	t.Run("CreatePubsubExportSubscription", func(t *testing.T) {
+		buf := new(bytes.Buffer)
+		err := createPubsubExportSubscription(buf, tc.ProjectID, testRegion, testZone, topicID, exportSubID, topicID)
+		if err != nil {
+			t.Fatalf("createPubsubExportSubscription: %v", err)
+		}
+		got := buf.String()
+		want := fmt.Sprintf("Created export subscription: %s\n", exportSubPath)
+		if diff := cmp.Diff(want, got); diff != "" {
+			t.Fatalf("createPubsubExportSubscription() mismatch: -want, +got:\n%s", diff)
+		}
+	})
+
 	t.Run("GetSubscription", func(t *testing.T) {
 		testutil.Retry(t, 3, 5*time.Second, func(r *testutil.R) {
 			buf := new(bytes.Buffer)
@@ -276,6 +305,7 @@ func TestSubscriptionAdmin(t *testing.T) {
 		}
 	})
 
+	client.DeleteSubscription(ctx, exportSubPath)
 	client.DeleteTopic(ctx, topicPath)
 }