func annotationsFromProto(res *pb.AnnotateImageResponse) *Annotations {
	as := &Annotations{}
	for _, a := range res.FaceAnnotations {
		as.Faces = append(as.Faces, faceAnnotationFromProto(a))
	}
	for _, a := range res.LandmarkAnnotations {
		as.Landmarks = append(as.Landmarks, entityAnnotationFromProto(a))
	}
	for _, a := range res.LogoAnnotations {
		as.Logos = append(as.Logos, entityAnnotationFromProto(a))
	}
	for _, a := range res.LabelAnnotations {
		as.Labels = append(as.Labels, entityAnnotationFromProto(a))
	}
	for _, a := range res.TextAnnotations {
		as.Texts = append(as.Texts, entityAnnotationFromProto(a))
	}
	as.SafeSearch = safeSearchAnnotationFromProto(res.SafeSearchAnnotation)
	as.ImageProps = imagePropertiesFromProto(res.ImagePropertiesAnnotation)
	if res.Error != nil {
		// res.Error is a google.rpc.Status. Convert to a Go error. Use a gRPC
		// error because it preserves the code as a separate field.
		// TODO(jba): preserve the details field.
		as.Error = grpc.Errorf(codes.Code(res.Error.Code), "%s", res.Error.Message)
	}
	return as
}

// Poll fetches the latest state of a long-running operation.
//
// If Poll fails, the error is returned and op is unmodified.
// If Poll succeeds and the operation has completed with failure,
// the error is returned and op.Done will return true.
// If Poll succeeds and the operation has completed successfully,
// op.Done will return true; if resp != nil, the response of the operation
// is stored in resp.
func (op *Operation) Poll(ctx context.Context, resp proto.Message) error {
	if !op.Done() {
		p, err := op.c.GetOperation(ctx, &pb.GetOperationRequest{Name: op.Name()})
		if err != nil {
			return err
		}
		op.proto = p
	}
	if !op.Done() {
		return nil
	}

	switch r := op.proto.Result.(type) {
	case *pb.Operation_Error:
		// TODO (pongad): r.Details may contain further information
		return grpc.Errorf(codes.Code(r.Error.Code), "%s", r.Error.Message)
	case *pb.Operation_Response:
		if resp == nil {
			return nil
		}
		return ptypes.UnmarshalAny(r.Response, resp)
	default:
		return fmt.Errorf("unsupported result type %[1]T: %[1]v", r)
	}
}

// DoStatusCodeAndMessage checks that the status code is propagated back to the client.
func DoStatusCodeAndMessage(tc testpb.TestServiceClient) {
	var code int32 = 2
	msg := "test status message"
	expectedErr := grpc.Errorf(codes.Code(code), msg)
	respStatus := &testpb.EchoStatus{
		Code:    proto.Int32(code),
		Message: proto.String(msg),
	}
	// Test UnaryCall.
	req := &testpb.SimpleRequest{
		ResponseStatus: respStatus,
	}
	if _, err := tc.UnaryCall(context.Background(), req); err == nil || err.Error() != expectedErr.Error() {
		grpclog.Fatalf("%v.UnaryCall(_, %v) = _, %v, want _, %v", tc, req, err, expectedErr)
	}
	// Test FullDuplexCall.
	stream, err := tc.FullDuplexCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.FullDuplexCall(_) = _, %v, want <nil>", tc, err)
	}
	streamReq := &testpb.StreamingOutputCallRequest{
		ResponseStatus: respStatus,
	}
	if err := stream.Send(streamReq); err != nil {
		grpclog.Fatalf("%v.Send(%v) = %v, want <nil>", stream, streamReq, err)
	}
	if err := stream.CloseSend(); err != nil {
		grpclog.Fatalf("%v.CloseSend() = %v, want <nil>", stream, err)
	}
	if _, err = stream.Recv(); err.Error() != expectedErr.Error() {
		grpclog.Fatalf("%v.Recv() returned error %v, want %v", stream, err, expectedErr)
	}
}

func (s *testServer) FullDuplexCall(stream testpb.TestService_FullDuplexCallServer) error {
	for {
		in, err := stream.Recv()
		if err == io.EOF {
			// read done.
			return nil
		}
		if err != nil {
			return err
		}
		status := in.GetResponseStatus()
		if status != nil && *status.Code != 0 {
			return grpc.Errorf(codes.Code(*status.Code), *status.Message)
		}
		cs := in.GetResponseParameters()
		for _, c := range cs {
			if us := c.GetIntervalUs(); us > 0 {
				time.Sleep(time.Duration(us) * time.Microsecond)
			}
			pl, err := serverNewPayload(in.GetResponseType(), c.GetSize())
			if err != nil {
				return err
			}
			if err := stream.Send(&testpb.StreamingOutputCallResponse{
				Payload: pl,
			}); err != nil {
				return err
			}
		}
	}
}

func newHPACKDecoder() *hpackDecoder {
	d := &hpackDecoder{}
	d.h = hpack.NewDecoder(http2InitHeaderTableSize, func(f hpack.HeaderField) {
		switch f.Name {
		case "content-type":
			// TODO(zhaoq): Tentatively disable the check until a bug is fixed.
			/*
				if !strings.Contains(f.Value, "application/grpc") {
					d.err = StreamErrorf(codes.FailedPrecondition, "transport: received the unexpected header")
					return
				}
			*/
		case "grpc-status":
			code, err := strconv.Atoi(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
				return
			}
			d.state.statusCode = codes.Code(code)
		case "grpc-message":
			d.state.statusDesc = f.Value
		case "grpc-timeout":
			d.state.timeoutSet = true
			var err error
			d.state.timeout, err = timeoutDecode(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
				return
			}
		case ":path":
			d.state.method = f.Value
		default:
			if !isReservedHeader(f.Name) {
				if f.Name == "user-agent" {
					i := strings.LastIndex(f.Value, " ")
					if i == -1 {
						// There is no application user agent string being set.
						return
					}
					// Extract the application user agent string.
					f.Value = f.Value[:i]
				}
				if d.state.mdata == nil {
					d.state.mdata = make(map[string][]string)
				}
				k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
				if err != nil {
					grpclog.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
					return
				}
				d.state.mdata[k] = append(d.state.mdata[k], v)
			}
		}
	})
	return d
}

func newHPACKDecoder(mdata map[string][]string) *hpackDecoder {
	d := &hpackDecoder{}
	for k, v := range mdata {
		d.mdata = make(map[string][]string)
		d.mdata[k] = v
	}
	d.h = hpack.NewDecoder(http2InitHeaderTableSize, func(f hpack.HeaderField) {
		switch f.Name {
		case "grpc-status":
			code, err := strconv.Atoi(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
				return
			}
			d.state.statusCode = codes.Code(code)
		case "grpc-message":
			d.state.statusDesc = f.Value
		case "grpc-timeout":
			d.state.timeoutSet = true
			var err error
			d.state.timeout, err = timeoutDecode(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
				return
			}
		case ":path":
			d.state.method = f.Value
		default:
			if !isReservedHeader(f.Name) {
				if f.Name == "user-agent" {
					i := strings.LastIndex(f.Value, " ")
					if i == -1 {
						// There is no application user agent string being set.
						return
					}
					// Extract the application user agent string.
					f.Value = f.Value[:i]
				}
				if d.state.mdata == nil {
					d.state.mdata = make(map[string][]string)
					for k, v := range d.mdata {
						d.state.mdata[k] = v
					}
				}
				k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
				if err != nil {
					grpclog.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
					return
				}
				d.state.mdata[k] = append(d.state.mdata[k], v)
			}
		}
	})
	return d
}

// ApplyBulk applies multiple Mutations.
// Each mutation is individually applied atomically,
// but the set of mutations may be applied in any order.
//
// Two types of failures may occur. If the entire process
// fails, (nil, err) will be returned. If specific mutations
// fail to apply, ([]err, nil) will be returned, and the errors
// will correspond to the relevant rowKeys/muts arguments.
//
// Depending on how the mutations are batched at the server one mutation may fail due to a problem
// with another mutation. In this case the same error will be reported for both mutations.
//
// Conditional mutations cannot be applied in bulk and providing one will result in an error.
func (t *Table) ApplyBulk(ctx context.Context, rowKeys []string, muts []*Mutation, opts ...ApplyOption) ([]error, error) {
	ctx = metadata.NewContext(ctx, t.md)
	if len(rowKeys) != len(muts) {
		return nil, fmt.Errorf("mismatched rowKeys and mutation array lengths: %d, %d", len(rowKeys), len(muts))
	}

	after := func(res proto.Message) {
		for _, o := range opts {
			o.after(res)
		}
	}

	req := &btspb.MutateRowsRequest{
		TableName: t.c.fullTableName(t.table),
		Entries:   make([]*btspb.MutateRowsRequest_Entry, len(rowKeys)),
	}
	for i, key := range rowKeys {
		mut := muts[i]
		if mut.cond != nil {
			return nil, fmt.Errorf("conditional mutations cannot be applied in bulk")
		}
		req.Entries[i] = &btspb.MutateRowsRequest_Entry{RowKey: []byte(key), Mutations: mut.ops}
	}
	stream, err := t.c.client.MutateRows(ctx, req)
	if err != nil {
		return nil, err
	}

	var errors []error // kept as nil if everything is OK
	for {
		res, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}

		for i, entry := range res.Entries {
			status := entry.Status
			if status.Code == int32(codes.OK) {
				continue
			}
			if errors == nil {
				errors = make([]error, len(rowKeys))
			}
			errors[i] = grpc.Errorf(codes.Code(status.Code), status.Message)
		}
		after(res)
	}

	return errors, nil
}

func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
	status := in.GetResponseStatus()
	if status != nil && *status.Code != 0 {
		return nil, grpc.Errorf(codes.Code(*status.Code), *status.Message)
	}
	pl, err := serverNewPayload(in.GetResponseType(), in.GetResponseSize())
	if err != nil {
		return nil, err
	}
	return &testpb.SimpleResponse{
		Payload: pl,
	}, nil
}

func (d *decodeState) processHeaderField(f hpack.HeaderField) {
	switch f.Name {
	case "content-type":
		if !strings.Contains(f.Value, "application/grpc") {
			d.setErr(StreamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value))
			return
		}
	case "grpc-encoding":
		d.encoding = f.Value
	case "grpc-status":
		code, err := strconv.Atoi(f.Value)
		if err != nil {
			d.setErr(StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err))
			return
		}
		d.statusCode = codes.Code(code)
	case "grpc-message":
		d.statusDesc = f.Value
	case "grpc-timeout":
		d.timeoutSet = true
		var err error
		d.timeout, err = timeoutDecode(f.Value)
		if err != nil {
			d.setErr(StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err))
			return
		}
	case ":path":
		d.method = f.Value
	default:
		if !isReservedHeader(f.Name) {
			if f.Name == "user-agent" {
				i := strings.LastIndex(f.Value, " ")
				if i == -1 {
					// There is no application user agent string being set.
					return
				}
				// Extract the application user agent string.
				f.Value = f.Value[:i]
			}
			if d.mdata == nil {
				d.mdata = make(map[string][]string)
			}
			k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
			if err != nil {
				grpclog.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
				return
			}
			d.mdata[k] = append(d.mdata[k], v)
		}
	}
}

func (s *testServer) FullDuplexCall(stream testpb.TestService_FullDuplexCallServer) error {
	if md, ok := metadata.FromContext(stream.Context()); ok {
		if initialMetadata, ok := md[initialMetadataKey]; ok {
			header := metadata.Pairs(initialMetadataKey, initialMetadata[0])
			stream.SendHeader(header)
		}
		if trailingMetadata, ok := md[trailingMetadataKey]; ok {
			trailer := metadata.Pairs(trailingMetadataKey, trailingMetadata[0])
			stream.SetTrailer(trailer)
		}
	}
	for {
		in, err := stream.Recv()
		if err == io.EOF {
			// read done.
			return nil
		}
		if err != nil {
			return err
		}
		status := in.GetResponseStatus()
		if status != nil && *status.Code != 0 {
			return grpc.Errorf(codes.Code(*status.Code), *status.Message)
		}
		cs := in.GetResponseParameters()
		for _, c := range cs {
			if us := c.GetIntervalUs(); us > 0 {
				time.Sleep(time.Duration(us) * time.Microsecond)
			}
			pl, err := serverNewPayload(in.GetResponseType(), c.GetSize())
			if err != nil {
				return err
			}
			if err := stream.Send(&testpb.StreamingOutputCallResponse{
				Payload: pl,
			}); err != nil {
				return err
			}
		}
	}
}

func newHPACKDecoder() *hpackDecoder {
	d := &hpackDecoder{}
	d.h = hpack.NewDecoder(http2InitHeaderTableSize, func(f hpack.HeaderField) {
		switch f.Name {
		case "grpc-status":
			code, err := strconv.Atoi(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
				return
			}
			d.state.statusCode = codes.Code(code)
		case "grpc-message":
			d.state.statusDesc = f.Value
		case "grpc-timeout":
			d.state.timeoutSet = true
			var err error
			d.state.timeout, err = timeoutDecode(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
				return
			}
		case ":path":
			d.state.method = f.Value
		default:
			if !isReservedHeader(f.Name) {
				if d.state.mdata == nil {
					d.state.mdata = make(map[string]string)
				}
				k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
				if err != nil {
					log.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
					return
				}
				d.state.mdata[k] = v
			}
		}
	})
	return d
}

func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
	status := in.GetResponseStatus()
	if md, ok := metadata.FromContext(ctx); ok {
		if initialMetadata, ok := md[initialMetadataKey]; ok {
			header := metadata.Pairs(initialMetadataKey, initialMetadata[0])
			grpc.SendHeader(ctx, header)
		}
		if trailingMetadata, ok := md[trailingMetadataKey]; ok {
			trailer := metadata.Pairs(trailingMetadataKey, trailingMetadata[0])
			grpc.SetTrailer(ctx, trailer)
		}
	}
	if status != nil && *status.Code != 0 {
		return nil, grpc.Errorf(codes.Code(*status.Code), *status.Message)
	}
	pl, err := serverNewPayload(in.GetResponseType(), in.GetResponseSize())
	if err != nil {
		return nil, err
	}
	return &testpb.SimpleResponse{
		Payload: pl,
	}, nil
}

// GRPCCode returns the gRPC version of the error code
func (b BCode) GRPCCode() codes.Code {
	return codes.Code(b)
}