func cleanup(t *testing.T, prefix string) {
	ctx := context.Background()
	client, err := NewClient(ctx, cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl)))
	if err != nil {
		t.Errorf("Could not create client: %v", err)
	}
	defer client.Close()

	var q *Query = &Query{
		Prefix: prefix,
	}
	for {
		o, err := client.Bucket(bucket).List(ctx, q)
		if err != nil {
			t.Fatalf("Cleanup List for bucket %v failed with error: %v", bucket, err)
		}
		for _, obj := range o.Results {
			t.Logf("Cleanup deletion of %v", obj.Name)
			if err = client.Bucket(bucket).Object(obj.Name).Delete(ctx); err != nil {
				t.Fatalf("Cleanup Delete for object %v failed with %v", obj.Name, err)
			}
		}
		if o.Next == nil {
			break
		}
		q = o.Next
	}
}

func TestAdminClient(t *testing.T) {
	ctx := context.Background()
	projectID := testutil.ProjID()
	newBucket := projectID + "copy"

	client, err := NewAdminClient(ctx, projectID, cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl)))
	if err != nil {
		t.Fatalf("Could not create client: %v", err)
	}
	defer client.Close()

	if err := client.CreateBucket(ctx, newBucket, nil); err != nil {
		t.Errorf("CreateBucket(%v, %v) failed %v", newBucket, nil, err)
	}
	if err := client.DeleteBucket(ctx, newBucket); err != nil {
		t.Errorf("DeleteBucket(%v) failed %v", newBucket, err)
		t.Logf("TODO: Warning this test left a new bucket in the cloud project, it must be deleted manually")
	}
	attrs := BucketAttrs{
		DefaultObjectACL: []ACLRule{{Entity: "domain-google.com", Role: RoleReader}},
	}
	if err := client.CreateBucket(ctx, newBucket, &attrs); err != nil {
		t.Errorf("CreateBucket(%v, %v) failed %v", newBucket, attrs, err)
	}
	if err := client.DeleteBucket(ctx, newBucket); err != nil {
		t.Errorf("DeleteBucket(%v) failed %v", newBucket, err)
		t.Logf("TODO: Warning this test left a new bucket in the cloud project, it must be deleted manually")
	}
}

func TestIntegrationPingBadProject(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}

	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	for _, projID := range []string{
		testutil.ProjID() + "-BAD", // nonexistent project
		"amazing-height-519",       // exists, but wrong creds
	} {
		c, err := NewClient(ctx, projID, "logging-integration-test", cloud.WithTokenSource(ts))
		if err != nil {
			t.Fatalf("project %s: error creating client: %v", projID, err)
		}
		if err := c.Ping(); err == nil {
			t.Errorf("project %s: want error pinging logging api, got nil", projID)
		}
		// Ping twice, just to make sure the deduping doesn't mess with the result.
		if err := c.Ping(); err == nil {
			t.Errorf("project %s: want error pinging logging api, got nil", projID)
		}
	}
}

func newClient(ctx context.Context) *Client {
	ts := testutil.TokenSource(ctx, ScopeDatastore, ScopeUserEmail)
	client, err := NewClient(ctx, testutil.ProjID(), cloud.WithTokenSource(ts))
	if err != nil {
		log.Fatal(err)
	}
	return client
}

func newClient(ctx context.Context, t *testing.T) *Client {
	ts := testutil.TokenSource(ctx, ScopeDatastore)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, testutil.ProjID(), cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("NewClient: %v", err)
	}
	return client
}

// config is like testConfig, but it doesn't need a *testing.T.
func config(ctx context.Context) (*Client, string) {
	ts := testutil.TokenSource(ctx, ScopeFullControl)
	if ts == nil {
		return nil, ""
	}
	p := testutil.ProjID()
	if p == "" {
		log.Fatal("The project ID must be set. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, cloud.WithTokenSource(ts))
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}
	return client, p
}

// testConfig returns the Client used to access GCS and the default bucket
// name to use.
func testConfig(ctx context.Context, t *testing.T) (*Client, string) {
	ts := cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl))
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	p := testutil.ProjID()
	if p == "" {
		log.Fatal("The project ID must be set. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, ts)
	if err != nil {
		t.Fatalf("NewClient: %v", err)
	}
	return client, p
}

func TestValidObjectNames(t *testing.T) {
	ctx := context.Background()
	client, err := NewClient(ctx, cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl)))
	if err != nil {
		t.Errorf("Could not create client: %v", err)
	}
	defer client.Close()

	bkt := client.Bucket(bucket)

	validNames := []string{
		"gopher",
		"Гоферови",
		"a",
		strings.Repeat("a", 1024),
	}
	for _, name := range validNames {
		w := bkt.Object(name).NewWriter(ctx)
		if _, err := w.Write([]byte("data")); err != nil {
			t.Errorf("Object %q write failed: %v. Want success", name, err)
			continue
		}
		if err := w.Close(); err != nil {
			t.Errorf("Object %q close failed: %v. Want success", name, err)
			continue
		}
		defer bkt.Object(name).Delete(ctx)
	}

	invalidNames := []string{
		"", // Too short.
		strings.Repeat("a", 1025), // Too long.
		"new\nlines",
		"bad\xffunicode",
	}
	for _, name := range invalidNames {
		w := bkt.Object(name).NewWriter(ctx)
		// Invalid object names will either cause failure during Write or Close.
		if _, err := w.Write([]byte("data")); err != nil {
			continue
		}
		if err := w.Close(); err != nil {
			continue
		}
		defer bkt.Object(name).Delete(ctx)
		t.Errorf("%q should have failed. Didn't", name)
	}
}

func TestIntegration(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}

	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	projID := testutil.ProjID()

	c, err := NewClient(ctx, projID, "logging-integration-test", cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("error creating client: %v", err)
	}

	if err := c.Ping(); err != nil {
		t.Fatalf("error pinging logging api: %v", err)
	}
	// Ping twice, to verify that deduping doesn't change the result.
	if err := c.Ping(); err != nil {
		t.Fatalf("error pinging logging api: %v", err)
	}

	if err := c.LogSync(Entry{Payload: customJSONObject{}}); err != nil {
		t.Fatalf("error writing log: %v", err)
	}

	if err := c.Log(Entry{Payload: customJSONObject{}}); err != nil {
		t.Fatalf("error writing log: %v", err)
	}

	if _, err := c.Writer(Default).Write([]byte("test log with io.Writer")); err != nil {
		t.Fatalf("error writing log using io.Writer: %v", err)
	}

	c.Logger(Default).Println("test log with log.Logger")

	if err := c.Flush(); err != nil {
		t.Fatalf("error flushing logs: %v", err)
	}
}

func TestAdminClient(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	ctx := context.Background()
	ts := testutil.TokenSource(ctx, ScopeFullControl)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	projectID := testutil.ProjID()

	newBucket := projectID + suffix
	t.Logf("Testing admin with Bucket %q", newBucket)

	client, err := NewAdminClient(ctx, projectID, cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("Could not create client: %v", err)
	}
	defer client.Close()

	if err := client.CreateBucket(ctx, newBucket, nil); err != nil {
		t.Errorf("CreateBucket(%v, %v) failed %v", newBucket, nil, err)
	}
	if err := client.DeleteBucket(ctx, newBucket); err != nil {
		t.Errorf("DeleteBucket(%v) failed %v", newBucket, err)
		t.Logf("TODO: Warning this test left a new bucket in the cloud project, it must be deleted manually")
	}
	attrs := BucketAttrs{
		DefaultObjectACL: []ACLRule{{Entity: "domain-google.com", Role: RoleReader}},
	}
	if err := client.CreateBucket(ctx, newBucket, &attrs); err != nil {
		t.Errorf("CreateBucket(%v, %v) failed %v", newBucket, attrs, err)
	}
	if err := client.DeleteBucket(ctx, newBucket); err != nil {
		t.Errorf("DeleteBucket(%v) failed %v", newBucket, err)
		t.Logf("TODO: Warning this test left a new bucket in the cloud project, it must be deleted manually")
	}
}

func TestAll(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	// TODO(djd): Replace this ctx with context.Background() when the new API is complete.
	ctx := testutil.Context(ScopePubSub, ScopeCloudPlatform)
	if ctx == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	now := time.Now()
	topicName := fmt.Sprintf("topic-%d", now.Unix())
	subName := fmt.Sprintf("subscription-%d", now.Unix())

	client, err := NewClient(ctx, testutil.ProjID(), cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("Creating client error: %v", err)
	}

	var topic *TopicHandle
	if topic, err = client.NewTopic(ctx, topicName); err != nil {
		t.Errorf("CreateTopic error: %v", err)
	}

	var sub *SubscriptionHandle
	if sub, err = topic.Subscribe(ctx, subName, nil); err != nil {
		t.Errorf("CreateSub error: %v", err)
	}

	exists, err := topic.Exists(ctx)
	if err != nil {
		t.Fatalf("TopicExists error: %v", err)
	}
	if !exists {
		t.Errorf("topic %s should exist, but it doesn't", topic)
	}

	exists, err = sub.Exists(ctx)
	if err != nil {
		t.Fatalf("SubExists error: %v", err)
	}
	if !exists {
		t.Errorf("subscription %s should exist, but it doesn't", subName)
	}

	max := 10
	msgs := make([]*Message, max)
	expectedMsgs := make(map[string]bool, max)
	for i := 0; i < max; i++ {
		text := fmt.Sprintf("a message with an index %d", i)
		attrs := make(map[string]string)
		attrs["foo"] = "bar"
		msgs[i] = &Message{
			Data:       []byte(text),
			Attributes: attrs,
		}
		expectedMsgs[text] = false
	}

	ids, err := Publish(ctx, topicName, msgs...)
	if err != nil {
		t.Fatalf("Publish (1) error: %v", err)
	}

	if len(ids) != max {
		t.Errorf("unexpected number of message IDs received; %d, want %d", len(ids), max)
	}

	expectedIDs := make(map[string]bool, max)
	for _, id := range ids {
		expectedIDs[id] = false
	}

	received, err := PullWait(ctx, subName, max)
	if err != nil {
		t.Fatalf("PullWait error: %v", err)
	}

	if len(received) != max {
		t.Errorf("unexpected number of messages received; %d, want %d", len(received), max)
	}

	for _, msg := range received {
		expectedMsgs[string(msg.Data)] = true
		expectedIDs[msg.ID] = true
		if msg.Attributes["foo"] != "bar" {
			t.Errorf("message attribute foo is expected to be 'bar', found '%s'", msg.Attributes["foo"])
		}
	}

	for msg, found := range expectedMsgs {
		if !found {
			t.Errorf("message '%s' should be received", msg)
		}
	}

//.........这里部分代码省略.........

func TestObjects(t *testing.T) {
	ctx := context.Background()
	client, err := NewClient(ctx, cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl)))
	if err != nil {
		t.Errorf("Could not create client: %v", err)
	}
	defer client.Close()

	bkt := client.Bucket(bucket)

	// Cleanup.
	cleanup(t, "obj")

	const defaultType = "text/plain"

	// Test Writer.
	for _, obj := range objects {
		t.Logf("Writing %v", obj)
		wc := bkt.Object(obj).NewWriter(ctx)
		wc.ContentType = defaultType
		c := randomContents()
		if _, err := wc.Write(c); err != nil {
			t.Errorf("Write for %v failed with %v", obj, err)
		}
		if err := wc.Close(); err != nil {
			t.Errorf("Close for %v failed with %v", obj, err)
		}
		contents[obj] = c
	}

	// Test Reader.
	for _, obj := range objects {
		t.Logf("Creating a reader to read %v", obj)
		rc, err := bkt.Object(obj).NewReader(ctx)
		if err != nil {
			t.Errorf("Can't create a reader for %v, errored with %v", obj, err)
		}
		slurp, err := ioutil.ReadAll(rc)
		if err != nil {
			t.Errorf("Can't ReadAll object %v, errored with %v", obj, err)
		}
		if got, want := slurp, contents[obj]; !bytes.Equal(got, want) {
			t.Errorf("Contents (%q) = %q; want %q", obj, got, want)
		}
		if got, want := rc.Size(), len(contents[obj]); got != int64(want) {
			t.Errorf("Size (%q) = %d; want %d", obj, got, want)
		}
		if got, want := rc.ContentType(), "text/plain"; got != want {
			t.Errorf("ContentType (%q) = %q; want %q", obj, got, want)
		}
		rc.Close()

		// Test SignedURL
		opts := &SignedURLOptions{
			GoogleAccessID: "[email protected]",
			PrivateKey:     dummyKey("rsa"),
			Method:         "GET",
			MD5:            []byte("202cb962ac59075b964b07152d234b70"),
			Expires:        time.Date(2020, time.October, 2, 10, 0, 0, 0, time.UTC),
			ContentType:    "application/json",
			Headers:        []string{"x-header1", "x-header2"},
		}
		u, err := SignedURL(bucket, obj, opts)
		if err != nil {
			t.Fatalf("SignedURL(%q, %q) errored with %v", bucket, obj, err)
		}
		res, err := client.hc.Get(u)
		if err != nil {
			t.Fatalf("Can't get URL %q: %v", u, err)
		}
		slurp, err = ioutil.ReadAll(res.Body)
		if err != nil {
			t.Fatalf("Can't ReadAll signed object %v, errored with %v", obj, err)
		}
		if got, want := slurp, contents[obj]; !bytes.Equal(got, want) {
			t.Errorf("Contents (%v) = %q; want %q", obj, got, want)
		}
		res.Body.Close()
	}

	// Test NotFound.
	_, err = bkt.Object("obj-not-exists").NewReader(ctx)
	if err != ErrObjectNotExist {
		t.Errorf("Object should not exist, err found to be %v", err)
	}

	name := objects[0]

	// Test StatObject.
	o, err := bkt.Object(name).Attrs(ctx)
	if err != nil {
		t.Error(err)
	}
	if got, want := o.Name, name; got != want {
		t.Errorf("Name (%v) = %q; want %q", name, got, want)
	}
	if got, want := o.ContentType, defaultType; got != want {
		t.Errorf("ContentType (%v) = %q; want %q", name, got, want)
	}

//.........这里部分代码省略.........

func TestACL(t *testing.T) {
	ctx := context.Background()
	client, err := NewClient(ctx, cloud.WithTokenSource(testutil.TokenSource(ctx, ScopeFullControl)))
	if err != nil {
		t.Errorf("Could not create client: %v", err)
	}
	defer client.Close()

	bkt := client.Bucket(bucket)

	cleanup(t, "acl")
	entity := ACLEntity("domain-google.com")
	if err := client.Bucket(bucket).DefaultObjectACL().Set(ctx, entity, RoleReader); err != nil {
		t.Errorf("Can't put default ACL rule for the bucket, errored with %v", err)
	}
	for _, obj := range aclObjects {
		t.Logf("Writing %v", obj)
		wc := bkt.Object(obj).NewWriter(ctx)
		c := randomContents()
		if _, err := wc.Write(c); err != nil {
			t.Errorf("Write for %v failed with %v", obj, err)
		}
		if err := wc.Close(); err != nil {
			t.Errorf("Close for %v failed with %v", obj, err)
		}
	}
	name := aclObjects[0]
	o := bkt.Object(name)
	acl, err := o.ACL().List(ctx)
	if err != nil {
		t.Errorf("Can't retrieve ACL of %v", name)
	}
	aclFound := false
	for _, rule := range acl {
		if rule.Entity == entity && rule.Role == RoleReader {
			aclFound = true
		}
	}
	if !aclFound {
		t.Error("Expected to find an ACL rule for google.com domain users, but not found")
	}
	if err := o.ACL().Delete(ctx, entity); err != nil {
		t.Errorf("Can't delete the ACL rule for the entity: %v", entity)
	}

	if err := bkt.ACL().Set(ctx, "[email protected]", RoleReader); err != nil {
		t.Errorf("Error while putting bucket ACL rule: %v", err)
	}
	bACL, err := bkt.ACL().List(ctx)
	if err != nil {
		t.Errorf("Error while getting the ACL of the bucket: %v", err)
	}
	bACLFound := false
	for _, rule := range bACL {
		if rule.Entity == "[email protected]" && rule.Role == RoleReader {
			bACLFound = true
		}
	}
	if !bACLFound {
		t.Error("Expected to find an ACL rule for [email protected] user, but not found")
	}
	if err := bkt.ACL().Delete(ctx, "[email protected]"); err != nil {
		t.Errorf("Error while deleting bucket ACL rule: %v", err)
	}
}

func TestAll(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	ctx := context.Background()
	ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	now := time.Now()
	topicName := fmt.Sprintf("topic-%d", now.Unix())
	subName := fmt.Sprintf("subscription-%d", now.Unix())

	client, err := NewClient(ctx, testutil.ProjID(), cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("Creating client error: %v", err)
	}

	var topic *Topic
	if topic, err = client.NewTopic(ctx, topicName); err != nil {
		t.Errorf("CreateTopic error: %v", err)
	}

	var sub *Subscription
	if sub, err = client.NewSubscription(ctx, subName, topic, 0, nil); err != nil {
		t.Errorf("CreateSub error: %v", err)
	}

	exists, err := topic.Exists(ctx)
	if err != nil {
		t.Fatalf("TopicExists error: %v", err)
	}
	if !exists {
		t.Errorf("topic %s should exist, but it doesn't", topic)
	}

	exists, err = sub.Exists(ctx)
	if err != nil {
		t.Fatalf("SubExists error: %v", err)
	}
	if !exists {
		t.Errorf("subscription %s should exist, but it doesn't", subName)
	}

	msgs := []*Message{}
	for i := 0; i < 10; i++ {
		text := fmt.Sprintf("a message with an index %d", i)
		attrs := make(map[string]string)
		attrs["foo"] = "bar"
		msgs = append(msgs, &Message{
			Data:       []byte(text),
			Attributes: attrs,
		})
	}

	ids, err := topic.Publish(ctx, msgs...)
	if err != nil {
		t.Fatalf("Publish (1) error: %v", err)
	}

	if len(ids) != len(msgs) {
		t.Errorf("unexpected number of message IDs received; %d, want %d", len(ids), len(msgs))
	}

	want := make(map[string]*messageData)
	for i, m := range msgs {
		md := extractMessageData(m)
		md.ID = ids[i]
		want[md.ID] = md
	}

	// Use a timeout to ensure that Pull does not block indefinitely if there are unexpectedly few messages available.
	timeoutCtx, _ := context.WithTimeout(ctx, time.Minute)
	it, err := sub.Pull(timeoutCtx)
	if err != nil {
		t.Fatalf("error constructing iterator: %v", err)
	}
	defer it.Stop()
	got := make(map[string]*messageData)
	for i := 0; i < len(want); i++ {
		m, err := it.Next()
		if err != nil {
			t.Fatalf("error getting next message: %v", err)
		}
		md := extractMessageData(m)
		got[md.ID] = md
		m.Done(true)
	}

	if !reflect.DeepEqual(got, want) {
		t.Errorf("messages: got: %v ; want: %v", got, want)
	}

	// base64 test
	data := "[email protected]~"
	_, err = topic.Publish(ctx, &Message{Data: []byte(data)})
	if err != nil {
		t.Fatalf("Publish error: %v", err)
	}
//.........这里部分代码省略.........

// TestEndToEnd pumps many messages into a topic and tests that they are all delivered to each subscription for the topic.
// It also tests that messages are not unexpectedly redelivered.
func TestEndToEnd(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	ctx := context.Background()
	ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	now := time.Now()
	topicName := fmt.Sprintf("endtoend-%d", now.Unix())
	subPrefix := fmt.Sprintf("endtoend-%d", now.Unix())

	client, err := NewClient(ctx, testutil.ProjID(), cloud.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("Creating client error: %v", err)
	}

	var topic *Topic
	if topic, err = client.NewTopic(ctx, topicName); err != nil {
		t.Fatalf("CreateTopic error: %v", err)
	}
	defer topic.Delete(ctx)

	// Three subscriptions to the same topic.
	var subA, subB, subC *Subscription
	if subA, err = client.NewSubscription(ctx, subPrefix+"-a", topic, ackDeadline, nil); err != nil {
		t.Fatalf("CreateSub error: %v", err)
	}
	defer subA.Delete(ctx)

	if subB, err = client.NewSubscription(ctx, subPrefix+"-b", topic, ackDeadline, nil); err != nil {
		t.Fatalf("CreateSub error: %v", err)
	}
	defer subB.Delete(ctx)

	if subC, err = client.NewSubscription(ctx, subPrefix+"-c", topic, ackDeadline, nil); err != nil {
		t.Fatalf("CreateSub error: %v", err)
	}
	defer subC.Delete(ctx)

	expectedCounts := make(map[string]int)
	for _, id := range publish(t, ctx, topic) {
		expectedCounts[id] = 1
	}

	// recv provides an indication that messages are still arriving.
	recv := make(chan struct{})

	// Keep track of the number of times each message (by message id) was
	// seen from each subscription.
	mcA := &messageCounter{counts: make(map[string]int), recv: recv}
	mcB := &messageCounter{counts: make(map[string]int), recv: recv}
	mcC := &messageCounter{counts: make(map[string]int), recv: recv}

	stopC := make(chan struct{})

	// We have three subscriptions to our topic.
	// Each subscription will get a copy of each pulished message.
	//
	// subA has just one iterator, while subB has two. The subB iterators
	// will each process roughly half of the messages for subB. All of
	// these iterators live until all messages have been consumed.  subC is
	// processed by a series of short-lived iterators.

	var wg sync.WaitGroup

	con := &consumer{
		concurrencyPerIterator: 1,
		iteratorsInFlight:      2,
		lifetimes:              immortal,
	}
	con.consume(t, ctx, subA, mcA, &wg, stopC)

	con = &consumer{
		concurrencyPerIterator: 1,
		iteratorsInFlight:      2,
		lifetimes:              immortal,
	}
	con.consume(t, ctx, subB, mcB, &wg, stopC)

	con = &consumer{
		concurrencyPerIterator: 1,
		iteratorsInFlight:      2,
		lifetimes: &explicitLifetimes{
			lifetimes: []time.Duration{ackDeadline, ackDeadline, ackDeadline / 2, ackDeadline / 2},
		},
	}
	con.consume(t, ctx, subC, mcC, &wg, stopC)

	go func() {
		timeoutC := time.After(timeout)
		// Every time this ticker ticks, we will check if we have received any
		// messages since the last time it ticked.  We check less frequently
		// than the ack deadline, so that we can detect if messages are
		// redelivered after having their ack deadline extended.
		checkQuiescence := time.NewTicker(ackDeadline * 3)
//.........这里部分代码省略.........