pubsub_test.go

Documentation: cloud.google.com/go/pubsub

     1  // Copyright 2021 Google LLC
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License");
     4  // you may not use this file except in compliance with the License.
     5  // You may obtain a copy of the License at
     6  //
     7  //      http://www.apache.org/licenses/LICENSE-2.0
     8  //
     9  // Unless required by applicable law or agreed to in writing, software
    10  // distributed under the License is distributed on an "AS IS" BASIS,
    11  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12  // See the License for the specific language governing permissions and
    13  // limitations under the License.
    14  
    15  package pubsub
    16  
    17  import (
    18  	"context"
    19  	"testing"
    20  	"time"
    21  
    22  	vkit "cloud.google.com/go/pubsub/apiv1"
    23  	pb "cloud.google.com/go/pubsub/apiv1/pubsubpb"
    24  	"cloud.google.com/go/pubsub/pstest"
    25  	"github.com/googleapis/gax-go/v2"
    26  	"google.golang.org/api/option"
    27  	"google.golang.org/grpc"
    28  	"google.golang.org/grpc/codes"
    29  	"google.golang.org/grpc/status"
    30  )
    31  
    32  func TestClient_ApplyClientConfig(t *testing.T) {
    33  	ctx := context.Background()
    34  	srv := pstest.NewServer()
    35  	// Add a retry for an obscure error.
    36  	pco := &vkit.PublisherCallOptions{
    37  		Publish: []gax.CallOption{
    38  			gax.WithRetry(func() gax.Retryer {
    39  				return gax.OnCodes([]codes.Code{
    40  					codes.DataLoss,
    41  				}, gax.Backoff{
    42  					Initial:    200 * time.Millisecond,
    43  					Max:        30000 * time.Millisecond,
    44  					Multiplier: 1.25,
    45  				})
    46  			}),
    47  		},
    48  	}
    49  	c, err := NewClientWithConfig(ctx, "P", &ClientConfig{
    50  		PublisherCallOptions: pco,
    51  	},
    52  		option.WithEndpoint(srv.Addr),
    53  		option.WithoutAuthentication(),
    54  		option.WithGRPCDialOption(grpc.WithInsecure()))
    55  	if err != nil {
    56  		t.Fatal(err)
    57  	}
    58  
    59  	srv.SetAutoPublishResponse(false)
    60  	// Create a fake publish response with the obscure error we are retrying.
    61  	srv.AddPublishResponse(&pb.PublishResponse{
    62  		MessageIds: []string{},
    63  	}, status.Errorf(codes.DataLoss, "obscure error"))
    64  
    65  	srv.AddPublishResponse(&pb.PublishResponse{
    66  		MessageIds: []string{"1"},
    67  	}, nil)
    68  
    69  	topic, err := c.CreateTopic(ctx, "t")
    70  	if err != nil {
    71  		t.Fatal(err)
    72  	}
    73  	res := topic.Publish(ctx, &Message{
    74  		Data: []byte("test"),
    75  	})
    76  	if id, err := res.Get(ctx); err != nil {
    77  		t.Fatalf("got error from res.Get(): %v", err)
    78  	} else {
    79  		if id != "1" {
    80  			t.Fatalf("got wrong message id from server, got %s, want 1", id)
    81  		}
    82  	}
    83  }
    84  
    85  func TestClient_EmptyProjectID(t *testing.T) {
    86  	ctx := context.Background()
    87  	_, err := NewClient(ctx, "")
    88  	if err != ErrEmptyProjectID {
    89  		t.Fatalf("passing empty project ID got %v, want%v", err, ErrEmptyProjectID)
    90  	}
    91  }
    92
View as plain text