feat(pubsublite): adding ability to create subscriptions at head

pubsublite/admin.go

@@ -147,9 +147,28 @@ func (ac *AdminClient) Topics(ctx context.Context, parent string) *TopicIterator
 	}
 }
 
+// StartingOffset is the offset at which a newly created subscription will start
+// receiving messages.
+type StartingOffset int
+
+const (
+	// End represents the current HEAD offset.
+	End StartingOffset = iota
+
+	// Beginning represents the offset of the oldest retained message.
+	Beginning
+)
+
 // CreateSubscription creates a new subscription from the given config. If the
 // subscription already exists an error will be returned.
 func (ac *AdminClient) CreateSubscription(ctx context.Context, config SubscriptionConfig) (*SubscriptionConfig, error) {
+	return ac.CreateSubscriptionAtOffset(ctx, config, End)
+}
+
+// CreateSubscriptionAtOffset creates a new subscription from the given config
+// at the provided starting offset. If the subscription already exists an error
+// will be returned.
+func (ac *AdminClient) CreateSubscriptionAtOffset(ctx context.Context, config SubscriptionConfig, startingOffset StartingOffset) (*SubscriptionConfig, error) {
 	subsPath, err := wire.ParseSubscriptionPath(config.Name)
 	if err != nil {
 		return nil, err
@@ -161,6 +180,7 @@ func (ac *AdminClient) CreateSubscription(ctx context.Context, config Subscripti
 		Parent:         subsPath.Location().String(),
 		Subscription:   config.toProto(),
 		SubscriptionId: subsPath.SubscriptionID,
+		SkipBacklog:    startingOffset == End,
 	}
 	subspb, err := ac.admin.CreateSubscription(ctx, req)
 	if err != nil {

pubsublite/admin_test.go

@@ -275,6 +275,13 @@ func TestAdminSubscriptionCRUD(t *testing.T) {
 		Parent:         "projects/my-proj/locations/us-central1-a",
 		SubscriptionId: "my-subscription",
 		Subscription:   subscriptionConfig.toProto(),
+		SkipBacklog:    true,
+	}
+	wantCreateAtBacklogReq := &pb.CreateSubscriptionRequest{
+		Parent:         "projects/my-proj/locations/us-central1-a",
+		SubscriptionId: "my-subscription",
+		Subscription:   subscriptionConfig.toProto(),
+		SkipBacklog:    false,
 	}
 	wantUpdateReq := updateConfig.toUpdateRequest()
 	wantGetReq := &pb.GetSubscriptionRequest{
@@ -286,6 +293,7 @@ func TestAdminSubscriptionCRUD(t *testing.T) {
 
 	verifiers := test.NewVerifiers(t)
 	verifiers.GlobalVerifier.Push(wantCreateReq, subscriptionConfig.toProto(), nil)
+	verifiers.GlobalVerifier.Push(wantCreateAtBacklogReq, subscriptionConfig.toProto(), nil)
 	verifiers.GlobalVerifier.Push(wantUpdateReq, subscriptionConfig.toProto(), nil)
 	verifiers.GlobalVerifier.Push(wantGetReq, subscriptionConfig.toProto(), nil)
 	verifiers.GlobalVerifier.Push(wantDeleteReq, &emptypb.Empty{}, nil)
@@ -301,6 +309,12 @@ func TestAdminSubscriptionCRUD(t *testing.T) {
 		t.Errorf("CreateSubscription() got: %v\nwant: %v", gotConfig, subscriptionConfig)
 	}
 
+	if gotConfig, err := admin.CreateSubscriptionAtOffset(ctx, subscriptionConfig, Beginning); err != nil {
+		t.Errorf("CreateSubscription() got err: %v", err)
+	} else if !testutil.Equal(gotConfig, &subscriptionConfig) {
+		t.Errorf("CreateSubscription() got: %v\nwant: %v", gotConfig, subscriptionConfig)
+	}
+
 	if gotConfig, err := admin.UpdateSubscription(ctx, updateConfig); err != nil {
 		t.Errorf("UpdateSubscription() got err: %v", err)
 	} else if !testutil.Equal(gotConfig, &subscriptionConfig) {